D-Net

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

			BasicDBObject newObj = (BasicDBObject)((BasicDBObject) currentObj).copy();

			//Must not have _id in the new object

			newObj.remove("_id");

			//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;

	}

}