D-Net

package eu.dnetlib.openaire.exporter.datasource;

import java.util.List;

import java.util.Queue;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ScheduledThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.TimeoutException;

import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import com.google.common.collect.Lists;

import com.google.common.collect.Queues;

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

import eu.dnetlib.OpenaireExporterConfig;

import eu.dnetlib.enabling.datasources.common.AggregationInfo;

import eu.dnetlib.enabling.datasources.common.AggregationStage;

import eu.dnetlib.enabling.datasources.common.DatasourceManagerException;

import eu.dnetlib.openaire.exporter.datasource.clients.*;

import eu.dnetlib.openaire.exporter.datasource.clients.utils.IndexDsInfo;

import eu.dnetlib.openaire.exporter.datasource.clients.utils.IndexRecordsInfo;

import eu.dnetlib.openaire.exporter.model.ApiDetailsResponse;

import eu.dnetlib.openaire.exporter.model.ConversionUtils;

import eu.dnetlib.openaire.exporter.model.DatasourceResponse;

import eu.dnetlib.openaire.exporter.model.datasource.*;

import eu.dnetlib.openaire.exporter.model.datasource.db.ApiDbEntry;

import eu.dnetlib.openaire.exporter.model.datasource.db.DatasourceDbEntry;

import eu.dnetlib.openaire.exporter.vocabularies.Country;

import org.apache.commons.lang3.exception.ExceptionUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.data.domain.Page;

import org.springframework.stereotype.Component;

import static eu.dnetlib.openaire.exporter.datasource.clients.ResponseUtils.apiResponse;

import static eu.dnetlib.openaire.exporter.datasource.clients.ResponseUtils.datasourceResponse;

@Component

public class DatasourceManagerCore {

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

	@Autowired

	private MongoLoggerClient mongoLoggerClient;

	@Autowired

	private ISLookupClient isLookupClient;

	@Autowired

	private ObjectStoreClient objectStoreClient;

	@Autowired

	private DatasourceIndexClient datasourceIndexClient;

	@Autowired

	private VocabularyClient vocabularyClient;

	@Autowired

	private DatasourceDao dsDao;

	@Autowired

	private OpenaireExporterConfig config;

	private ListeningExecutorService executor;

	@PostConstruct

