D-Net

package eu.dnetlib.openaire.exporter.datasource.clients;

import java.util.List;

import java.util.function.Function;

import java.util.stream.Collectors;

import com.google.common.collect.Streams;

import com.google.common.primitives.Ints;

import com.mongodb.BasicDBObject;

import com.mongodb.MongoClient;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoCursor;

import com.mongodb.client.model.Projections;

import eu.dnetlib.OpenaireExporterConfig;

import eu.dnetlib.OpenaireExporterConfig.Datasource;

import eu.dnetlib.Utils;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.openaire.exporter.datasource.ApiException;

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

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

import org.apache.commons.lang.StringUtils;

import org.apache.commons.lang.time.DateFormatUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.Document;

import org.bson.conversions.Bson;

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

import org.springframework.cache.annotation.CacheEvict;

import org.springframework.cache.annotation.Cacheable;

import org.springframework.stereotype.Component;

import static com.mongodb.client.model.Filters.*;

/**

 * Created by claudio on 20/10/2016.

 */

@Component

public class MongoLoggerClient {

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

	@Autowired

	private MongoClient datasourcePublisherMongoClient;

	@Autowired

	private OpenaireExporterConfig config;

	private static Bson fields = getFields();

	private static MongoCollection<Document> collection = null;

	private Bson getQuery(final String dsId) {

		return and(

				eq("parentDatasourceId", dsId),

				eq("system:profileFamily", "aggregator"),

				eq("system:isCompletedSuccessfully", "true"),

				regex("system:wfName", "(collect|transform)", "i"));

	}

	private synchronized MongoCollection<Document> getCollection() {

		if (collection == null) {

			log.info("inizializing mongodb collection ...");

			final Datasource conf = config.getDatasource();

			collection = datasourcePublisherMongoClient.getDatabase(conf.getMongoDbName()).getCollection(conf.getMongoCollectionName());

		}

		return collection;

	}

	@Cacheable("datasources-mongo-cache")

	public List<AggregationInfo> getAggregationHistory(final String dsId) throws ApiException {

		log.warn(String.format("getAggregationHistory(dsId = %s): not using cache", dsId));

		final Datasource conf = config.getDatasource();

		final Bson query = getQuery(dsId);

		final MongoCursor<Document> cursor = getCollection().find(query)

				.projection(fields)

				.limit(conf.getMongoQueryLimit())

				.sort(dbo("system:startHumanDate", -1))

				.iterator();

		return Streams.stream(cursor)

				.map(getMapper())

				.filter(ai -> ai.getNumberOfRecords() >= 0 && StringUtils.isNotBlank(ai.getDate()))

				.collect(Collectors.toList());

	}

	@CacheEvict(cacheNames = "datasources-mongo-cache", allEntries = true)

	public void dropCache() {

		log.info("dropped dsManager aggregation history cache");

	}

	private Function<Document, AggregationInfo> getMapper() {

		return new Function<Document, AggregationInfo>() {

			@Override

			public AggregationInfo apply(final Document d) {

				AggregationInfo info = null;

				final AggregationStage stage = AggregationStage.parse(d.getString("system:wfName"));

				switch (stage) {

				case COLLECT:

					CollectionInfo cInfo = new CollectionInfo();

					cInfo.setAggregationStage(stage);

					final String collectionMode = d.getString("system:node:SELECT_MODE:selection");

					cInfo.setCollectionMode(StringUtils.isNotBlank(collectionMode) ? CollectionMode.valueOf(collectionMode) : null);

					cInfo.setNumberOfRecords(getNumberOfRecords(d));

					cInfo.setDate(getDate(d));

					info = cInfo;

					break;

				case TRANSFORM:

					TransformationInfo tInfo = new TransformationInfo();

					tInfo.setAggregationStage(stage);

					tInfo.setNumberOfRecords(getNumberOfRecords(d));

					tInfo.setDate(getDate(d));

					info = tInfo;

					break;

				}

				return info;

			}

			private String getDate(final Document d) {

				final String dateString = d.getString("system:startHumanDate");

				if (StringUtils.isBlank(dateString)) {

					return "";

				}

				return DateFormatUtils.format(new DateUtils().parse(dateString), DatasourceFunctions.DATE_FORMAT);

			}

			private Integer getNumberOfRecords(final Document d) {

				final String sinkSize = d.getString("mainlog:sinkSize");

				final String total = d.getString("mainlog:total");

				if(StringUtils.isNotBlank(sinkSize)) {

					return Ints.tryParse(sinkSize);

				} else if(StringUtils.isNotBlank(total)) {

					return Ints.tryParse(total);

				} else {

					return -1;

				}

			}

		};

	}

	private static Bson getFields() {

		return Projections.fields(

				eq("system:wfName", 1),

				eq("system:node:SELECT_MODE:selection", 1),

				eq("mainlog:sinkSize", 1),

				eq("mainlog:writeOps", 1),

				eq("mainlog:total", 1),

				eq("system:startHumanDate", 1),

				eq("system:profileName", 1));

	}

	private static BasicDBObject dbo(final String key, final Object value) {

		return new BasicDBObject(key, value);

	}

}