D-Net

package eu.dnetlib.datasource.publisher.clients;

import java.util.List;

import java.util.function.Function;

import java.util.stream.Collectors;

import com.google.common.primitives.Ints;

import com.mongodb.BasicDBObject;

import com.mongodb.MongoClient;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.model.Projections;

import eu.dnetlib.datasource.publisher.ApiException;

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

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

import eu.dnetlib.datasource.publisher.utils.Utils;

import eu.dnetlib.miscutils.datetime.DateUtils;

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.beans.factory.annotation.Value;

import org.springframework.cache.annotation.Cacheable;

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

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

/**

 * Created by claudio on 20/10/2016.

 */

public class MongoLoggerClient {

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

	@Autowired

	private MongoClient datasourcePublisherMongoClient;

	@Value("${datasource.publisher.mongodb.collection}")

	private String collectionName;

	@Value("${datasource.publisher.mongodb.db}")

	private String dbName;

	@Value("${datasource.publisher.mongodb.query.limit}")

	private int limit;

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

	}

	private synchronized MongoCollection<Document> getCollection() {

		if (collection == null) {

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

			collection = datasourcePublisherMongoClient.getDatabase(dbName).getCollection(collectionName);

		}

		return collection;

	}

	@Cacheable("datasources-mongo-cache")

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

		log.warn("getAggregationHistory(): not using cache");

		final Bson query = getQuery(dsId);

		return Utils.stream(getCollection().find(query).projection(fields).limit(limit).sort(dbo("system:startHumanDate", -1)).iterator())

				.map(getMapper())

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

				.collect(Collectors.toList());

	}

	private Function<Document, AggregationInfo> getMapper() {

		return new java.util.function.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);

	}

}