/*
 * Copyright 2019 EPAM Systems
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.epam.ta.reportportal.core.analyzer.auto.client.impl;

import com.epam.ta.reportportal.core.analyzer.auto.client.IndexerServiceClient;
import com.epam.ta.reportportal.core.analyzer.auto.client.RabbitMqManagementClient;
import com.epam.ta.reportportal.ws.model.analyzer.CleanIndexRq;
import com.epam.ta.reportportal.ws.model.analyzer.IndexLaunch;
import com.epam.ta.reportportal.ws.model.analyzer.IndexRs;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.stereotype.Service;

import java.util.AbstractMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

import static com.epam.ta.reportportal.core.analyzer.auto.client.impl.AnalyzerUtils.DOES_SUPPORT_INDEX;
import static com.epam.ta.reportportal.core.analyzer.auto.client.impl.AnalyzerUtils.EXCHANGE_PRIORITY;

/**
 * @author <a href="mailto:pavel_bortnik@epam.com">Pavel Bortnik</a>
 */
@Service
public class IndexerServiceClientImpl implements IndexerServiceClient {

	private static final String INDEX_ROUTE = "index";
	private static final String DELETE_ROUTE = "delete";
	private static final String CLEAN_ROUTE = "clean";

	private final RabbitMqManagementClient rabbitMqManagementClient;

	private final RabbitTemplate rabbitTemplate;

	public IndexerServiceClientImpl(RabbitMqManagementClient rabbitMqManagementClient,
			@Qualifier("analyzerRabbitTemplate") RabbitTemplate rabbitTemplate) {
		this.rabbitMqManagementClient = rabbitMqManagementClient;
		this.rabbitTemplate = rabbitTemplate;
	}

	@Override
	public Long index(List<IndexLaunch> rq) {
		return rabbitMqManagementClient.getAnalyzerExchangesInfo()
				.stream()
				.filter(DOES_SUPPORT_INDEX)
				.map(exchange -> rabbitTemplate.convertSendAndReceiveAsType(exchange.getName(),
						INDEX_ROUTE,
						rq,
						new ParameterizedTypeReference<IndexRs>() {
						}
				))
				.mapToLong(it -> {
					if (Objects.nonNull(it) && it.getItems() != null) {
						return it.getItems().size();
					}
					return 0;
				})
				.sum();
	}

	@Override
	public Long cleanIndex(Long index, List<Long> ids) {
		Map<Integer, Long> priorityToCleanedLogsCountMapping = rabbitMqManagementClient.getAnalyzerExchangesInfo()
				.stream()
				.collect(Collectors.toMap(EXCHANGE_PRIORITY::applyAsInt,
						exchange -> rabbitTemplate.convertSendAndReceiveAsType(exchange.getName(),
								CLEAN_ROUTE,
								new CleanIndexRq(index, ids),
								new ParameterizedTypeReference<Long>() {
								}
						)
				));
		return priorityToCleanedLogsCountMapping.entrySet()
				.stream()
				.min(Map.Entry.comparingByKey())
				.orElseGet(() -> new AbstractMap.SimpleEntry<Integer, Long>(0, 0L))
				.getValue();
	}

	@Override
	public void deleteIndex(Long index) {
		rabbitMqManagementClient.getAnalyzerExchangesInfo()
				.forEach(exchange -> rabbitTemplate.convertAndSend(exchange.getName(), DELETE_ROUTE, index));
	}
}

@@ -1002,16 +1002,73 @@
 ndexRs;%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A
 import o
@@ -1795,16 +1795,107 @@
 ient %7B%0A%0A
+%09private static final Logger LOGGER = LoggerFactory.getLogger(IndexerServiceClient.class);%0A
 %09private
@@ -2043,16 +2043,77 @@
 %22clean%22;
+%0A%09private static final Integer DELETE_INDEX_SUCCESS_CODE = 1;
 %0A%0A%09priva
@@ -3755,15 +3755,25 @@
 %09%09%09.
-forEach
+stream()%0A%09%09%09%09.map
 (exc
@@ -3807,15 +3807,28 @@
 vert
-And
 Send
+AndReceiveAsType
 (exc
@@ -3843,17 +3843,23 @@
 tName(),
- 
+%0A%09%09%09%09%09%09
 DELETE_R
@@ -3863,23 +3863,308 @@
 E_ROUTE,
- index)
+%0A%09%09%09%09%09%09index,%0A%09%09%09%09%09%09new ParameterizedTypeReference%3CInteger%3E() %7B%0A%09%09%09%09%09%09%7D%0A%09%09%09%09))%0A%09%09%09%09.forEach(it -%3E %7B%0A%09%09%09%09%09if (DELETE_INDEX_SUCCESS_CODE.equals(it)) %7B%0A%09%09%09%09%09%09LOGGER.info(%22Successfully deleted index '%7B%7D'%22, index);%0A%09%09%09%09%09%7D else %7B%0A%09%09%09%09%09%09LOGGER.error(%22Error deleting index '%7B%7D'%22, index);%0A%09%09%09%09%09%7D%0A%09%09%09%09%7D
 );%0A%09%7D%0A%7D%0A

delete response handling from analyzer