D-Net

package eu.dnetlib.datasource.publisher.clients;

import java.util.List;

import com.google.common.base.Function;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.google.common.primitives.Ints;

import com.mongodb.BasicDBObject;

import com.mongodb.MongoClient;

import com.mongodb.client.FindIterable;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.model.Filters;

import com.mongodb.client.model.Projections;

import eu.dnetlib.datasource.publisher.DatasourceException;

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

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

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

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

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

import org.springframework.cache.annotation.Cacheable;

/**

 * 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 Filters.and(dbo("parentDatasourceId", dsId), dbo("system:profileFamily", "aggregator"), dbo("system:isCompletedSuccessfully", "true"));

	}

	private synchronized MongoCollection<Document> getCollection() {

		if (collection == null) {

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

		}

		return collection;

	}

	@Cacheable("datasources-mongo-cache")

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

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

		final Bson query = getQuery(dsId);

		final FindIterable<Document> iterable = getCollection().find(query).projection(fields).limit(limit).sort(dbo("system:startHumanDate", -1));

		final Iterable<AggregationInfo> transform = Iterables.transform(iterable, new Function<Document, AggregationInfo>() {

			@Override

			public AggregationInfo apply(final Document d) {

				final AggregationInfo info = new AggregationInfo();

				info.setAggregationStage(AggregationStage.parse(d.getString("system:wfName")));

				if (AggregationStage.COLLECT.equals(info.getAggregationStage())) {

					info.setCollectionMode(CollectionMode.valueOf(d.getString("system:node:SELECT_MODE:selection")));

				}

				info.setNumberOfRecords(getNumberOfRecords(d));

				info.setDate(getDate(d));

				return info;

			}

			private String getDate(final Document d) {

				return DateFormatUtils.format(new DateUtils().parse(d.getString("system:startHumanDate")), 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;

				}

			}

		});

		return Lists.newArrayList(transform);

	}

	private static Bson getFields() {

		return Projections.fields(dbo("system:wfName", 1),

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

				dbo("mainlog:sinkSize", 1),

				dbo("mainlog:writeOps", 1),

				dbo("mainlog:total", 1),

				dbo("system:startHumanDate", 1),

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

	}

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

		return new BasicDBObject(key, value);

	}

}