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.collect.Lists;

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.*;

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

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

import org.apache.commons.lang.StringUtils;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.http.HttpStatus;

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

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

import org.springframework.data.domain.Pageable;

import org.springframework.data.domain.Slice;

import org.springframework.util.concurrent.ListenableFuture;

import org.springframework.util.concurrent.ListenableFutureCallback;

/**

 * Created by claudio on 20/10/2016.

 */

public class DatasourceDao {

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

	@Autowired

	private MongoLoggerClient mongoLoggerClient;

	@Autowired

	private DatasourceIndexClient datasourceIndexClient;

	@Autowired

	private DatasourceRepository dsRepository;

	@Autowired

	private SearchInterfaceRepository searchInterfaceRepository;

	@Autowired

	private CountryTermRepository countryTermRepository;

	@Autowired

	private TypologyTermRepository typologyTermRepository;

	@Autowired

	private ProtocolTermRepository protocolTermRepository;

	@Autowired

	private CompatibilityTermRepository compatibilityTermRepository;

	@Autowired

	private ActivationTermRepository activationTermRepository;

	@Autowired

	private ApiRepository apiRepository;

	@Resource(name = "datasourceIsLookupClient")

	private ISLookupClient isLookupClient;

	private ListeningExecutorService service;

	private final static int WORKERS = 100;

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

	private int timeout = 10;

