D-Net

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet45-parent</artifactId>

		<version>1.0.0</version>

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

	<artifactId>cnr-mongo-mdstore</artifactId>

	<packaging>jar</packaging>

	<version>6.0.3</version>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet45/modules/cnr-mongo-mdstore/tags/cnr-mongo-mdstore-6.0.3</developerConnection>

	</scm>

	<dependencies>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>${junit.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.springframework</groupId>

			<artifactId>spring-test</artifactId>

			<version>${spring.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.mockito</groupId>

			<artifactId>mockito-core</artifactId>

			<version>${mockito.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-test-utils</artifactId>

			<version>[1.0.0,2.0.0)</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-modular-mdstore-service</artifactId>

			<version>[6.0.0,7.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-inspector</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>org.mongodb</groupId>

			<artifactId>mongo-java-driver</artifactId>

			<version>${mongodb.driver.version}</version>

		</dependency>

		<dependency>

			<groupId>com.google.code.gson</groupId>

			<artifactId>gson</artifactId>

			<version>2.2</version>

		</dependency>

		<dependency>

			<groupId>joda-time</groupId>

			<artifactId>joda-time</artifactId>

			<version>2.3</version>

		</dependency>

		<dependency>

			<groupId>com.ximpleware</groupId>

			<artifactId>vtd-xml</artifactId>

			<version>[2.12, 3.0.0)</version>

		</dependency>

	</dependencies>

</project>

package eu.dnetlib.data.mdstore.modular.mongodb;

import java.io.File;

import java.io.FileInputStream;

import java.io.IOException;

import java.util.Iterator;

import java.util.UUID;

import com.mongodb.DBObject;

import com.mongodb.client.MongoDatabase;

import eu.dnetlib.data.mdstore.MDStoreServiceException;

import eu.dnetlib.data.mdstore.modular.RecordParserFactory;

import eu.dnetlib.data.mdstore.modular.connector.MDStoreDao;

import org.apache.commons.io.IOUtils;

import org.junit.Before;

import org.junit.Ignore;

import org.junit.Test;

import org.junit.runner.RunWith;

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

import org.springframework.core.io.ClassPathResource;

import org.springframework.test.context.ContextConfiguration;

import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@Ignore

@RunWith(SpringJUnit4ClassRunner.class)

@ContextConfiguration(classes = ConfigurationTestConfig.class)

public class FeedSpeedTest {

	private static final int N_RECORDS = 68271;

	@Autowired

	private MongoDatabase db;

	@Autowired

	private MDStoreDao dao;

	@Autowired

	private RecordParserFactory recordParserfactory;

	@Before

	public void setup() throws MDStoreServiceException {

		dao.createMDStore("speed_test", "testFormat", "testInterpretation", "testLayout");

	}

	@Test

	public void testSpeedFromFolder() throws IOException {

		Iterable<String> iterable = new Iterable<String>() {

			private int counter = 0;

			private double last = System.currentTimeMillis();

			@Override

			public Iterator<String> iterator() {

				return new Iterator<String>() {

					@Override

					public boolean hasNext() {

						return counter < N_RECORDS;

					}

					@Override

					public String next() {

						if (counter % 10000 == 0) {

							System.out.println("10K records processed in " + (System.currentTimeMillis() - last) / 1000 + " seconds");

							last = System.currentTimeMillis();

						}

						File f = new File(String.format("/var/lib/eagle/content/EDH/HD%06d.xml", counter++));

						if (f.exists()) {

							try {

								FileInputStream fileInputStream = new FileInputStream(f);

								String s = IOUtils.toString(fileInputStream);

								fileInputStream.close();

								return s;

							} catch (Exception e) {

								return null;

							}

						} else {

							try {

								FileInputStream fileInputStream = new FileInputStream(new File("/var/lib/eagle/content/EDH/HD000001.xml"));

								String s = IOUtils.toString(fileInputStream);

								fileInputStream.close();

								return s;

							} catch (Exception e) {

								return null;

							}

						}

					}

					@Override

					public void remove() {}

				};

			}

		};

		MongoMDStore mdStore =

				new MongoMDStore(UUID.randomUUID().toString(), db.getCollection("speed_test", DBObject.class), recordParserfactory.newInstance(), true, db);

		mdStore.feed(iterable, false);

	}

	//@Ignore

	@Test

	public void testFeedSpeedFromTemplate() throws MDStoreServiceException, IOException {

		MongoMDStore mdStore =

				new MongoMDStore(UUID.randomUUID().toString(), db.getCollection("speed_test", DBObject.class), recordParserfactory.newInstance(), false, db);

		mdStore.feed(new Iterable<String>() {

			private int counter = 0;

			private double last = System.currentTimeMillis();

			private String templateRecord = IOUtils.toString(new ClassPathResource("/eu/dnetlib/data/mdstore/modular/mongodb/templateRecord.xml")

					.getInputStream());

			@Override

			public Iterator<String> iterator() {

				return new Iterator<String>() {

					@Override

					public boolean hasNext() {

						return counter < N_RECORDS;

					}

					@Override

					public String next() {

						if (counter % 10000 == 0) {

							System.out.println("10K records processed in " + (System.currentTimeMillis() - last) / 1000 + " seconds");

							last = System.currentTimeMillis();

						}

						File f = new File(String.format("/var/lib/eagle/content/EDH/HD%06d.xml", counter++));

						if (f.exists()) {

							try {

								FileInputStream fileInputStream = new FileInputStream(f);

								String s = IOUtils.toString(fileInputStream);

								fileInputStream.close();

								return s;

							} catch (Exception e) {

								return null;

							}

						} else {

							counter++;

							try {

								FileInputStream fileInputStream = new FileInputStream(new File("/var/lib/eagle/content/EDH/HD000009.xml"));

								String s = IOUtils.toString(fileInputStream);

								fileInputStream.close();

								return s;

							} catch (Exception e) {

								return null;

							}

						}

					}

					@Override

					public void remove() {}

				};

			}

		}, false);

	}

}

package eu.dnetlib.data.mdstore.modular.mongodb;

import java.util.ArrayList;

import java.util.Date;

import java.util.List;

import com.google.common.collect.Lists;

import com.mongodb.BasicDBList;

import com.mongodb.BasicDBObject;

import com.mongodb.DBObject;

import com.mongodb.WriteConcern;

import com.mongodb.client.FindIterable;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoDatabase;

import com.mongodb.client.MongoIterable;

import com.mongodb.client.model.Filters;

import com.mongodb.client.model.IndexOptions;

import com.mongodb.client.model.UpdateOptions;

import eu.dnetlib.data.mdstore.MDStoreServiceException;

import eu.dnetlib.data.mdstore.modular.connector.*;

import eu.dnetlib.data.mdstore.modular.mongodb.utils.MDStoreUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.conversions.Bson;

import org.joda.time.DateTime;

import org.joda.time.Days;

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

/**

 * The Class MDStoreTransactionManager.

 */

public class MDStoreTransactionManagerImpl implements MDStoreTransactionManager {

	/** The Constant log. */

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

	/**

	 * The table name.

	 */

	private static String TABLE_NAME = "metadataManager";

	/** The max number of concurrent transactions per mdstore. */

	private int maxTransactions = 1;

	/**

	 * The db.

	 */

	private MongoDatabase db;

	/**

	 * The manager table.

	 */

	private MongoCollection<DBObject> managerTable;

	/** The expired days. */

	private int expiredDays;

	private final IndexOptions options = new IndexOptions().background(true);

	/**

	 * Bootstrap manager.

	 */

	private void bootstrapManager() {

		log.debug("Bootstrap Manager start");

		final MongoCollection<DBObject> metadataColl = db.getCollection("metadata", DBObject.class);

		final FindIterable<DBObject> values = metadataColl.find();

		this.setManagerTable(db.getCollection(TABLE_NAME, DBObject.class));

		for (DBObject object : values) {

			final String id = (String) object.get("mdId");

			String newId = id;

			if (id.contains("_")) {

				newId = StringUtils.substringBefore(id, "_");

			}

			final BasicDBObject input = new BasicDBObject();

			input.put("mdId", id);

			input.put("currentId", newId);

			input.put("expiring", new String[] {});

			input.put("transactions", new String[] {});

			getManagerTable().insertOne(input);

			log.debug(String.format("Added %s to Metadata Manager data structure", id));

		}

		final BasicDBObject ensureIndex = new BasicDBObject();

		ensureIndex.put("mdId", 1);

		log.debug("Create index in MetadaManager ");

		this.getManagerTable().createIndex(ensureIndex, options);

	}

	/**

	 * Gets the DBObject describing an mdstore. null if there is no mdstore with the given id.

	 *

	 * @param mdstoreID the mdStore identifier

	 * @return DBObject or null

	 */

	private DBObject getMDStoreAsDBObject(final String mdstoreID) throws MDStoreServiceException {

		final BasicDBObject query = new BasicDBObject();

		query.put("mdId", mdstoreID);

		final FindIterable<DBObject> it = this.getManagerTable().find(query);

		return it.first();

	}

	/**

	 * Verify consistency.

	 *

	 * @throws MDStoreServiceException

	 *             the MD store service exception

	 */

	@Override

	public void verifyConsistency() throws MDStoreServiceException {

		if (this.getManagerTable() == null) {

			if (!Lists.newArrayList(db.listCollectionNames()).contains(TABLE_NAME))

				bootstrapManager();

			else {

				this.setManagerTable(db.getCollection(TABLE_NAME, DBObject.class));

			}

		}

		if (this.getManagerTable() == null) throw new MDStoreServiceException("Something bad happen, unable to create managerTable");

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#createMDStore(java.lang.String)

	 */

	@Override

	public void createMDStore(final String mdId) throws MDStoreServiceException {

		log.debug("Creating new mdstore");

		verifyConsistency();

		String newId = mdId;

		if (mdId.contains("_")) {

			newId = StringUtils.substringBefore(mdId, "_");

		}

		final BasicDBObject instance = new BasicDBObject();

		instance.put("mdId", mdId);

		instance.put("currentId", newId);

		instance.put("expiring", new String[] {});

		getManagerTable().insertOne(instance);

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#dropMDStore(java.lang.String)

	 */

	@Override

	public void dropMDStore(final String mdId) throws MDStoreServiceException {

		verifyConsistency();

		log.debug("Droping MDStore: " + mdId);

		final BasicDBObject query = new BasicDBObject();

		query.put("mdId", mdId);

		final DBObject dropped = this.getManagerTable().findOneAndDelete(query);

		garbage();

		final String collectionName = (String) dropped.get("currentId");

		this.db.getCollection(collectionName).drop();

		this.db.getCollection("discarded-" + collectionName).drop();

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#getMDStoreCollection(java.lang.String)

	 */

	@Override

	public String getMDStoreCollection(final String mdId) throws MDStoreServiceException {

		verifyConsistency();

		DBObject mdstoreInfo = getMDStoreAsDBObject(mdId);

		if (mdstoreInfo != null)

			return (String) mdstoreInfo.get("currentId");

		else return null;

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#startTransaction(java.lang.String, boolean)

	 */

	@Override

	public String startTransaction(final String mdId, final boolean refresh) throws MDStoreServiceException {

		verifyConsistency();

		log.info("Start transaction for metadata store " + mdId);

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdId);

		if (mdstoreInfo == null) throw new MDStoreServiceException("Error, unable to find Mdstore with Id " + mdId);

		String idCreation = StringUtils.substringBefore(mdId, "_");

		idCreation = idCreation + "::" + System.currentTimeMillis();

		BasicDBList values;

		if (mdstoreInfo.containsField("transactions")) {

			values = (BasicDBList) mdstoreInfo.get("transactions");

			if (values.size() > getMaxTransactions())

				throw new MDStoreServiceException("Cannot create more than " + getMaxTransactions() + " transactions, found: " + values.size() + ", mdId:"

						+ mdId);

		} else {

			values = new BasicDBList();

		}

		final BasicDBObject transactionMetadata = new BasicDBObject();

		transactionMetadata.put("id", idCreation);

		transactionMetadata.put("refresh", refresh);

		transactionMetadata.put("date", new Date());

		values.add(transactionMetadata);

		mdstoreInfo.put("transactions", values);

		this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", mdstoreInfo.get("_id")), mdstoreInfo);

		return idCreation;

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#commit(java.lang.String, java.lang.String,

	 *      eu.dnetlib.data.mdstore.modular.connector.MDStore)

	 */

	@Override

	public boolean commit(final String transactionId, final String mdstoreId, final MDStore current) throws MDStoreServiceException {

		verifyConsistency();

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdstoreId);

		if (mdstoreInfo == null) throw new MDStoreServiceException("Error, unable to find Mdstore with Id " + mdstoreId);

		final BasicDBList transactions = (BasicDBList) mdstoreInfo.get("transactions");

		final DBObject transaction = findTransaction(transactions, transactionId);

		if (transaction == null) throw new MDStoreServiceException("Error, unable to find transaction with Id " + transactionId);

		final boolean refresh = (Boolean) transaction.get("refresh");

		transactions.remove(transaction);

		final String oldId = (String) mdstoreInfo.get("currentId");

		if (refresh) {

			mdstoreInfo.put("currentId", transactionId);

			final BasicDBList stillUsed = (BasicDBList) mdstoreInfo.get("expiring");

			if (stillUsed.size() == 0) {

				db.getCollection(oldId).drop();

				db.getCollection("discarded-" + oldId).drop();

			}

			log.debug("Replaced collection ");

		} else {

			log.debug("commit incremental ");

			updateIncremental(transactionId, oldId);

			db.getCollection(transactionId).drop();

			db.getCollection("discarded-" + transactionId).drop();

		}

		this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", mdstoreInfo.get("_id")), mdstoreInfo);

		log.info("Committed transaction for metadata store " + mdstoreId);

		return true;

	}

	/**

	 * Find transaction.

	 *

	 * @param transactions

	 *            the transactions

	 * @param transactionId

	 *            the transaction id

	 * @return the DB object

	 */

	private DBObject findTransaction(final BasicDBList transactions, final String transactionId) {

		if (transactions.size() == 0) return null;

		for (Object tx : transactions) {

			final BasicDBObject transaction = (BasicDBObject) tx;

			final String id = (String) transaction.get("id");

			if (transactionId.equals(id)) return transaction;

		}

		return null;

	}

	/**

	 * Gets the db.

	 *

	 * @return the db

	 */

	public MongoDatabase getDb() {

		return db;

	}

	/**

	 * Sets the db.

	 *

	 * @param db

	 *            the db to set

	 */

	@Required

	public void setDb(final MongoDatabase db) {

		this.db = db;

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#readMdStore(java.lang.String)

	 */

	@Override

	public String readMdStore(final String mdStoreId) throws MDStoreServiceException {

		verifyConsistency();

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdStoreId);

		if (mdstoreInfo == null) throw new MDStoreServiceException("Error, unable to find Mdstore with Id " + mdStoreId);

		final String currentId = (String) mdstoreInfo.get("currentId");

		final BasicDBList values = (BasicDBList) mdstoreInfo.get("expiring");

		updateMdstoreUsed(values, currentId);

		this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", mdstoreInfo.get("_id")), mdstoreInfo);

		return currentId;

	}

	/**

	 * Update mdstore used.

	 *

	 * @param values

	 *            the values

	 * @param mdId

	 *            the md id

	 */

	private void updateMdstoreUsed(final BasicDBList values, final String mdId) {

		if (values.size() > 0) {

			for (Object value : values) {

				final DBObject obj = (DBObject) value;

				final String id = (String) obj.get("id");

				if (mdId.equals(id)) {

					obj.put("lastRead", new Date());

					return;

				}

			}

		}

		final BasicDBObject readStore = new BasicDBObject();

		readStore.put("id", mdId);

		readStore.put("lastRead", new Date());

		values.add(readStore);

	}

	/**

	 * Gets the manager table.

	 *

	 * @return the managerTable

	 */

	public MongoCollection<DBObject> getManagerTable() {

		return managerTable;

	}

	/**

	 * Sets the manager table.

	 *

	 * @param managerTable

	 *            the managerTable to set

	 */

	public void setManagerTable(final MongoCollection<DBObject> managerTable) {

		this.managerTable = managerTable;

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#getInfoForCurrentMdStore(java.lang.String)

	 */

	@Override

	public MDStoreManagerInfo getInfoForCurrentMdStore(final String mdStoreId) throws MDStoreServiceException {

		verifyConsistency();

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdStoreId);

		if (mdstoreInfo == null) throw new MDStoreServiceException("Error, unable to find Mdstore with Id " + mdStoreId);

		final MDStoreManagerInfo result = new MDStoreManagerInfo();

		result.setCurrentId((String) mdstoreInfo.get("currentId"));

		result.setMdId((String) mdstoreInfo.get("mdId"));

		final BasicDBList values = (BasicDBList) mdstoreInfo.get("expiring");

		for (Object v : values) {

			final MDStoreExpiredInfo stillused = new MDStoreExpiredInfo();

			final DBObject value = (DBObject) v;

			stillused.setId((String) value.get("id"));

			stillused.setLastRead((Date) value.get("lastRead"));

			result.addExpiredItem(stillused);

		}

		final BasicDBList transactions = (BasicDBList) mdstoreInfo.get("transactions");

		if (transactions != null) {

			for (Object tx : transactions) {

				final MDStoreTransactionInfo transaction = new MDStoreTransactionInfo();

				final DBObject value = (DBObject) tx;

				final String transactionId = (String) value.get("id");

				transaction.setId(transactionId);

				transaction.setDate((Date) value.get("date"));

				transaction.setRefresh((Boolean) value.get("refresh"));

				transaction.setSize(db.getCollection(transactionId).count());

				result.addTransactionInfo(transaction);

			}

		}

		return result;

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#dropUsed(java.lang.String, java.lang.String)

	 */

	@Override

	public Boolean dropUsed(final String mdId, final String idToDrop) throws MDStoreServiceException {

		verifyConsistency();

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdId);

		if (mdstoreInfo == null) throw new MDStoreServiceException("Error, unable to find Mdstore with Id " + mdId);

		return dropStore(mdstoreInfo, idToDrop, "expiring");

	}

	private boolean dropStore(DBObject mdstoreInfo, final String idToDrop, String transactionListName) throws MDStoreServiceException {

		final BasicDBList transactionList = (BasicDBList) mdstoreInfo.get(transactionListName);

		for (int i = 0; i < transactionList.size(); i++) {

			final DBObject value = (DBObject) transactionList.get(i);

			final String currentUsedId = (String) value.get("id");

			if (currentUsedId.equals(idToDrop)) {

				db.getCollection(idToDrop).drop();

				db.getCollection("discarded-" + idToDrop).drop();

				transactionList.remove(value);

				mdstoreInfo.put(transactionListName, transactionList);

				this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", mdstoreInfo.get("_id")), mdstoreInfo);

				return true;

			}

		}

		throw new MDStoreServiceException("Error, unable to drop collection " + idToDrop);

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#garbage()

	 */

	@Override

	public void garbage() throws MDStoreServiceException {

		verifyConsistency();

		log.info("Start garbage collection of MdStore");

		final FindIterable<DBObject> it = this.managerTable.find();

		int totalDeleted = 0;

		for (DBObject currentObject :  it){

			if (log.isDebugEnabled()) {

				log.debug("start to check id: " + currentObject.get("currentId"));

			}

			garbageExpiring(currentObject, (String) currentObject.get("currentId"));

			garbageTransactions(currentObject, (String) currentObject.get("currentId"));

			this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", currentObject.get("_id")), currentObject);

		}

		// DELETING Collection that are not in the metadataManager table

		MongoIterable<String> collections = this.db.listCollectionNames();

		for (String collection : collections) {

			if ((collection.length() > 30) && (!collection.contains("discarded-"))) {

				final DBObject item = getMetadataObjectForCollections(collection);

				if (shouldDelete(collection, item)) {

					if (log.isDebugEnabled()) {

						log.debug("delete collection: " + collection + " from mongo");

					}

					db.getCollection(collection).drop();

					db.getCollection("discarded-" + collection).drop();

					if (log.isDebugEnabled()) {

						log.debug("delete collection: discarded-" + collection + " from mongo");

					}

				}

			}

		}

		log.info("Complete garbage collection of MdStore, total store deleted: " + totalDeleted);

	}

	private DBObject getMetadataObjectForCollections(final String collectionName) {

		if (collectionName == null) return null;

		final String postfix = "_TURTdG9yZURTUmVzb3VyY2VzL01EU3RvcmVEU1Jlc291cmNlVHlwZQ==";

		final String tmp = collectionName.contains("discarded-") ? StringUtils.substringAfter(collectionName, "discarded-") : collectionName;

		final String collectionNameCleaned = StringUtils.substringBefore(tmp, "::") + postfix;

		//DBObject query = QueryBuilder.start("mdId").is(collectionNameCleaned).get();

		Bson query = Filters.eq("mdId", collectionNameCleaned);

		return this.managerTable.find(query).first();

	}

	private boolean shouldDelete(final String collectionName, final DBObject metadataManagerInstance) {

		log.debug("should delete instance "+metadataManagerInstance);

		if((metadataManagerInstance== null) || (metadataManagerInstance.get("currentId")== null)){

			log.debug("the instance has not currentID");

			return true;

		}

		String currentId = (String) metadataManagerInstance.get("currentId");

		if (collectionName.equals(currentId)) return false;

		BasicDBList expiringList = (BasicDBList) metadataManagerInstance.get("expiring");

		if (findInList(expiringList, collectionName, "id")) return false;

		BasicDBList transactionsList = (BasicDBList) metadataManagerInstance.get("transactions");

		return !findInList(transactionsList, collectionName, "id");

	}

	private boolean findInList(final BasicDBList list, final String object, final String tagname) {

		if (list == null) return false;

		for (Object o : list) {

			DBObject currentObject = (DBObject) o;

			final String id = (String) currentObject.get(tagname);

			if (id.equals(object)) return true;

		}

		return false;

	}

	/**

	 * Delete.

	 *

	 * @param list

	 *            the list

	 * @param toRemove

	 *            the to remove

	 */

	private void delete(final BasicDBList list, final List<DBObject> toRemove) {

		for (final DBObject obj : toRemove) {

			if (log.isDebugEnabled()) {

				log.debug("deleting " + obj);

			}

			list.remove(obj);

		}

	}

	/**

	 * Garbage transactions.

	 *

	 * @param currentObject

	 *            the current object

	 * @param currentId

	 *            the current id

	 */

	private void garbageTransactions(final DBObject currentObject, final String currentId) {

		if (log.isDebugEnabled()) {

			log.debug("Start garbage transactions ");

		}

		final BasicDBList expiring = (BasicDBList) currentObject.get("transactions");

		if ((expiring == null) || (expiring.size() <= getMaxTransactions())) return;

		List<DBObject> expiringList = Lists.newArrayList();

		for (Object o : expiring) {

			final DBObject cobj = (DBObject) o;

			if (cobj != null) {

				expiringList.add((DBObject) o);

			}

		}

		expiringList.sort(MDStoreUtils.getComparatorOnDate());

		List<DBObject> toRemove = Lists.newArrayList();

		int i = 0;

		// We should remove the k item less recent

		// where k = numberOftotalTransaction - maxNumberOfTransaction

		// k = numberOfItemToRemove

		while (((expiringList.size() - toRemove.size()) > getMaxTransactions()) || (i < expiringList.size())) {

			DBObject currentObj = expiringList.get(i++);

			String objectId = (String) currentObj.get("id");

			if (!objectId.equals(currentId)) {

				if (log.isDebugEnabled()) {

					log.debug("delete collection: " + objectId + " from mongo");

				}

				db.getCollection(objectId).drop();

				db.getCollection("discarded-" + objectId).drop();

				if (log.isDebugEnabled()) {

					log.debug("delete collection: discarded-" + objectId + " from mongo");

				}

				toRemove.add(currentObj);

			} else {

				if (log.isDebugEnabled()) {

					log.debug("Cannot remove transaction " + objectId + " because is the currentId: " + currentId);

				}

			}

		}

		delete(expiring, toRemove);

		log.info("Deleted " + toRemove.size() + " transactions, mdStore Id:" + currentObject.get("mdId"));

	}

	/**

	 * Garbage expiring.

	 *

	 * @param currentObject

	 *            the current object

	 * @param currentId

	 *            the current id

	 */

	private void garbageExpiring(final DBObject currentObject, final String currentId) {

		if (log.isDebugEnabled()) {

			log.debug("Start to search expiring mdstores for id: " + currentObject.get("mdId"));

		}

		final BasicDBList expiring = (BasicDBList) currentObject.get("expiring");

		final List<DBObject> toRemove = Lists.newArrayList();

		if (log.isDebugEnabled()) {

			if (expiring == null) {

				log.debug("expiring list is null");

			} else {

				log.debug("expiring list size is :" + expiring.size());

			}

		}

		if ((expiring == null) || (expiring.size() == 0)) {

			log.debug("Deleted  0  expired  collections, mdStore Id:" + currentObject.get("mdId"));

			return;

		}

		for (Object anExpiring : expiring) {

			final DBObject currentExpiredStore = (DBObject) anExpiring;

			final String currentUsedId = (String) currentExpiredStore.get("id");

			final Days d = getExpiringDays(currentExpiredStore, "lastRead");

			if (log.isDebugEnabled()) {

				log.debug("the store :" + currentId + " expired since " + d.getDays() + "days ");

			}

			// DELETE the collection where the last time they was read

			// is more than 3 days ago

			if (d.getDays() > getExpiredDays()) {

				if (!currentUsedId.equals(currentId)) {

					db.getCollection(currentUsedId).drop();

					db.getCollection("discarded-" + currentUsedId).drop();

					log.debug("deleted collection " + currentUsedId);

				}

				toRemove.add(currentExpiredStore);

			}

		}

		delete(expiring, toRemove);

		log.debug("Deleted expired " + toRemove.size() + "collections, mdStore Id:" + currentObject.get("mdId"));

	}

	/**

	 * Gets the expiring days.

	 *

	 * @param value

	 *            the value

	 * @param paramName

	 *            the param name

	 * @return the expiring days

	 */

	private Days getExpiringDays(final DBObject value, final String paramName) {

		final Date lastRead = (Date) value.get(paramName);

		final DateTime last = new DateTime(lastRead);

		final DateTime today = new DateTime();

		final Days d = Days.daysBetween(last, today);

		return d;

	}

	/**

	 * Gets the expired days.

	 *

	 * @return the expiredDays

	 */

	public int getExpiredDays() {

		if (this.expiredDays == 0) return 3;

		return expiredDays;

	}

	/**

	 * Sets the expired days.

	 *

	 * @param expiredDays

	 *            the expiredDays to set

	 */

	public void setExpiredDays(final int expiredDays) {

		this.expiredDays = expiredDays;

	}

	/**

	 * Update incremental.

	 *

	 * @param transactionId

	 *            the transaction id

	 * @param currentId

	 *            the current id

	 */

	private void updateIncremental(final String transactionId, final String currentId) {

		final MongoCollection<DBObject> transaction = db.getCollection(transactionId, DBObject.class);

		final MongoCollection<DBObject> mdstore = db.getCollection(currentId, DBObject.class);

		final FindIterable<DBObject> it = transaction.find().noCursorTimeout(true);

		for (DBObject currentObj : it) {

			final String id = (String) currentObj.get("id");

			DBObject newObj = new BasicDBObject(currentObj.toMap());

			//setting to journaled write concern to be sure that when the write returns everything has been flushed to disk (https://docs.mongodb.org/manual/faq/developers/#when-does-mongodb-write-updates-to-disk)

			//the explicit fsync command can't be run anymore: 'Command failed with error 13: 'fsync may only be run against the admin database.'

			final MongoCollection<DBObject> mdstoreWrite = mdstore.withWriteConcern(WriteConcern.JOURNALED);

			mdstoreWrite.replaceOne(new BasicDBObject("id", id), newObj, new UpdateOptions().upsert(true));

		}

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#dropTransaction(java.lang.String, java.lang.String)

	 */

	@Override

	public Boolean dropTransaction(final String mdId, final String idToDrop) throws MDStoreServiceException {

		verifyConsistency();

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdId);

		if (mdstoreInfo == null) throw new MDStoreServiceException("Error, unable to find Mdstore with Id " + mdId);

		return dropStore(mdstoreInfo, idToDrop, "transactions");

	}

	public void garbageTransactionsOnStart() throws MDStoreServiceException {

		verifyConsistency();

		FindIterable<DBObject> it = this.managerTable.find();

		final List<String> currentMdStoreId = Lists.newArrayList();

		for (DBObject currentObject : it){

			currentMdStoreId.add((String) currentObject.get("currentId"));

			final BasicDBList transactions = (BasicDBList) currentObject.get("transactions");

			if ((transactions != null) && (transactions.size() > 0)) {

				for (Object tx : transactions) {

					final DBObject currentTransactions = (DBObject) tx;

					final String id = (String) currentTransactions.get("id");

					db.getCollection(id).drop();

					db.getCollection("discarded-" + id).drop();

					log.debug("deleted collection " + id);

				}

				currentObject.put("transactions", new BasicDBList());

				this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", currentObject.get("_id")), currentObject);

			}

		}

		//DELETING ALL THE DISCARDED COLLECTION THAT DISCARDED COLLECTION OF THE CURRENT MDSTORE

		final ArrayList<String> collectionsNames = Lists.newArrayList(db.listCollectionNames());

		for (String item : collectionsNames) {

			if (item.startsWith("discarded-")) {

				final String currentCollection = StringUtils.substringAfter(item, "discarded-");

				if (!currentMdStoreId.contains(currentCollection)) {

					log.info("Deleting discarded collection :" + item);

					this.db.getCollection(item).drop();

				}

			}

		}

	}

	/**

	 * Gets the max transactions.

	 *

	 * @return the maxTransactions

	 */

	public int getMaxTransactions() {

		return maxTransactions;

	}

	/**

	 * Sets the max transactions.

	 *

	 * @param maxTransactions

	 *            the maxTransactions to set

	 */

	public void setMaxTransactions(final int maxTransactions) {

		this.maxTransactions = maxTransactions;

	}

}

<?xml version="1.0" encoding="UTF-8"?>

<record xmlns:dc="http://purl.org/dc/elements/1.1/"

        xmlns:dr="http://www.driver-repository.eu/namespace/dr"

        xmlns:dri="http://www.driver-repository.eu/namespace/dri"

        xmlns:oaf="http://namespace.openaire.eu/oaf"

        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

    <header xmlns="http://namespace.openaire.eu/">

        <dri:objIdentifier>od______2367::0001a50c6388e9bfcb791a924ec4b837</dri:objIdentifier>

        <dri:recordIdentifier>oai:pumaoai.isti.cnr.it:cnr.imati/cnr.ian.pv/1999-PP-018</dri:recordIdentifier>

        <dri:dateOfCollection/>

        <dri:mdFormat/>

        <dri:mdFormatInterpretation/>

        <dri:repositoryId>

            10d18b66-1d2a-4579-9adc-aa57b0821c7f_UmVwb3NpdG9yeVNlcnZpY2VSZXNvdXJjZXMvUmVwb3NpdG9yeVNlcnZpY2VSZXNvdXJjZVR5cGU=

        </dri:repositoryId>

        <dr:objectIdentifier/>

        <dr:dateOfCollection>2016-11-22T09:15:30.817Z</dr:dateOfCollection>

        <dr:dateOfTransformation>2016-11-22T09:18:15.904Z</dr:dateOfTransformation>

        <oaf:datasourceprefix>od______2367</oaf:datasourceprefix>

    </header>

    <metadata xmlns="http://namespace.openaire.eu/">

        <dc:title>Anisotropic mechanisms for multiphasic unipolar electrograms. Simulation studies and experimental

            recordings

        </dc:title>

        <dc:creator>Colli Franzone, Piero,</dc:creator>

        <dc:creator>Guerri, Luciano,</dc:creator>

        <dc:creator>Pennacchio, Micol,</dc:creator>

        <dc:creator>Taccardi, Bruno</dc:creator>

        <dc:date>1999-09-30</dc:date>

        <dc:description>The origin of the multiple, complex morphologies observed in

            unipolar epicardial electrograms, and their relationships with myocardial

            architecture, have not been fully elucidated. To clarify this problem we

            simulated electrograms (EGs) with a model representing the heart as an

            anisotropic bidomain with unequal anisotropy ratio, ellipsoidal ventricular

            geometry, transmural fiber rotation, epi-endocardial obliqueness of fiber

            direction and a simplified conduction system. The electrograms were compared

            with those directly recorded from the surface of isolated dog hearts immersed

            in a conducting medium. The model accurately reproduced the recorded EG

            morphologies for excitation wave fronts that reach the recording sites by

            spreading either along or across fibers.The

            origin of the multiple waves that constitute the QRS complex could be better

            understood after splitting the current sources, the potential distributions and

            the EGs into a field component (further subdivided into an axial and a conormal component) and a "reference"

            component. The split model provides an explanation of the interaction between the three-dimensional geometry

            and direction of

            propagation of a spreading wave front, the architecture of the fibers through

            which excitation is spreading, the potential distributions and the QRS wave

            forms. Because epicardial potentials, electrograms and isochrone contours can be

            computed noninvasively from body surface measurements, interpreting epicardial

            EGs in terms of intramural events may have clinical relevance.

        </dc:description>

        <dc:identifier>http://puma.isti.cnr.it/dfdownloadnew.php?ident=cnr.imati/cnr.ian.pv/1999-PP-018</dc:identifier>

        <dc:identifier>

            http://puma.isti.cnr.it/rmydownload.php?filename=cnr.imati/cnr.ian.pv/1999-PP-018/1999-PP-018_0.ps

        </dc:identifier>

        <dc:language>eng</dc:language>

        <dc:source>Preprint ercim.cnr.ian//1999-1151, 1999.</dc:source>

        <dc:subject>Electrograms, bidomain model, reference potential, cardiac potential maps, anisotropic propagation,

            source splitting

        </dc:subject>

        <dc:subject>info:eu-repo/classification/msc/78A70,65N30</dc:subject>

        <dc:rights>info:eu-repo/semantics/openAccess</dc:rights>

        <dc:type>info:eu-repo/semantics/preprint</dc:type>

        <dr:CobjCategory>0016</dr:CobjCategory>

        <dr:CobjIdentifier/>

        <oaf:dateAccepted>1999-09-30</oaf:dateAccepted>

        <oaf:collectedDatasourceid>opendoar____::2367</oaf:collectedDatasourceid>

        <oaf:accessrights>OPEN</oaf:accessrights>

        <oaf:hostedBy id="opendoar____::2367" name="PUblication MAnagement"/>

        <oaf:collectedFrom id="opendoar____::2367" name="PUblication MAnagement"/>

    </metadata>

    <about xmlns:oai="http://www.openarchives.org/OAI/2.0/" xmlns="http://namespace.openaire.eu/">

        <provenance xmlns="http://www.openarchives.org/OAI/2.0/provenance"

                    xsi:schemaLocation="http://www.openarchives.org/OAI/2.0/provenance http://www.openarchives.org/OAI/2.0/provenance.xsd">

            <originDescription altered="true" harvestDate="2016-11-22T09:15:30.817Z">

                <baseURL>http://pumaoai.isti.cnr.it/openoai2.php</baseURL>

                <identifier>oai:pumaoai.isti.cnr.it:cnr.imati/cnr.ian.pv/1999-PP-018</identifier>

                <datestamp>1999-10-08</datestamp>

                <metadataNamespace>http://www.openarchives.org/OAI/2.0/oai_dc/</metadataNamespace>

            </originDescription>

        </provenance>

        <oaf:datainfo>

            <oaf:inferred>false</oaf:inferred>

            <oaf:deletedbyinference>false</oaf:deletedbyinference>

            <oaf:trust>0.9</oaf:trust>

            <oaf:inferenceprovenance/>

            <oaf:provenanceaction classid="sysimport:crosswalk:repository"

                                  classname="sysimport:crosswalk:repository"

                                  schemeid="dnet:provenanceActions" schemename="dnet:provenanceActions"/>

        </oaf:datainfo>

    </about>

</record>

package eu.dnetlib.data.mdstore.modular.mongodb;

import java.util.List;

import java.util.Map;

import javax.annotation.Nullable;

import com.google.common.base.Predicate;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.mongodb.BasicDBObject;

import com.mongodb.DBObject;

import com.mongodb.WriteConcern;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.model.*;

import eu.dnetlib.data.mdstore.MDStoreServiceException;

import eu.dnetlib.data.mdstore.modular.MDFormatDescription;

import eu.dnetlib.data.mdstore.modular.RecordParser;

import eu.dnetlib.data.mdstore.modular.mongodb.utils.IndexFieldRecordParser;

import eu.dnetlib.data.mdstore.modular.mongodb.utils.IndexFieldRecordParserException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import static eu.dnetlib.data.mdstore.modular.MDStoreConstants.*;

public class MongoBulkWritesManager {

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

	private final boolean discardRecords;

	private final boolean indexRecords;

	private final IndexFieldRecordParser indexFieldRecordParser = new IndexFieldRecordParser();

	private final List<MDFormatDescription> mdref;

	private RecordParser recordParser;

	private MongoCollection<DBObject> validCollection;

	private List<WriteModel<DBObject>> validBulkOperationList;

	private BulkWriteOptions writeOptions;

	private MongoCollection<DBObject> discardedCollection;

	private List<WriteModel<DBObject>> discardedBulkOperationList;

	private int bulkSize;

	public MongoBulkWritesManager(final MongoCollection<DBObject> collection,

			final MongoCollection<DBObject> discardedCollection,

			final List<MDFormatDescription> mdref,

			final int bulkSize,

			final RecordParser parser,

			final boolean discardRecords) {

		this.validCollection = collection.withWriteConcern(WriteConcern.ACKNOWLEDGED);

		this.validBulkOperationList = Lists.newArrayList();

		this.discardedCollection = discardedCollection.withWriteConcern(WriteConcern.ACKNOWLEDGED);

		this.discardedBulkOperationList = Lists.newArrayList();

		this.bulkSize = bulkSize;

		this.recordParser = parser;

		this.discardRecords = discardRecords;

		this.mdref = mdref;

		this.indexRecords = (this.mdref != null &&  !this.mdref.isEmpty());

		this.writeOptions = new BulkWriteOptions().ordered(false);

	}

	public void insert(final String record) throws MDStoreServiceException {

		Map<String, String> recordProperties = null;

		try {

			recordProperties = recordParser.parseRecord(record);

		} catch (Throwable e) {

			if (discardRecords) {

				log.debug("unhandled exception: " + e.getMessage());

				discardRecord(record);

			}

		}

		Map<String, List<String>> indexRecordField = null;

		try {

			if (indexRecords) {

				indexRecordField = indexFieldRecordParser.parseRecord(record, mdref);

			}

		} catch (IndexFieldRecordParserException e) {

			// could not index record fields

			throw new MDStoreServiceException("Are you using the correct type of store / index definition for the records in " + validCollection.getNamespace() + " ?", e);

		}

		log.debug("found props: " + recordProperties);

		if (recordProperties.containsKey(ID)) {

			final DBObject obj = buildDBObject(record, recordProperties, indexRecordField);

			if (log.isDebugEnabled()) {

				log.debug("Saving object" + obj);

			}

			validBulkOperationList.add(new ReplaceOneModel(new BasicDBObject(ID, obj.get(ID)), obj, new UpdateOptions().upsert(true)));

			if (((validBulkOperationList.size() % bulkSize) == 0) && !validBulkOperationList.isEmpty()) {

				validCollection.bulkWrite(validBulkOperationList, writeOptions);

				validBulkOperationList.clear();

			}

		} else {

			if (discardRecords) {

				log.debug("parsed record seems invalid");

				discardRecord(record);

			}

		}

	}

	private void discardRecord(final String record) {

		discardedBulkOperationList.add(new InsertOneModel(new BasicDBObject(BODY, record)));

		if (((discardedBulkOperationList.size() % bulkSize) == 0) && !discardedBulkOperationList.isEmpty()) {

			discardedCollection.bulkWrite(discardedBulkOperationList, writeOptions);

			discardedBulkOperationList.clear();

		}

	}

	public void flushBulks() {

		//setting to journaled write concern to be sure that when the write returns everything has been flushed to disk (https://docs.mongodb.org/manual/faq/developers/#when-does-mongodb-write-updates-to-disk)

		//the explicit fsync command can't be run anymore: 'Command failed with error 13: 'fsync may only be run against the admin database.'

		if (!validBulkOperationList.isEmpty()) {

			validCollection = getCollectionWithWriteConcern(validCollection, WriteConcern.JOURNALED);

			validCollection.bulkWrite(validBulkOperationList, writeOptions);

		}

		if (!discardedBulkOperationList.isEmpty()) {

			discardedCollection = getCollectionWithWriteConcern(discardedCollection, WriteConcern.JOURNALED);

			discardedCollection.bulkWrite(discardedBulkOperationList, writeOptions);

		}

		//setting write concern back to ACKNOWLEDGE to avoid the execution of future writes all in Journaled mode

		validCollection = getCollectionWithWriteConcern(validCollection, WriteConcern.ACKNOWLEDGED);

		discardedCollection = getCollectionWithWriteConcern(discardedCollection, WriteConcern.ACKNOWLEDGED);

	}

	protected DBObject buildDBObject(final String record, final Map<String, String> recordProperties, final Map<String, List<String>> indexFieldProperties) {

		final DBObject obj = new BasicDBObject();

		obj.put(ID, recordProperties.get(ID));

		obj.put(ORIGINALID, recordProperties.get(ORIGINALID));

		obj.put(BODY, record);

		obj.put(TIMESTAMP, Long.valueOf(recordProperties.get(TIMESTAMP)));

		if (indexFieldProperties != null)

			obj.putAll(Maps.filterKeys(indexFieldProperties, new Predicate<String>() {

				//ensure we do not override the mandatory fields above with some unexpected value

				@Override

				public boolean apply(@Nullable final String s) {

					return !s.equalsIgnoreCase(ID) && !s.equalsIgnoreCase(ORIGINALID) && !s.equalsIgnoreCase(BODY) && !s.equalsIgnoreCase(TIMESTAMP);

				}

			}));

		return obj;

	}

	private MongoCollection<DBObject> getCollectionWithWriteConcern(MongoCollection<DBObject> collection, WriteConcern writeConcern) {

		return collection.withWriteConcern(writeConcern);

	}

}

{"type_source": "SVN", "goal": "package -U -T 4C source:jar", "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet45/modules/cnr-mongo-mdstore/trunk/", "deploy_repository": "dnet45-snapshots", "version": "4", "mail": "sandro.labruzzo@isti.cnr.it,michele.artini@isti.cnr.it, claudio.atzori@isti.cnr.it, alessia.bardi@isti.cnr.it", "deploy_repository_url": "http://maven.research-infrastructures.eu/nexus/content/repositories/dnet45-snapshots", "name": "cnr-mongo-mdstore"}

package eu.dnetlib.data.mdstore.modular.mongodb;

import java.util.List;

import com.google.common.collect.Lists;

import com.google.common.primitives.Ints;

import com.mongodb.BasicDBObject;

import com.mongodb.BasicDBObjectBuilder;

import com.mongodb.DBObject;

import com.mongodb.client.AggregateIterable;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoDatabase;

import com.mongodb.client.model.CreateCollectionOptions;

import eu.dnetlib.data.mdstore.MDStoreServiceException;

import eu.dnetlib.data.mdstore.modular.MDStoreDescription;

import eu.dnetlib.data.mdstore.modular.RecordParser;

import eu.dnetlib.data.mdstore.modular.RecordParserFactory;

import eu.dnetlib.data.mdstore.modular.connector.MDStore;

import eu.dnetlib.data.mdstore.modular.connector.MDStoreDBStatus;

import eu.dnetlib.data.mdstore.modular.connector.MDStoreDao;

import eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager;

import eu.dnetlib.data.mdstore.modular.mongodb.utils.MDStoreUtils;

import eu.dnetlib.miscutils.collections.FilteredCollection;

import eu.dnetlib.miscutils.collections.MappedCollection;

import eu.dnetlib.miscutils.datetime.DateUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.Document;

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

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

/**

 * Factory bean for MongoMDStore instances.

 *

 * @author marko

 */

public class MDStoreDaoImpl implements MDStoreDao {

	public static final String MD_ID = "mdId";

	public static final String FORMAT = "format";

	public static final String INTERPRETATION = "interpretation";

	public static final String LAYOUT = "layout";

	public static final String SIZE = "size";

	public static final String METADATA_NAME = "metadata";

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

	private MongoDatabase db;

	private RecordParserFactory recordParserFactory;

	private boolean discardRecords = true;

	@Autowired

	private MDStoreTransactionManager transactionManager;

	/**

	 * {@inheritDoc}

	 */

	@Override

	public void createMDStore(final String mdId, final String format, final String interpretation, final String layout) throws MDStoreServiceException {

		transactionManager.createMDStore(mdId);

		final String internalId = transactionManager.getMDStoreCollection(mdId);

		if (!Lists.newArrayList(getDb().listCollectionNames()).contains(internalId)) {

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

			getDb().createCollection(internalId, new CreateCollectionOptions());

		}

		final MongoCollection<DBObject> coll = getDb().getCollection(METADATA_NAME, DBObject.class);

		final BasicDBObject obj = new BasicDBObject();

		obj.put(MD_ID, mdId);

		obj.put(FORMAT, format);

		obj.put(INTERPRETATION, interpretation);

		obj.put(LAYOUT, layout);

		obj.put(SIZE, 0);

		coll.insertOne(obj);

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public void deleteMDStore(final String mdId) throws MDStoreServiceException {

		final MongoCollection<DBObject> metadata = getDb().getCollection(METADATA_NAME, DBObject.class);

		if (metadata == null) throw new MDStoreServiceException("cannot find metadata collection");

		transactionManager.dropMDStore(mdId);

		metadata.deleteOne(new BasicDBObject(MD_ID, mdId));

		log.info("deleted mdId: " + mdId);

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public MDStore getMDStore(final String mdId) throws MDStoreServiceException {

		final String internalId = transactionManager.getMDStoreCollection(mdId);

		return new MongoMDStore(mdId, getDb().getCollection(internalId, DBObject.class), getRecordParser(), isDiscardRecords(), getDb());

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public MDStore readMDStore(final String mdId) throws MDStoreServiceException {

		final String internalId = transactionManager.readMdStore(mdId);

		return new MongoMDStore(mdId, getDb().getCollection(internalId, DBObject.class), getRecordParser(), isDiscardRecords(), getDb());

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public MDStore startTransaction(final String mdId, final boolean refresh) throws MDStoreServiceException {

		final String transactionId = transactionManager.startTransaction(mdId, refresh);

		return new MongoMDStore(transactionId, getDb().getCollection(transactionId, DBObject.class), getRecordParser(), isDiscardRecords(),

				getDb());

	}

	private RecordParser getRecordParser() {

		final RecordParser parser = getRecordParserFactory().newInstance();

		parser.setTimestamp(DateUtils.now());

		return parser;

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public List<MDStoreDescription> listMDStores() {

		return MappedCollection.listMap(getDb().getCollection(METADATA_NAME, DBObject.class).find(), input -> {

			final String mdId = (String) input.get(MD_ID);

			log.debug("Getting info for " + mdId);

			final String format = (String) input.get(FORMAT);

			final String layout = (String) input.get(LAYOUT);

			final String interpretation = (String) input.get(INTERPRETATION);

			MongoMDStore currentMDStore = null;

			final MDStoreDescription description = new MDStoreDescription();

			try {

				currentMDStore = (MongoMDStore) getMDStore(mdId);

			} catch (final MDStoreServiceException e) {

				log.error("Error on retrieving mdstore for getting info mdId " + mdId);

			}

			int size = 0;

			if (input.containsField(SIZE)) {

				log.debug("Size retrieved from metadata for mdId :" + mdId);

				size = (Integer) input.get(SIZE);

			} else {

				if (currentMDStore != null) {

					log.debug("Size not Found in metadata for mdId :" + mdId + " calling getCount ");

					size = currentMDStore.getSize();

					input.put("size", size);

					getDb().getCollection(METADATA_NAME, DBObject.class).findOneAndReplace(new BasicDBObject(MD_ID, mdId), input);

				}

			}

			if (currentMDStore != null) {

				description.setIndexed(currentMDStore.isIndexed());

			}

			description.setId(mdId);

			description.setFormat(format);

			description.setLayout(layout);

			description.setInterpretation(interpretation);

			description.setSize(size);

			return description;

		});

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public List<String> listMDStores(final String format, final String layout, final String interpretation) {

		return MappedCollection.listMap(

				FilteredCollection.listFilter(getDb().getCollection(METADATA_NAME, DBObject.class).find(), MDStoreUtils.dboFilter(format, layout, interpretation)),

				MDStoreUtils.mdId());

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public int getCachedSize(final String id) throws MDStoreServiceException {

		log.debug("retrieve cached size for mdstore: " + id);

		final DBObject desc = getDb().getCollection(METADATA_NAME, DBObject.class).find(new BasicDBObject(MD_ID, id)).first();

		if (!desc.containsField(SIZE)) {

			desc.put(SIZE, getMDStore(id).getSize());

		}

		final Object oSize = desc.get(SIZE);

		return (Integer) oSize;

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public void refreshSizes() throws MDStoreServiceException {

		for (final MDStoreDescription mdStoreId : listMDStores()) {

			refreshSize(mdStoreId.getId());

		}

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public int refreshSize(final String mdStoreId) throws MDStoreServiceException {

		final int size = (int) getDb().getCollection(transactionManager.getMDStoreCollection(mdStoreId)).count();

		final MongoCollection<DBObject> metadata = getDb().getCollection(METADATA_NAME, DBObject.class);

		metadata.updateOne(new BasicDBObject(MD_ID, mdStoreId), new BasicDBObject("$set", new BasicDBObject(SIZE, size)));

		return size;

	}

	@Override

	public int getSumOfSizes(final String format, final String layout, final String interpretation) throws MDStoreServiceException {

		final MongoCollection<DBObject> metadata = getDb().getCollection(METADATA_NAME, DBObject.class);

		BasicDBObject matchObj = (BasicDBObject) BasicDBObjectBuilder.start("$match",

				BasicDBObjectBuilder.start("format", format).add("layout", layout).add("interpretation", interpretation).get()).get();

		BasicDBObject groupObj = (BasicDBObject) BasicDBObjectBuilder.start("$group",

				BasicDBObjectBuilder.start("_id", "").add("total", new BasicDBObject("$sum", "$" + SIZE)).get()).get();

		BasicDBObject projectObj = new BasicDBObject("$project", new BasicDBObject("_id", 0).append("total", 1));

		List<BasicDBObject> pipeline = Lists.newArrayList(matchObj, groupObj, projectObj);

		AggregateIterable<DBObject> output = metadata.aggregate(pipeline, DBObject.class);

		DBObject result = output.first();

		if (result == null || !result.containsField("total")) {

			log.debug("No total found");

			return 0;

		} else return (Integer) result.get("total");

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public void commit(final String transactionId, final String mdId) throws MDStoreServiceException {

		transactionManager.commit(transactionId, mdId, getMDStore(mdId));

	}

	/**

	 * Getter for property 'db'.

	 *

	 * @return Value for property 'db'.

	 */

	public MongoDatabase getDb() {

		return db;

	}

	/**

	 * Setter for property 'db'.

	 *

	 * @param db

	 *            Value to set for property 'db'.

	 */