D-Net

package eu.dnetlib.datasource.publisher.clients;

import java.util.List;

import java.util.Queue;

import java.util.concurrent.*;

import javax.annotation.Resource;

import com.google.common.util.concurrent.*;

import eu.dnetlib.datasource.publisher.ApiException;

import eu.dnetlib.datasource.publisher.clients.utils.IndexDsInfo;

import eu.dnetlib.datasource.publisher.clients.utils.IndexRecordsInfo;

import eu.dnetlib.datasource.publisher.model.AggregationInfo;

import eu.dnetlib.datasource.publisher.model.AggregationStage;

import eu.dnetlib.datasource.publisher.model.DatasourceResponse;

import eu.dnetlib.datasource.publisher.model.db.Datasource;

import eu.dnetlib.datasource.publisher.model.db.SearchInterfacesEntry;

import eu.dnetlib.datasource.publisher.repository.DatasourceRepository;

import eu.dnetlib.datasource.publisher.repository.SearchApiRepository;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.data.domain.Pageable;

import org.springframework.util.concurrent.ListenableFutureCallback;

/**

 * Created by claudio on 20/10/2016.

 */

public class DatasourceInfoRetriever {

	private static final Log log = LogFactory.getLog(DatasourceInfoRetriever.class);

	@Autowired

	private MongoLoggerClient mongoLoggerClient;

	@Autowired

	private DatasourceIndexClient datasourceIndexClient;

	@Autowired

	private DatasourceRepository dsRepository;

	@Autowired

	private SearchApiRepository searchApiRepository;

	@Resource(name = "datasourceIsLookupClient")

	private ISLookupClient isLookupClient;

	private ListeningExecutorService service;

	private final static int WORKERS = 100;

	@Value("${datasource.publisher.timeout}")

	private int timeout = 10;

	public DatasourceInfoRetriever() {

		service = MoreExecutors.listeningDecorator(

				new ScheduledThreadPoolExecutor(WORKERS,

						new ThreadFactoryBuilder().setNameFormat("datasource-info-retriever-%d").build()));

	}

	public List<String> listIds(final Pageable pageable) throws ApiException {

		return dsRepository.findAll(pageable)

				.map(d -> d.getId())

				.getContent();

	}

	public ClientResponse getInfo(final String dsId) {

		final CountDownLatch outerLatch = new CountDownLatch(3);

		final Queue<Throwable> errors = new ConcurrentLinkedQueue<>();

		final DatasourceResponse datasourceResponse = new DatasourceResponse();

		Futures.addCallback(getAggregationHistory(dsId), new FutureCallback<List<AggregationInfo>>() {

			public void onSuccess(List<AggregationInfo> info) {

				setAggregationHistory(datasourceResponse, info);

				outerLatch.countDown();

			}

			public void onFailure(Throwable e) {

				errors.offer(e);

				outerLatch.countDown();

			}

		});

		dsRepository.findOneById(dsId).addCallback(new ListenableFutureCallback<Datasource>() {

			@Override

			public void onSuccess(final Datasource datasource) {

				datasourceResponse.setDatasource(datasource);

				outerLatch.countDown();

			}

			@Override

			public void onFailure(final Throwable e) {

				errors.offer(e);

				outerLatch.countDown();

			}

		});

		Futures.addCallback(calculateCurrentIndexDsInfo(), new FutureCallback<IndexDsInfo>() {

			public void onSuccess(final IndexDsInfo info) {

				final CountDownLatch innerLatch = new CountDownLatch(1);

				Futures.addCallback(getIndexInfo(dsId, info), new FutureCallback<IndexRecordsInfo>() {

					public void onSuccess(IndexRecordsInfo info) {

						datasourceResponse.setIndexRecords(info.getCount()).setLastIndexingDate(info.getDate());

						innerLatch.countDown();

					}

					public void onFailure(Throwable e) {

						errors.offer(e);

						innerLatch.countDown();

					}

				});

				waitLatch(innerLatch, errors, timeout);

				outerLatch.countDown();

			}

			public void onFailure(final Throwable e) {

				errors.offer(e);

				outerLatch.countDown();

			}

		});

		waitLatch(outerLatch, errors, timeout);

		/*

		if (!errors.isEmpty()) {

			datasourceResponse.getResponseHeader().setError(Joiner.on("\n").skipNulls().join(errors.stream().map(e -> e.getMessage()).collect(Collectors.toList())));

			log.error(Joiner.on("\n").skipNulls().join(errors.stream().map(e -> ExceptionUtils.getFullStackTrace(e)).collect(Collectors.toList())));

		}

		*/

		return new ClientResponse().datasourceInfo(datasourceResponse).errors(errors);

	}

	public List<SearchInterfacesEntry> searchInterface(final String field, final String value) {

		switch (field) {

		case "__SEARCH__":

			return searchApiRepository

					.findByRepoidContainingOrRepoNameContainingOrAlternativeNameContainingOrRepoPrefixContainingOrRepoOrganizationContainingAllIgnoreCase(

							value, value, value, value, value);

		case "country":

			break;

		case "type":

			break;

		case "protocol":

			break;

		case "compliance":

			break;

		case "active":

			break;

		default:

			throw new IllegalArgumentException("");

		}

		return null;

	}

	private ListenableFuture<IndexDsInfo> calculateCurrentIndexDsInfo() {

		return service.submit(() -> isLookupClient.calculateCurrentIndexDsInfo());

	}

	private ListenableFuture<IndexRecordsInfo> getIndexInfo(final String dsId, final IndexDsInfo info) {

		return service.submit(() -> datasourceIndexClient.getIndexInfo(dsId, info));

	}

	private ListenableFuture<List<AggregationInfo>> getAggregationHistory(final String dsId) {

		return service.submit(() -> mongoLoggerClient.getAggregationHistory(dsId));

	}

	private void setAggregationHistory(DatasourceResponse datasourceResponse, final List<AggregationInfo> info) {

		datasourceResponse.setAggregationHistory(info);

		if (!info.isEmpty()) {

			datasourceResponse

					.setLastCollection(info.stream().filter(a -> AggregationStage.COLLECT.equals(a.getAggregationStage())).findFirst().get())

					.setLastTransformation(info.stream().filter(a -> AggregationStage.TRANSFORM.equals(a.getAggregationStage())).findFirst().get());

		}

	}

	private void waitLatch(final CountDownLatch latch, final Queue<Throwable> errors, final int waitSeconds) {

		try {

			if (!latch.await(waitSeconds, TimeUnit.SECONDS)) {

				errors.offer(new TimeoutException("Waiting for requests to complete has timed out."));

			}

		} catch (final InterruptedException e) {

			errors.offer(e);

		}

	}

}