D-Net

package eu.dnetlib.common.logging.dao;

import java.util.*;

import java.util.Map.Entry;

import java.util.function.Consumer;

import javax.annotation.Resource;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import com.mongodb.BasicDBObject;

import com.mongodb.BasicDBObjectBuilder;

import com.mongodb.DBObject;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoCursor;

import com.mongodb.client.MongoDatabase;

import com.mongodb.client.model.Filters;

import com.mongodb.client.model.IndexOptions;

import eu.dnetlib.common.logging.LogMessage;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.Document;

import org.bson.conversions.Bson;

public class DnetLoggerMongoDao implements DnetLoggerDao {

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

	@Resource(name = "loggingMongoDB")

	private MongoDatabase db;

	private IndexOptions indexOptions = new IndexOptions().background(true);

	@Override

	public void init(final String collection) {

		final Set<String> collections = Sets.newHashSet();

		db.listCollectionNames().forEach((Consumer<? super String>) c -> collections.add(c));

		if (!Lists.newArrayList(collections).contains(collection)) {

			log.info(String.format("creating collection %s", collection));

			db.createCollection(collection);

		}

	}

	@Override

	public void configureIndex(final String collection, final Map<String, IndexConf> conf) {

		final MongoCollection<Document> coll = db.getCollection(collection);

		coll.dropIndexes();

		for (String key : conf.keySet()) {

			coll.createIndex(new BasicDBObject(key, 1), indexOptions);

		}

	}

	@Override

	public void writeLog(final String collection, final Map<String, Object> map) {

		final MongoCollection coll = db.getCollection(collection);

		final Document doc = new Document(replaceKeyNames(map));

		coll.insertOne(doc);

	}

	private Map<String, Object> replaceKeyNames(final Map<String, Object> inputMap) {

		final Map<String, Object> outMap = Maps.newHashMap();

		if (inputMap != null) {

			for (Entry<String, Object> e : inputMap.entrySet()) {

				final String k = e.getKey();

				if (!StringUtils.isBlank(k)) {

					Object v = e.getValue();

					outMap.put(k.replaceAll("\\.", "_"), v != null ? v : "null");

				}

			}

		}

		return outMap;

	}

	@Override

	public Map<String, String> findOne(final String collection, final String key, final String value) {

		final MongoCollection coll = db.getCollection(collection);

		final DBObject obj = (DBObject) coll.find(Filters.eq(key, value), DBObject.class).first();

		//final DBObject obj = coll.find(getBasicQuery(key, value)).as(BasicDBObject.class);

		return dbObject2Map(obj);

	}

	@Override

	public Iterator<Map<String, String>> obtainLogIterator(final String collection) {

		final MongoCollection coll = db.getCollection(collection);

		return iter(coll.find().iterator());

	}

	@Override

	public Iterator<Map<String, String>> find(final String collection, final String key, final String value) {

		final MongoCollection coll = db.getCollection(collection);

		return iter(coll.find(getBasicQuery(key, value)).iterator());

	}

	@Override

	public Iterator<Map<String, String>> find(final String collection, final Map<String, Object> criteria) {

		final MongoCollection coll = db.getCollection(collection);

		return iter(coll.find(getCustomQuery(criteria)).iterator());

	}

	@Override

	public Iterator<Map<String, String>> findByDateRange(final String collection, final Date startDate, final Date endDate) {

		final MongoCollection coll = db.getCollection(collection);

		return iter(coll.find(getDateQuery(startDate, endDate)).iterator());

	}

	@Override

	public Iterator<Map<String, String>> findByDateRange(final String collection, final Date startDate, final Date endDate, final String key, final String value) {

		final MongoCollection coll = db.getCollection(collection);

		final Bson query = Filters.and(getBasicQuery(key, value), getDateQuery(startDate, endDate));

		return iter(coll.find(query).iterator());

	}

	@Override

	public Iterator<Map<String, String>> findByDateRange(final String collection, final Date startDate, final Date endDate, final Map<String, Object> criteria) {

		final MongoCollection coll = db.getCollection(collection);

		final Bson query = Filters.and(getCustomQuery(criteria), getDateQuery(startDate, endDate));

		return iter(coll.find(query).iterator());

	}

	@Override

	public Iterator<Map<String, String>> find(final String collection, final String mongoquery, final List<String> fields) {

		final MongoCollection coll = db.getCollection(collection);

		final Bson query = BasicDBObject.parse(mongoquery);

		return iter(coll.find(query).iterator());

	}

	private Bson getBasicQuery(final String key, final String value) {

		return new BasicDBObject(key, value);

	}

	private Bson getDateQuery(final Date startDate, final Date endDate) {

		return Filters.and(

				Filters.gte(LogMessage.LOG_DATE_FIELD, startDate.getTime()),

				Filters.lt(LogMessage.LOG_DATE_FIELD, endDate.getTime()));

	}

	private Bson getCustomQuery(final Map<String, Object> criteria) {

		final BasicDBObjectBuilder b = new BasicDBObjectBuilder();

		criteria.forEach((k, v) -> b.add(k, v));

		return (BasicDBObject) b.get();

	}

	private Bson dateRangeQuery(final Date startDate, final Date endDate) {

		Bson dateFilter = Filters.and(Filters.gte(LogMessage.LOG_DATE_FIELD, startDate.getTime()), Filters.lt(LogMessage.LOG_DATE_FIELD, endDate.getTime()));

		log.debug("Date filter created: " + dateFilter);

		//return new BasicDBObject(LogMessage.LOG_DATE_FIELD, BasicDBObjectBuilder.start("$gte", startDate.getTime()).append("$lt", endDate.getTime()).get());

		return dateFilter;

	}

	private Iterator<Map<String, String>> iter(final MongoCursor<BasicDBObject> cursor) {

		return new Iterator<Map<String, String>>() {

			@Override

			public boolean hasNext() {

				return cursor.hasNext();

			}

			@Override

			public Map<String, String> next() {

				return dbObject2Map(cursor.next());

			}

			@Override

			public void remove() {

				throw new RuntimeException("NOT IMPLEMENTED");

			}

		};

	}

	private Map<String, String> dbObject2Map(final DBObject obj) {

		final Map<String, String> res = Maps.newHashMap();

		if (obj != null) {

			for (String k : obj.keySet()) {

				res.put(k, "" + obj.get(k));

			}

		}

		return res;

	}

}