	public DatasourceDao() {

		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 List<DatasourceResponse> searchByName(final String name, final Pageable pageable) {

		final List<DatasourceResponse> datasourceResponse = Lists.newArrayList();

		final CountDownLatch outerLatch = new CountDownLatch(3);

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

		log.debug(String.format("search ds by name '%s'", name));

		final ListenableFuture<Slice<Datasource>> c =

				dsRepository.findByOfficialnameContainingOrEnglishnameContainingAllIgnoreCase(name, name, pageable);

		c.addCallback(getSearchCallback(outerLatch, errors, datasourceResponse));

		waitLatch(outerLatch, errors, timeout);

		return datasourceResponse;

	}

	public List<DatasourceResponse> searchByCountry(final String country, final Pageable pageable) {

		final List<DatasourceResponse> datasourceResponse = Lists.newArrayList();

		final CountDownLatch outerLatch = new CountDownLatch(3);

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

		log.debug(String.format("search ds by country '%s'", country));

		//dsRepository.findByOrganizationsCountryIgnoreCase(country, pageable).addCallback(getSearchCallback(outerLatch, errors, datasourceResponse));

		waitLatch(outerLatch, errors, timeout);

		return datasourceResponse;

	}

	public List<DatasourceResponse> searchByContactemail(final String email, final Pageable pageable) {

		final List<DatasourceResponse> datasourceResponse = Lists.newArrayList();

		final CountDownLatch outerLatch = new CountDownLatch(3);

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

		log.debug(String.format("search ds by email '%s'", email));

		dsRepository.findByContactemailContainingAllIgnoreCase(email, pageable).addCallback(getSearchCallback(outerLatch, errors, datasourceResponse));

		waitLatch(outerLatch, errors, timeout);

		return datasourceResponse;

	}

	private ListenableFutureCallback<Slice<Datasource>> getSearchCallback(final CountDownLatch outerLatch,

			final Queue<Throwable> errors,

			final List<DatasourceResponse> datasourceResponse) {

		return new ListenableFutureCallback<Slice<Datasource>>() {

			@Override

			public void onSuccess(final Slice<Datasource> datasources) {

				datasources.forEach(d -> {

					final DatasourceResponse response = new DatasourceResponse();

					response.setDatasource(d);

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

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

					datasourceResponse.add(response);

				});

				outerLatch.countDown();

			}

			@Override

			public void onFailure(final Throwable e) {

				errors.offer(e);

				outerLatch.countDown();

			}

		};

	}

	public ClientResponse getInfo(final String dsId) {

		final CountDownLatch outerLatch = new CountDownLatch(3);

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

		final DatasourceResponse datasourceResponse = new DatasourceResponse();

		getAggregationHistory(dsId, outerLatch, errors, datasourceResponse);

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

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

				errors.offer(e);

				outerLatch.countDown();

			}

		});

		getIndexDsInfo(dsId, outerLatch, errors, datasourceResponse);

		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 void setManaged(final String id, final boolean managed) {

		log.info(String.format("setting managed = '%s' for ds '%s'", managed, id));

		dsRepository.setManaged(id, managed);

	}

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

		switch (field) {

		case "__SEARCH__":

			return searchInterfaceRepository

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

	}

	public List<CountryTerm> browseCountries() {

		return countryTermRepository.findAll();

	}

	public List<TypologyTerm> browseTypologies() {

		return typologyTermRepository.findAll();

	}

	public List<ProtocolTerm> browseProtocols() {

		return protocolTermRepository.findAll();

	}

	public List<CompatibilityTerm> browseCompatibility() {

		return compatibilityTermRepository.findAll();

	}

	public List<ActivationTerm> browseActivation() {

		return activationTermRepository.findAll();

	}

	public List<Api> getApi(final String dsId) {

		return apiRepository.findByDatasource(dsId);

	}

	public void deleteApi(final String apiId) {

		apiRepository.delete(apiId);

		log.info(String.format("deleted api '%s'", apiId));

	}

	public void addApi(final Api api) {

		if (StringUtils.isBlank(api.getId())) {

			api.setId(ApiRepository.createId(api));

			log.info(String.format("missing api id, created '%s'"));

		}

		apiRepository.save(api);

	}

	public boolean exist(final Datasource d) throws ApiException {

		try {

			return dsRepository.findOneById(d.getId()).get() != null;

		} catch (Exception e) {

			log.error(e);

			throw new ApiException(HttpStatus.SC_INTERNAL_SERVER_ERROR, String.format("error retrieving datasource information '%s'", d.getId()), e);

		}

	}

	public Datasource save(final Datasource d) {

		log.info(String.format("saving datasource '%s'", d.getId()));

		final Datasource datasource = dsRepository.save(d);

		log.info(String.format("saved datasource '%s'", datasource.getId()));

		return datasource;

	}

	public void updateOfficialName(final String dsId, final String officialname) {

		dsRepository.setOfficialname(dsId, officialname);

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

	}

	public void updateEnglishName(final String dsId, final String englishname) {

		dsRepository.setEnglishname(dsId, englishname);

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

	}

	public void updateLatitude(final String dsId, final Double latitude) {

		dsRepository.setLatitude(dsId, latitude);

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

	}

	public void updateLongitude(final String dsId, final Double longitude) {

		dsRepository.setLongitude(dsId, longitude);

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

	}

	private void getIndexDsInfo(final String dsId,

			final CountDownLatch outerLatch,

			final Queue<Throwable> errors,

			final DatasourceResponse datasourceResponse) {

		Futures.addCallback(

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

				new FutureCallback<IndexDsInfo>() {

					public void onSuccess(final IndexDsInfo info) {

						final CountDownLatch innerLatch = new CountDownLatch(1);

						Futures.addCallback(

								service.submit(() -> datasourceIndexClient.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) {

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

						errors.offer(e);

						outerLatch.countDown();

					}

		});

	}

	private void getAggregationHistory(final String dsId,

			final CountDownLatch outerLatch,

			final Queue<Throwable> errors,

			final DatasourceResponse datasourceResponse) {

		Futures.addCallback(

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

				new FutureCallback<List<AggregationInfo>>() {

					public void onSuccess(List<AggregationInfo> info) {

						setAggregationHistory(datasourceResponse, info);

						outerLatch.countDown();

					}

					public void onFailure(Throwable e) {

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

						errors.offer(e);

						outerLatch.countDown();

					}

		});

	}

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

		}

	}

}