	public void init() {

		executor = MoreExecutors.listeningDecorator(

				new ScheduledThreadPoolExecutor(config.getRequestWorkers(),

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

	}

	public List<Country> listCountries() throws DatasourceManagerException {

		return dsDao.listCountries();

	}

	public DatasourceResponse search(RequestSort requestSortBy, RequestSortOrder order, RequestFilter requestFilter, int page, int size)

			throws DatasourceManagerException {

		final List<DatasourceInfo> datasourceInfo = Lists.newArrayList();

		final Queue<Throwable> errors = Queues.newLinkedBlockingQueue();

		final CountDownLatch outerLatch = new CountDownLatch(2);

		final Page<DatasourceDbEntry> dsPage = dsDao.search(requestSortBy, order, requestFilter, page, size);

		if (dsPage.getTotalElements() > 0 && dsPage.getNumberOfElements() > 0) {

			dsPage.forEach(d -> datasourceInfo.add(enrichDatasourceInfo(ConversionUtils.asDetails(d), outerLatch, errors)));

			waitLatch(outerLatch, errors, config.getRequestTimeout());

		}

		return datasourceResponse(datasourceInfo, errors, dsPage.getTotalElements());

	}

	public List<String> findBaseURLs(final RequestFilter requestFilter, final int page, final int size) throws DatasourceManagerException {

		return dsDao.findBaseURLs(requestFilter, page, size);

	}

	public ApiDetailsResponse getApis(final String dsId) throws DatasourceManagerException {

		final List<ApiDbEntry> apis = dsDao.getApis(dsId);

		final List<ApiDetails> api = apis.stream()

				.map(ConversionUtils::asDetails)

				.collect(Collectors.toList());

		return apiResponse(api, api.size());

	}

	public void setManaged(final String dsId, final boolean managed) throws DatasourceManagerException {

		log.info(String.format("updated api '%s' managed with '%s'", dsId, managed));

		dsDao.setManaged(dsId, managed);

	}

	public boolean isManaged(final String dsId) throws DatasourceManagerException {

		return dsDao.isManaged(dsId);

	}

	public boolean exist(final DatasourceDetails d) throws DatasourceManagerException {

		return ((DatasourceDaoImpl) dsDao).existDs(d.getId());

	}

	public void save(final DatasourceDetails d) throws DatasourceManagerException {

		dsDao.saveDs(ConversionUtils.asDbEntry(d));

		//TODO add also to the IS

	}

	public void updateDatasourcename(final String dsId, final String officialname, final String englishname) throws DatasourceManagerException {

		log.info(String.format("updated datasource '%s' with officialname '%s' and englishname '%s'", dsId, officialname, englishname));

		dsDao.updateDatasourceName(dsId, officialname, englishname);

		//TODO update also to the IS

	}

	public void updateCoordinates(final String dsId, final Double latitude, final Double longitude) throws DatasourceManagerException {

		log.info(String.format("updated datasource '%s' with coordinates Lat:'%s', Lon:'%s'", dsId, latitude, longitude));

		dsDao.updateCoordinates(dsId, latitude, longitude);

		//TODO update also to the IS

	}

	public void updateApiBaseurl(final String apiId, final String baseUrl) throws DatasourceManagerException {

		log.info(String.format("updated api '%s' baseurl with '%s'", apiId, baseUrl));

		dsDao.updateBaseUrl(apiId, baseUrl);

		//TODO update also to the IS

	}

	public void updateTimezone(final String dsId, final String timezone) throws DatasourceManagerException {

		log.info(String.format("updated datasource '%s' timezone with '%s'", dsId, timezone));

		dsDao.updateTimezone(dsId, timezone);

		//TODO update also to the IS

	}

	public void updateDsTypology(final String dsId, final String typology) throws DatasourceManagerException {

		log.info(String.format("updated datasource '%s' typology with '%s'", dsId, typology));

		dsDao.updateTypology(dsId, typology);

		//TODO update also to the IS

	}

	public void updateApiCompatibility(final String apiId, final String compliance, final boolean override) throws DatasourceManagerException {

		log.info(String.format("updated api '%s' compliance with '%s'", apiId, compliance));

		dsDao.updateCompliance(null, apiId, compliance, override);

		//TODO update also to the IS

	}

	public void deleteDs(final String dsId) throws DatasourceManagerException {

		log.info(String.format("deleted datasource '%s'", dsId));

		dsDao.deleteDs(dsId);

	}

	public void deleteApi(final String apiId) throws DatasourceManagerException {

		dsDao.deleteApi(null, apiId);

	}

	public void dropCaches() {

		mongoLoggerClient.dropCache();

		isLookupClient.dropCache();

		vocabularyClient.dropCache();

	}

	// HELPERS //////////////

	private DatasourceInfo enrichDatasourceInfo(final DatasourceDetails d, final CountDownLatch outerLatch, final Queue<Throwable> errors) {

		final DatasourceInfo dsInfo = new DatasourceInfo().setDatasource(d);

		getAggregationHistory(d.getId(), outerLatch, errors, dsInfo);

		getIndexDsInfo(d.getId(), outerLatch, errors, dsInfo);

		return dsInfo;

	}

	private void getAggregationHistory(final String dsId,

			final CountDownLatch outerLatch,

			final Queue<Throwable> errors,

			final DatasourceInfo datasourceInfo) {

		Futures.addCallback(

				executor.submit(() -> mongoLoggerClient.getAggregationHistory(dsId)),

				new FutureCallback<List<AggregationInfo>>() {

					public void onSuccess(List<AggregationInfo> info) {

						setAggregationHistory(datasourceInfo, info);

						outerLatch.countDown();

					}

					public void onFailure(Throwable e) {

						log.error(ExceptionUtils.getStackTrace(e));

						errors.offer(e);

						outerLatch.countDown();

					}

				}, executor);

	}

	private void setAggregationHistory(final DatasourceInfo datasourceInfo, final List<AggregationInfo> info) {

		datasourceInfo.setAggregationHistory(info);

		if (!info.isEmpty()) {

			datasourceInfo

					.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 getIndexDsInfo(final String dsId,

			final CountDownLatch outerLatch,

			final Queue<Throwable> errors,

			final DatasourceInfo datasourceInfo) {

		Futures.addCallback(

				executor.submit(() -> isLookupClient.calculateCurrentIndexDsInfo()),

				new FutureCallback<IndexDsInfo>() {

					public void onSuccess(final IndexDsInfo info) {

						final CountDownLatch innerLatch = new CountDownLatch(2);

						Futures.addCallback(

								executor.submit(() -> datasourceIndexClient.getIndexInfo(dsId, info, errors)),

								new FutureCallback<IndexRecordsInfo>() {

									public void onSuccess(IndexRecordsInfo info) {

										datasourceInfo

												.setIndexRecords(info.getTotal())

												.setFundedContent(info.getFunded())

												.setLastIndexingDate(info.getDate());

										innerLatch.countDown();

									}

									public void onFailure(Throwable e) {

										errors.offer(e);

										innerLatch.countDown();

									}

								}, executor);

						Futures.addCallback(

								executor.submit(() ->

										objectStoreClient.getObjectStoreSize(

												isLookupClient.getObjectStoreId(dsId, errors),

												errors)),

								new FutureCallback<Long>() {

									@Override

									public void onSuccess(final Long objectStoreSize) {

										datasourceInfo.setFulltexts(objectStoreSize);

										innerLatch.countDown();

									}

									@Override

									public void onFailure(final Throwable e) {

										errors.offer(e);

										innerLatch.countDown();

									}

								}, executor);

						waitLatch(innerLatch, errors, config.getRequestTimeout());

						outerLatch.countDown();

					}

					public void onFailure(final Throwable e) {

						//log.error(ExceptionUtils.getStackTrace(e));

						errors.offer(e);

						outerLatch.countDown();

					}

				}, executor);

	}

	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);

		}

	}

}