D-Net

{"type_source": "SVN", "goal": "package -U -T 4C source:jar", "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-oai-store-service/trunk/", "deploy_repository": "dnet4-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/dnet4-snapshots", "name": "dnet-oai-store-service"}

package eu.dnetlib.data.oai.store.mongo;

import static org.junit.Assert.assertEquals;

import static org.junit.Assert.assertFalse;

import static org.junit.Assert.assertNotNull;

import static org.junit.Assert.assertTrue;

import java.util.List;

import javax.annotation.Resource;

import org.junit.After;

import org.junit.Before;

import org.junit.Ignore;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.test.context.ContextConfiguration;

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

import com.mongodb.DB;

import com.mongodb.Mongo;

import eu.dnetlib.data.information.oai.publisher.OaiPublisherException;

import eu.dnetlib.data.oai.store.Cursor;

import eu.dnetlib.data.oai.store.PublisherStore;

import eu.dnetlib.data.oai.store.PublisherStoreDAO;

@RunWith(SpringJUnit4ClassRunner.class)

@ContextConfiguration

@Ignore

public class MongoPublisherStoreDAOTest {

	@Resource

	private PublisherStoreDAO<PublisherStore<Cursor>, Cursor> dao;

	private String storeId;

	@Resource

	private Mongo mongoServer;

	private String dbName = "test";

	@Before

	public void setUp() throws Exception {

		PublisherStore<Cursor> theStore = dao.createStore("oai_dc", "final", "index", dbName);

		this.storeId = theStore.getId();

	}

	@After

	public void tearDown() throws Exception {

		DB db = mongoServer.getDB(dbName);

		db.dropDatabase();

	}

	@Test

	public void testListPublisherStores() throws OaiPublisherException {

		List<PublisherStore<Cursor>> list = dao.listPublisherStores(dbName);

		assertEquals(1, list.size());

		dao.createStore("oai_dc2", "final", "index", dbName);

		List<PublisherStore<Cursor>> list2 = dao.listPublisherStores(dbName);

		assertEquals(2, list2.size());

	}

	@Test

	public void testGetStoreString() {

		PublisherStore<Cursor> theStore = dao.getStore(storeId, dbName);

		assertEquals("oai_dc", theStore.getMetadataFormat());

		assertEquals("final", theStore.getInterpretation());

		assertEquals("index", theStore.getLayout());

	}

	@Test

	public void testGetStoreStringStringString() {

		PublisherStore<Cursor> theStore = dao.getStore("oai_dc", "final", "index", dbName);

		assertEquals(this.storeId, theStore.getId());

	}

	@Ignore

	@Test

	public void testGetStoreFor() {

		PublisherStore<Cursor> theStore = dao.getStoreFor("oai_dc", dbName);

		assertEquals("oai_dc", theStore.getMetadataFormat());

		assertEquals("final", theStore.getInterpretation());

		assertEquals("index", theStore.getLayout());

		assertEquals(this.storeId, theStore.getId());

	}

	@Test

	public void testCreateStore() throws OaiPublisherException {

		PublisherStore<Cursor> store = dao.createStore("foo", "fooInterp", "fooLayout", dbName);

		assertNotNull(store);

		assertTrue(dao.listPublisherStores(dbName).contains(store));

	}

	@Test

	public void testDeleteStoreString() throws OaiPublisherException {

		PublisherStore<Cursor> store = dao.createStore("foo", "fooInterp", "fooLayout", dbName);

		assertNotNull(store);

		String id = store.getId();

		assertTrue(dao.deleteStore(id, dbName));

		assertFalse(dao.deleteStore(id, dbName));

	}

	@Test

	public void testDeleteStoreStringStringString() throws OaiPublisherException {

		PublisherStore<Cursor> store = dao.createStore("foo", "fooInterp", "fooLayout", dbName);

		assertNotNull(store);

		assertTrue(dao.deleteStore("foo", "fooInterp", "fooLayout", dbName));

		assertFalse(dao.deleteStore("foo", "fooInterp", "fooLayout", dbName));

	}

	@Test

	public void testTwoDBs() throws OaiPublisherException {

		String db2 = "test2";

		PublisherStore<Cursor> store = dao.createStore("foo", "fooInterp", "fooLayout", dbName);

		assertTrue(dao.listPublisherStores(dbName).contains(store));

		assertFalse(dao.listPublisherStores(db2).contains(store));

	}

}

package eu.dnetlib.data.oai.store.parser;

import static org.junit.Assert.assertEquals;

import java.util.Date;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.types.ObjectId;

import org.joda.time.format.DateTimeFormat;

import org.joda.time.format.DateTimeFormatter;

import org.joda.time.format.ISODateTimeFormat;

import org.junit.Test;

import com.google.common.collect.Lists;

import com.mongodb.BasicDBObject;

import com.mongodb.DBObject;

public class MongoQueryParserTest {

	private final MongoQueryParser mongoParser = new MongoQueryParser();

	private static final Log log = LogFactory.getLog(MongoQueryParserTest.class); // NOPMD by marko on 11/24/08 5:02 PM

	@Test

	public void testParseEq() {

		DBObject expected = new BasicDBObject("set", "CEDIASManuscripts");

		DBObject o = this.mongoParser.parse("set = \"CEDIASManuscripts\"");

		assertEquals(expected, o);

	}

	@Test

	public void testParseNeq() {

		DBObject expected = new BasicDBObject("set", new BasicDBObject("$ne", "CEDIASManuscripts"));

		DBObject o = this.mongoParser.parse("set <> \"CEDIASManuscripts\"");

		assertEquals(expected, o);

	}

	@Test

	public void testParseAnd() {

		DBObject expected = new BasicDBObject("$and", Lists.newArrayList(new BasicDBObject("set", new BasicDBObject("$ne", "CEDIASManuscripts")),

				new BasicDBObject("pippo", new BasicDBObject("$gt", "x"))));

		DBObject o = this.mongoParser.parse("set <> \"CEDIASManuscripts\" AND pippo > x");

		log.info(o);

		assertEquals(expected, o);

	}

	@Test

	public void testParseOr() {

		DBObject expected = new BasicDBObject("$or", Lists.newArrayList(new BasicDBObject("set", new BasicDBObject("$ne", "CEDIASManuscripts")),

				new BasicDBObject("pippo", new BasicDBObject("$gt", "x"))));

		DBObject o = this.mongoParser.parse("set <> \"CEDIASManuscripts\" OR pippo > x");

		log.info(o);

		assertEquals(expected, o);

	}

	@Test

	public void testParseNot() {

		DBObject expected = new BasicDBObject("$and", Lists.newArrayList(new BasicDBObject("set", "CEDIASManuscripts"), new BasicDBObject("$not",

				new BasicDBObject("pippo", new BasicDBObject("$gt", "x")))));

		DBObject o = this.mongoParser.parse("set = \"CEDIASManuscripts\" NOT pippo > x");

		log.info(o);

		assertEquals(expected, o);

	}

	@Test

	public void testParseStar() {

		DBObject expected = new BasicDBObject();

		DBObject o = this.mongoParser.parse("*");

		DBObject o2 = this.mongoParser.parse("*=*");

		assertEquals(expected, o);

		assertEquals(expected, o2);

	}

	@Test

	public void testParseStarAnd() {

		DBObject expected = new BasicDBObject("$and", Lists.newArrayList(new BasicDBObject(), new BasicDBObject("pippo", new BasicDBObject("$gt", "x"))));

		DBObject o = this.mongoParser.parse("* AND pippo > x");

		DBObject o2 = this.mongoParser.parse("*=* AND pippo > x");

		assertEquals(expected, o);

		assertEquals(expected, o2);

	}

	@Test

	public void testParseIdQuery() {

		DBObject expected = new BasicDBObject("_id", new BasicDBObject("$gt", new ObjectId("5225e093aabf055637bf2c65")));

		DBObject o = this.mongoParser.parse("_id > 5225e093aabf055637bf2c65");

		assertEquals(expected, o);

	}

	@Test

	public void testParseUntilDatestamp() {

		Date dateTime = this.parseDate("2014-04-02T00:00:00.000Z");

		// { "$and" : [ { "oaftype" : { "$ne" : "person"}} , { "datestamp" : { "$lte" : { "$date" : "2014-04-02T00:00:00.000Z"}}}]}

		DBObject expected = new BasicDBObject("$and", Lists.newArrayList(new BasicDBObject("oaftype", new BasicDBObject("$ne", "person")), new BasicDBObject(

				"datestamp", new BasicDBObject("$lte", dateTime))));

		// System.out.println(expected);

		DBObject o = this.mongoParser.parse("(oaftype <> \"person\") AND datestamp <= 2014-04-02");

		assertEquals(expected, o);

		// System.out.println(o);

	}

	private Date parseDate(final String date) {

		DateTimeFormatter dateNoTimeFormatter = DateTimeFormat.forPattern("yyyy-MM-dd").withZoneUTC();

		DateTimeFormatter iso8601NoMsTimeFormatter = DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ssZ").withZoneUTC();

		DateTimeFormatter iso8601Formatter = ISODateTimeFormat.dateTime().withZoneUTC();

		try {

			log.debug("Using default " + iso8601Formatter.getClass());

			return iso8601Formatter.parseDateTime(date).toDate();

		} catch (Exception e) {

			try {

				log.debug("Switching to ISO with no millisecond date formatter: yyyy-MM-dd'T'HH:mm:ssZ");

				return iso8601NoMsTimeFormatter.parseDateTime(date).toDate();

			} catch (Exception ex) {

				log.debug("Switching to simple date formatter: yyyy-MM-dd");

				return dateNoTimeFormatter.parseDateTime(date).toDate();

			}

		}

	}

}

package eu.dnetlib.data.oai.store.sets;

import static org.junit.Assert.assertEquals;

import org.junit.Before;

import org.junit.Test;

public class MongoSetCollectionTest {

	private MongoSetCollection mongoSetCollection;

	private String strangeSet = "Наукові журнали Національного Авіаційного Університету";

	@Before

	public void setup() {

		this.mongoSetCollection = new MongoSetCollection();

	}

	@Test

	public void test() {

		String normalised = this.mongoSetCollection.normalizeSetSpec(strangeSet);

		assertEquals(MongoSetCollection.DEFAULT_SET, normalised);

	}

}

<RESOURCE_PROFILE>

	<HEADER>

		<RESOURCE_IDENTIFIER value="OAIPublisherConfiguration" />

		<RESOURCE_TYPE value="OAIPublisherConfigurationDSResourceType" />

		<RESOURCE_KIND value="OAIPublisherConfigurationDSResources" />

		<RESOURCE_URI value="" />

		<DATE_OF_CREATION value="2001-12-31T12:00:00" />

	</HEADER>

	<BODY>

		    <CONFIGURATION>

            <OAISETS>

                <OAISET enabled="true">

                    <spec>OpenAccess</spec>

                    <name>Set of Open Access articles</name>

                    <description>Set of records having 'OPEN' license</description>

                    <query>(license = "OPEN")</query>

                </OAISET>

                <OAISET enabled="true">

                    <spec>ArticlesInNature</spec>

                    <name>Articles published by Nature</name>

                    <description>Set of articles published by Nature Publishing Group</description>

                    <query>(publisher = "Nature Publishing Group")</query>

                </OAISET>

                <OAISET enabled="true">

                    <spec>publications</spec>

                    <name>Publications</name>

                    <description>Set of all Publications</description>

                    <query>resulttypeid="publication"</query>

                </OAISET>

            </OAISETS>

            <METADATAFORMATS>

                <METADATAFORMAT exportable="true" metadataPrefix="oaf">

                    <NAMESPACE>http://namespace.openaire.eu/oaf</NAMESPACE>

                    <SCHEMA>http://www.openaire.eu/schema/0.1/oaf-0.1.xsd</SCHEMA>

                    <SOURCE_METADATA_FORMAT interpretation="openaire" layout="index" name="oaf"/>

                    <TRANSFORMATION_RULE/>

                    <BASE_QUERY>*</BASE_QUERY>

                </METADATAFORMAT>

                <METADATAFORMAT metadataPrefix="oai_dc" exportable="false">

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

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

                    <SOURCE_METADATA_FORMAT interpretation="openaire" layout="index" name="oaf"/>

                    <TRANSFORMATION_RULE>oaf2dc_VHJhbnNmb3JtYXRpb25SdWxlRFNSZXNvdXJjZXMvVHJhbnNmb3JtYXRpb25SdWxlRFNSZXNvdXJjZVR5cGU=</TRANSFORMATION_RULE>

                    <BASE_QUERY>oaftype="result"</BASE_QUERY>

                </METADATAFORMAT>

            </METADATAFORMATS>

            <INDICES>

                <INDEX name="objIdentifier" repeatable="false">

                    <SOURCE name="oaf" layout="index" interpretation="openaire" path="//*[local-name() ='objIdentifier']"/>

                </INDEX>

                <INDEX name="set" repeatable="true">

                    <SOURCE name="oaf" layout="index" interpretation="openaire" path="//collectedfrom/@name"/>

                </INDEX>

                <INDEX name="publisher" repeatable="true">

                    <SOURCE name="oaf" layout="index" interpretation="openaire" path="//publisher"/>

                </INDEX>

                <INDEX name="license" repeatable="false">

                    <SOURCE name="oaf" layout="index" interpretation="openaire" path="//bestlicense/@classid"/>

                </INDEX>

                <INDEX name="oaftype" repeatable="false">

                    <SOURCE name="oaf" layout="index" interpretation="openaire" path="local-name(//*[local-name()='entity']/*)"/>

                </INDEX>

                <INDEX name="resulttypeid" repeatable="false">

                    <SOURCE name="oaf" layout="index" interpretation="openaire" path="//*[local-name()='entity']/*[local-name()='result']/resulttype/@classid"/>

                </INDEX>

            </INDICES>

        </CONFIGURATION>

		<STATUS>

			<LAST_UPDATE value="2001-12-31T12:00:00" />

		</STATUS>

		<SECURITY_PARAMETERS>SECURITY_PARAMETERS</SECURITY_PARAMETERS>

	</BODY>

</RESOURCE_PROFILE>

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

<beans xmlns="http://www.springframework.org/schema/beans"

	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p"

	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

	<bean id="mongodbPublisherStoreDao" class="eu.dnetlib.data.oai.store.mongo.MongoPublisherStoreDAO"

		p:metadataCollection="metadata" />

	<bean id="publisherMongoServer" class="com.mongodb.Mongo">

		<constructor-arg index="0" type="com.mongodb.ServerAddress">

			<bean class="com.mongodb.ServerAddress">

				<constructor-arg index="0" value="localhost" />

				<constructor-arg index="1" value="27017" />

			</bean>

		</constructor-arg>

	</bean>

	<bean id="mongoQueryParser" class="eu.dnetlib.data.oai.store.parser.MongoQueryParser" />

	<bean id="oaiISLookupClient" class="eu.dnetlib.data.information.oai.publisher.conf.ISLookUpClient" />

	<bean id="oaiConfigurationExistReader" class="eu.dnetlib.data.information.oai.publisher.conf.OAIConfigurationExistReader" p:idScheme="${services.oai.publisher.id.scheme}" p:idNamespace="${services.oai.publisher.id.namespace}" />

	<bean id="oaiConfigurationStringReader" class="eu.dnetlib.data.information.oai.publisher.conf.OAIConfigurationStringReader" p:idScheme="${services.oai.publisher.id.scheme}" p:idNamespace="${services.oai.publisher.id.namespace}" />

	<bean id="configurationParser" class="eu.dnetlib.data.information.oai.publisher.conf.OAIConfigurationParser" />

	<bean id="publisherConfigurationWriter" class="eu.dnetlib.data.information.oai.publisher.conf.OAIConfigurationWriter" />

	<bean id="queryExecutor" class="eu.dnetlib.enabling.tools.SplittedQueryExecutor"

		p:lookupLocator-ref="lookupLocator" />

	<bean id="lookupLocator" class="eu.dnetlib.enabling.tools.StaticServiceLocator">

		<property name="service">

			<bean class="eu.dnetlib.test.utils.MockBeanFactory" p:clazz="eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService" />

		</property>

	</bean>

	<bean id="recordInfoGenerator" class="eu.dnetlib.data.oai.store.mongo.RecordInfoGenerator" />

	<bean id="metadataExtractor" class="eu.dnetlib.data.oai.store.mongo.MetadataExtractor" />

	<bean id="recordChangeDetector" class="eu.dnetlib.data.oai.store.DummyRecordChangeDetector" />

	<bean id="mongoSetCollection" class="eu.dnetlib.data.oai.store.sets.MongoSetCollection" />

	<bean id="oaiSetHelper" class="eu.dnetlib.data.oai.store.conf.OAISetHelper"

		p:configurationReader-ref="oaiConfigurationExistReader"

		p:mongoSetCollection-ref="mongoSetCollection" />

	<bean id="oaiSetsCounter" class="eu.dnetlib.data.oai.store.conf.OAISetsCounter" />

</beans>

org.apache.cxf.Logger=org.apache.cxf.common.logging.Log4jLogger

log4j.rootLogger=WARN, CONSOLE

log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender

log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout

log4j.appender.CONSOLE.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

log4j.logger.eu.dnetlib=INFO

log4j.logger.eu.dnetlib.data=DEBUG

package eu.dnetlib.data.oai.store.mongo;

import java.io.StringReader;

import java.util.Date;

import java.util.List;

import javax.annotation.Resource;

import org.apache.commons.lang.StringEscapeUtils;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.io.SAXReader;

import com.google.common.collect.Sets;

import com.mongodb.DBObject;

import eu.dnetlib.data.information.oai.publisher.OaiPublisherRuntimeException;

import eu.dnetlib.data.information.oai.publisher.conf.OAIConfigurationReader;

import eu.dnetlib.data.information.oai.publisher.info.RecordInfo;

/**

 * Helper class to generate a RecordInfo object from a Mongo DBObject.

 * 

 * @author alessia

 * 

 */

public class RecordInfoGenerator {

	@Resource

	private MetadataExtractor metadataExtractor;

	@SuppressWarnings("unchecked")

	public RecordInfo transformDBObject(final DBObject object, final boolean includeBody) {

		String id = (String) object.get(OAIConfigurationReader.ID_FIELD);

		// need to escape the identifier, otherwise the XML breaks

		id = StringEscapeUtils.escapeXml(id);

		boolean deleted = (Boolean) object.get("deleted");

		RecordInfo record = new RecordInfo();

		record.setIdentifier(id);

		record.setInternalId(object.get("_id").toString());

		record.setDatestamp((Date) object.get(OAIConfigurationReader.DATESTAMP_FIELD));

		record.setDeleted(deleted);

		List<String> sets = (List<String>) object.get(OAIConfigurationReader.SET_FIELD);

		if (sets != null) {

			record.setSetspecs(Sets.newHashSet(sets));

		}

		if (includeBody && !deleted) {

			String body = (String) object.get(OAIConfigurationReader.BODY_FIELD);

			final SAXReader reader = new SAXReader();

			Document doc;

			try {

				doc = reader.read(new StringReader(body));

				record.setMetadata(this.metadataExtractor.evaluate(doc));

			} catch (DocumentException e) {

				throw new OaiPublisherRuntimeException(e);

			}

		}

		return record;

	}

	public MetadataExtractor getMetadataExtractor() {

		return metadataExtractor;

	}

	public void setMetadataExtractor(final MetadataExtractor metadataExtractor) {

		this.metadataExtractor = metadataExtractor;

	}

}

package eu.dnetlib.data.oai.store.mongo;

import java.util.Collection;

import java.util.Date;

import java.util.List;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.base.Function;

import com.google.common.base.Predicate;

import com.google.common.base.Stopwatch;

import com.google.common.collect.Iterables;

import com.google.common.collect.Multimap;

import com.mongodb.BasicDBObject;

import com.mongodb.BasicDBObjectBuilder;

import com.mongodb.DBCollection;

import com.mongodb.DBCursor;

import com.mongodb.DBObject;

import com.mongodb.WriteResult;

import eu.dnetlib.data.information.oai.publisher.PublisherField;

import eu.dnetlib.data.information.oai.publisher.conf.OAIConfigurationReader;

import eu.dnetlib.data.information.oai.publisher.info.RecordInfo;

import eu.dnetlib.data.information.oai.publisher.info.SetInfo;

import eu.dnetlib.data.oai.store.PublisherStore;

import eu.dnetlib.data.oai.store.RecordChangeDetector;

import eu.dnetlib.data.oai.store.parser.MongoQueryParser;

import eu.dnetlib.data.oai.store.parser.PublisherRecordParser;

import eu.dnetlib.data.oai.store.sets.MongoSetCollection;

import eu.dnetlib.miscutils.functional.UnaryFunction;

public class MongoPublisherStore implements PublisherStore<MongoCursor> {

	private static final Log log = LogFactory.getLog(MongoPublisherStore.class); // NOPMD by marko on 11/24/08 5:02 PM

	private String id, metadataFormat, interpretation, layout;

	/** Keeps information about the fields to be created in mongo. **/

	private List<PublisherField> mongoFields;

	private DBCollection collection;

	private DBCollection discardedCollection;

	private RecordInfoGenerator recordInfoGenerator;

	private MetadataExtractor metadataExtractor;

	private MongoQueryParser queryParser;

	private RecordChangeDetector recordChangeDetector;

	private MongoSetCollection mongoSetCollection;

	/**

	 * Used to generate the OAI identifiers compliant to the protocol. See

	 * http://www.openarchives.org/OAI/openarchivesprotocol.html#UniqueIdentifier. See property services.oai.publisher.id.scheme.

	 */

	private String idScheme;

	/**

	 * Used to generate the OAI identifiers compliant to the protocol. See

	 * http://www.openarchives.org/OAI/openarchivesprotocol.html#UniqueIdentifier. See property services.oai.publisher.id.namespace.

	 */

	private String idNamespace;

	@Override

	public RecordInfo getRecord(final String recordId) {

		DBObject query = new BasicDBObject(OAIConfigurationReader.ID_FIELD, recordId);

		DBObject result = this.collection.findOne(query);

		log.debug(result);

		return this.recordInfoGenerator.transformDBObject(result, true);

	}

	@Override

	public RecordInfo getRecord(final String recordId, final UnaryFunction<String, String> unaryFunction) {

		RecordInfo result = this.getRecord(recordId);

		if (result != null) {

			String transformedBody = unaryFunction.evaluate(result.getMetadata());

			result.setMetadata(transformedBody);

		}

		return result;

	}

	@Override

	public MongoCursor getRecords(final String queryString, final boolean bodyIncluded, final int limit) {

		DBCursor cursor = loggedFindByQuery(queryString, limit);

		return new MongoCursor(cursor, bodyIncluded, this.recordInfoGenerator, this.metadataExtractor);

	}

	@Override

	public MongoCursor getRecords(final String queryString, final UnaryFunction<String, String> unaryFunction, final boolean bodyIncluded, final int limit) {

		DBCursor cursor = loggedFindByQuery(queryString, limit);

		return new MongoCursor(cursor, unaryFunction, bodyIncluded, this.recordInfoGenerator, this.metadataExtractor);

	}

	private DBCursor loggedFindByQuery(final String queryString, final int limit) {

		DBObject query = this.queryParser.parse(queryString);

		long start = System.currentTimeMillis();

		DBCursor cursor = this.collection.find(query).sort(new BasicDBObject("_id", 1)).limit(limit);

		long end = System.currentTimeMillis();

		log.debug("Query:" + query + "\ntime to get mongo cursor (ms): " + (end - start));

		return cursor;

	}

	@Override

	public List<PublisherField> getIndices() {

		return this.mongoFields;

	}

	/**

	 * <p>

	 * Ensure indices on the configuration-defined fields and on the system fields DATESTAMP_FIELD and LAST_COLLECTION_DATE_FIELD.

	 * <p>

	 * <p>

	 * Note that by default ID_FIELD, SET_FIELD, DELETED_FIELD, BODY_FIELD, UPDATED_FIELD are not indexed. If you want an index on those,

	 * then you have to specify it in the configuration file of the OAI Publisher: <br>

	 * <INDEX name="deleted">

	 * </p>

	 * 

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.data.information.oai.publisher.store.PublisherStore#ensureIndices()

	 */

	@Override

	public void ensureIndices() {

		List<DBObject> list = this.collection.getIndexInfo();

		this.collection.resetIndexCache();

		Stopwatch sw = new Stopwatch();

		sw.start();

		// I want to keep the composite indexes that might have been defined manually

		log.debug("Ensuring currently defined composite indexes:");

		DBObject indexOptions = new BasicDBObject("background", true);

		for (DBObject o : list) {

			DBObject fieldIndexed = (DBObject) o.get("key");

			if (fieldIndexed.keySet().size() > 1) {

				log.debug(o);

				this.collection.ensureIndex(fieldIndexed, indexOptions);

			}

		}

		// Indexes on single fields.

		for (PublisherField field : this.mongoFields) {

			DBObject mongoIdx = new BasicDBObject(field.getFieldName(), 1);

			log.debug("Creating index : " + mongoIdx);

			this.collection.ensureIndex(mongoIdx, indexOptions);

		}

		log.debug("Creating index over : " + OAIConfigurationReader.DATESTAMP_FIELD);

		this.collection.ensureIndex(new BasicDBObject(OAIConfigurationReader.DATESTAMP_FIELD, 1));

		log.debug("Creating index over : " + OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD);

		this.collection.ensureIndex(new BasicDBObject(OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD, 1));

		sw.stop();

		log.info("All indexes have been updated in " + sw.elapsed(TimeUnit.MILLISECONDS) + " milliseconds");

	}

	/**

	 * Creates a compound index over the specified fields on the given store.

	 * <p>

	 * The creation is performed on the background

	 * </p>

	 * 

	 * @param fieldNames

	 *            List of fields to be included in the compound index

	 * @theStore MongoPublisherStore where to create the index

	 */

	public void createCompoundIndex(final List<String> fieldNames) {

		if ((fieldNames == null) || fieldNames.isEmpty()) {

			log.fatal("No fields specified for the creation of the compound index");

		}

		DBObject indexOptions = new BasicDBObject("background", true);

		BasicDBObjectBuilder theIndexBuilder = BasicDBObjectBuilder.start();

		for (String f : fieldNames) {

			theIndexBuilder.add(f, 1);

		}

		DBObject theIndex = theIndexBuilder.get();

		log.info("Creating index " + theIndex + " on " + this.getId());

		this.getCollection().ensureIndex(theIndex, indexOptions);

	}

	private void dropDiscarded(final String source) {

		if (StringUtils.isBlank(source)) {

			log.debug("Dropping discarded records from publisherStore " + id);

			discardedCollection.drop();

		} else {

			log.debug("Dropping discarded records for source " + source + " from publisherStore " + id);

			discardedCollection.remove(new BasicDBObject(OAIConfigurationReader.SET_FIELD, source));

		}

	}

	@Override

	public int feed(final Iterable<String> records, final String source) {

		final BlockingQueue<Object> queue = new ArrayBlockingQueue<Object>(80);

		final Object sentinel = new Object();

		this.dropDiscarded(source);

		final Date feedDate = new Date();

		Thread background = new Thread(new Runnable() {

			@Override

			public void run() {

				while (true) {

					try {

						Object record = queue.take();

						if (record == sentinel) {

							break;

						}

						safeFeedRecord((String) record, source, feedDate);

					} catch (InterruptedException e) {

						log.fatal("got exception in background thread", e);

						throw new IllegalStateException(e);

					}

				}

			}

		});

		background.start();

		long startFeed = feedDate.getTime();

		try {

			log.info("feeding publisherStore " + id);

			for (final String record : records) {

				queue.put(record);

			}

			queue.put(sentinel);

			log.info("finished feeding publisherStore " + id);

			background.join();

		} catch (InterruptedException e) {

			throw new IllegalStateException(e);

		}

		long endFeed = System.currentTimeMillis();

		log.fatal("OAI STORE " + id + " FEEDING COMPLETED IN " + (endFeed - startFeed) + "ms");

		this.setDeletedFlags(feedDate, source);

		return this.count();

	}

	/**

	 * Launches the thread that flags the records to be considered as 'deleted'.

	 * <p>

	 * The datestamp of the deleted records must be updated as well, according to the OAI specs available at

	 * http://www.openarchives.org/OAI/openarchivesprotocol.html#DeletedRecords: if a repository does keep track of deletions then the

	 * datestamp of the deleted record must be the date and time that it was deleted.

	 * </p>

	 * 

	 * @param feedDate

	 * @param source

	 */

	private void setDeletedFlags(final Date feedDate, final String source) {

		Thread deletedSetter = new Thread(new Runnable() {

			@Override

			public void run() {

				DBObject query = BasicDBObjectBuilder.start(OAIConfigurationReader.DELETED_FIELD, false)

						.append(OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD, new BasicDBObject("$lt", feedDate)).get();

				if (!StringUtils.isBlank(source)) {

					query.put(OAIConfigurationReader.SET_FIELD, source);

				}

				log.debug("Delete flag query: " + query.toString());

				DBObject update = new BasicDBObject("$set", BasicDBObjectBuilder.start(OAIConfigurationReader.DELETED_FIELD, true)

						.append(OAIConfigurationReader.DATESTAMP_FIELD, feedDate).append(OAIConfigurationReader.UPDATED_FIELD, true).get());

				log.debug("Updating as: " + update.toString());

				WriteResult wr = collection.update(query, update, false, true);

				log.debug("Deleted flags set for source: " + source + " #records = " + wr.getN());

			}

		});

		deletedSetter.start();

		try {

			deletedSetter.join();

		} catch (InterruptedException e) {

			throw new IllegalStateException(e);

		}

	}

	@Override

	public void drop() {

		this.collection.drop();

	}

	@Override

	public void drop(final String queryString) {

		DBObject query = this.queryParser.parse(queryString);

		this.collection.remove(query);

	}

	@Override

	public int count() {

		return (int) this.collection.count();

	}

	@Override

	public int count(final String queryString) {

		if (StringUtils.isBlank(queryString)) return (int) this.collection.count();

		DBObject query = this.queryParser.parse(queryString);

		return (int) this.collection.count(query);

	}

	public List<String> getDistinctSetNamesFromRecords() {

		log.info("Going to ask for all distinct sets in the oaistore " + id + ": this may take a long time...");

		@SuppressWarnings("unchecked")

		List<String> distinctSets = this.collection.distinct(OAIConfigurationReader.SET_FIELD);

		return distinctSets;

	}

	// ***********************************************************************************************//

	// Feed utilities

	// ***********************************************************************************************//

	private boolean safeFeedRecord(final String record, final String source, final Date feedDate) {

		try {

			if (!record.isEmpty()) return feedRecord(record, source, feedDate);

		} catch (final Throwable e) {

			log.error("Got unhandled exception while parsing record", e);

			discardedCollection.insert(new BasicDBObject(OAIConfigurationReader.SET_FIELD, source).append(OAIConfigurationReader.BODY_FIELD, record));

		}

		return false;

	}

	/**

	 * Feed the record to the store.

	 * 

	 * @return true if the record is new, false otherwise

	 */

	private boolean feedRecord(final String record, final String source, final Date feedDate) {

		PublisherRecordParser parser = new PublisherRecordParser(this.mongoFields);

		final Multimap<String, String> recordProperties = parser.parseRecord(record);

		String id = "";

		String oaiID = "";

		if (recordProperties.containsKey(OAIConfigurationReader.ID_FIELD)) {

			id = recordProperties.get(OAIConfigurationReader.ID_FIELD).iterator().next();

			oaiID = getOAIIdentifier(id);

			if (isNewRecord(oaiID)) {

				feedNew(oaiID, record, recordProperties, feedDate);

				return true;

			} else {

				if (isChanged(oaiID, record)) {

					updateRecord(oaiID, record, recordProperties, feedDate);

				} else {

					// it is not changed, I only have to update the last collection date

					handleRecord(oaiID, feedDate);

				}

			}

		} else {

			log.error("parsed record seems invalid -- no identifier property with name: " + OAIConfigurationReader.ID_FIELD);

			discardedCollection.insert(new BasicDBObject(OAIConfigurationReader.SET_FIELD, source).append(OAIConfigurationReader.BODY_FIELD, record).append(

					OAIConfigurationReader.DATESTAMP_FIELD, feedDate));

		}

		return false;

	}

	private DBObject createBasicObject(final String oaiID, final String record, final Multimap<String, String> recordProperties) {

		DBObject obj = new BasicDBObject();

		for (final String key : recordProperties.keySet()) {

			if (key.equals(OAIConfigurationReader.ID_FIELD)) {

				obj.put(key, oaiID);

			} else {

				Collection<String> values = recordProperties.get(key);

				if (key.equals(OAIConfigurationReader.SET_FIELD)) {

					Iterable<String> setSpecs = Iterables.transform(values, new Function<String, String>() {

						@Override

						public String apply(final String s) {

							return mongoSetCollection.normalizeSetSpec(s);

						}

					});

					obj.put(key, setSpecs);

				} else {

					// let's check if the key is the name of a repeatable field or not

					PublisherField keyField = Iterables.find(this.mongoFields, new Predicate<PublisherField>() {

						@Override

						public boolean apply(final PublisherField field) {

							return field.getFieldName().equals(key);

						}

					}, null);

					if (keyField == null) {

						log.warn("Expected field to index: " + key + " could not be found, but we keep going...");

					}

					if ((keyField != null) && !keyField.isRepeatable()) {

						if ((values != null) && !values.isEmpty()) {

							obj.put(key, values.iterator().next());

						}

					} else {

						obj.put(key, values);

					}

				}

			}

		}

		obj.put(OAIConfigurationReader.BODY_FIELD, record);

		obj.put(OAIConfigurationReader.DELETED_FIELD, false);

		return obj;

	}

	private void feedNew(final String oaiID, final String record, final Multimap<String, String> recordProperties, final Date feedDate) {

		log.debug("New record received. Assigned oai id: " + oaiID);

		DBObject obj = this.createBasicObject(oaiID, record, recordProperties);

		obj.put(OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD, feedDate);

		obj.put(OAIConfigurationReader.DATESTAMP_FIELD, feedDate);

		obj.put(OAIConfigurationReader.UPDATED_FIELD, false);

		collection.insert(obj);

		this.upsertSets(recordProperties.get(OAIConfigurationReader.SET_FIELD));

	}

	private void updateRecord(final String oaiID, final String record, final Multimap<String, String> recordProperties, final Date feedDate) {

		log.debug("updating record " + oaiID);

		DBObject obj = this.createBasicObject(oaiID, record, recordProperties);

		obj.put(OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD, feedDate);

		obj.put(OAIConfigurationReader.DATESTAMP_FIELD, feedDate);

		obj.put(OAIConfigurationReader.UPDATED_FIELD, true);

		DBObject oldObj = new BasicDBObject(OAIConfigurationReader.ID_FIELD, oaiID);

		collection.update(oldObj, obj, true, false);

		this.upsertSets(recordProperties.get(OAIConfigurationReader.SET_FIELD));

	}

	public void upsertSets(final Iterable<String> setNames) {

		// feed the list of sets, if any

		if (setNames != null) {

			for (String setName : setNames) {

				if (StringUtils.isNotBlank(setName)) {

					final SetInfo set = new SetInfo();

					String setSpec = this.mongoSetCollection.normalizeSetSpec(setName);

					set.setSetSpec(setSpec);

					set.setSetName(setName);

					set.setSetDescription("This set contains metadata records whose provenance is " + setName);

					set.setEnabled(true);

					String query = "(" + OAIConfigurationReader.SET_FIELD + " = \"" + setSpec + "\") ";

					set.setQuery(query);

					this.mongoSetCollection.upsertSet(set, false, getCollection().getDB().getName());

				}

			}

		}

	}

	private void handleRecord(final String oaiID, final Date lastCollectionDate) {

		log.debug("handling unchanged record " + oaiID);

		DBObject oldObj = new BasicDBObject(OAIConfigurationReader.ID_FIELD, oaiID);

		DBObject update = new BasicDBObject("$set", new BasicDBObject(OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD, lastCollectionDate));

		collection.update(oldObj, update, true, false);

	}

	private boolean isNewRecord(final String oaiIdentifier) {

		return this.collection.findOne(new BasicDBObject(OAIConfigurationReader.ID_FIELD, oaiIdentifier)) == null;

	}

	private boolean isChanged(final String oaiID, final String record) {

		RecordInfo oldRecord = getRecord(oaiID);

		return this.recordChangeDetector.differs(oldRecord.getMetadata(), record);

	}

	private String getOAIIdentifier(final String id) {

		return this.idScheme + ":" + this.idNamespace + ":" + id;

	}

	// ***********************************************************************************************//

	// Setters / Getters / Basic utilities

	// ***********************************************************************************************//

	@Override

	public int hashCode() {

		final int prime = 31;

		int result = 1;

		result = (prime * result) + ((collection == null) ? 0 : collection.hashCode());

		result = (prime * result) + ((id == null) ? 0 : id.hashCode());

		result = (prime * result) + ((interpretation == null) ? 0 : interpretation.hashCode());

		result = (prime * result) + ((layout == null) ? 0 : layout.hashCode());

		result = (prime * result) + ((metadataFormat == null) ? 0 : metadataFormat.hashCode());

		return result;

	}

	@Override

	public boolean equals(final Object obj) {

		if (this == obj) return true;

		if (obj == null) return false;

		if (!(obj instanceof MongoPublisherStore)) return false;

		MongoPublisherStore other = (MongoPublisherStore) obj;

		if (collection == null) {

			if (other.collection != null) return false;

		} else if (!collection.equals(other.collection)) return false;

		if (id == null) {

			if (other.id != null) return false;

		} else if (!id.equals(other.id)) return false;

		if (interpretation == null) {

			if (other.interpretation != null) return false;

		} else if (!interpretation.equals(other.interpretation)) return false;

		if (layout == null) {

			if (other.layout != null) return false;

		} else if (!layout.equals(other.layout)) return false;

		if (metadataFormat == null) {

			if (other.metadataFormat != null) return false;

		} else if (!metadataFormat.equals(other.metadataFormat)) return false;

		return true;

	}

	public MongoPublisherStore() {

		super();

	}

	public MongoPublisherStore(final String id, final String metadataFormat, final String interpretation, final String layout, final DBCollection collection,

			final List<PublisherField> mongoFields, final MongoQueryParser queryParser, final RecordInfoGenerator recordInfoGenerator, final String idScheme,

			final String idNamespace, final MetadataExtractor metadataExtractor, final RecordChangeDetector recordChangeDetector) {

		super();

		this.id = id;

		this.metadataFormat = metadataFormat;

		this.interpretation = interpretation;

		this.layout = layout;

		this.collection = collection;

		this.discardedCollection = collection.getDB().getCollection("discarded-" + collection.getName());

		this.mongoFields = mongoFields;

		this.queryParser = queryParser;

		this.recordInfoGenerator = recordInfoGenerator;

		this.idScheme = idScheme;

		this.idNamespace = idNamespace;

		this.recordChangeDetector = recordChangeDetector;

	}

	public void setId(final String id) {

		this.id = id;

	}

	public void setMetadataFormat(final String metadataFormat) {

		this.metadataFormat = metadataFormat;

	}

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

	public void setLayout(final String layout) {

		this.layout = layout;

	}

	public DBCollection getCollection() {

		return collection;

	}

	public void setCollection(final DBCollection collection) {

		this.collection = collection;

	}

	public MongoQueryParser getQueryParser() {

		return queryParser;

	}

	public void setQueryParser(final MongoQueryParser queryParser) {

		this.queryParser = queryParser;

	}

	public DBCollection getDiscardedCollection() {

		return discardedCollection;

	}

	public void setDiscardedCollection(final DBCollection discardedCollection) {

		this.discardedCollection = discardedCollection;

	}

	public String getIdScheme() {

		return idScheme;

	}

	public void setIdScheme(final String idScheme) {

		this.idScheme = idScheme;

	}

	public String getIdNamespace() {

		return idNamespace;

	}

	public void setIdNamespace(final String idNamespace) {

		this.idNamespace = idNamespace;

	}

	public RecordInfoGenerator getRecordInfoGenerator() {

		return recordInfoGenerator;

	}

	public void setRecordInfoGenerator(final RecordInfoGenerator recordInfoGenerator) {

		this.recordInfoGenerator = recordInfoGenerator;

	}

	public MetadataExtractor getMetadataExtractor() {

		return metadataExtractor;

	}

	public void setMetadataExtractor(final MetadataExtractor metadataExtractor) {

		this.metadataExtractor = metadataExtractor;

	}

	public RecordChangeDetector getRecordChangeDetector() {

		return recordChangeDetector;

	}

	public void setRecordChangeDetector(final RecordChangeDetector recordChangeDetector) {

		this.recordChangeDetector = recordChangeDetector;

	}

	@Override

	public String getId() {

		return this.id;

	}

	@Override

	public String getMetadataFormat() {

		return this.metadataFormat;

	}

	@Override

	public String getInterpretation() {

		return this.interpretation;

	}

	@Override

	public String getLayout() {

		return this.layout;

	}

	public MongoSetCollection getMongoSetCollection() {

		return mongoSetCollection;

	}

	public void setMongoSetCollection(final MongoSetCollection mongoSetCollection) {

		this.mongoSetCollection = mongoSetCollection;

	}

	public List<PublisherField> getMongoFields() {

		return mongoFields;

	}

	public void setMongoFields(final List<PublisherField> mongoFields) {

		this.mongoFields = mongoFields;

	}

}

package eu.dnetlib.data.oai.store.mongo;

import org.dom4j.Document;

import org.dom4j.Node;

import eu.dnetlib.miscutils.functional.UnaryFunction;

/**

 * Function to skip the header of the record and deliver only its metadata content as XML String.

 */

public class MetadataExtractor implements UnaryFunction<String, Document> {

	@Override

	public String evaluate(final Document xmlDoc) {

		Node metadataNode = xmlDoc.selectSingleNode("//*[local-name() = 'metadata']/*");

		return metadataNode.asXML();

	}

}

package eu.dnetlib.data.oai.store.mongo;

import java.util.Iterator;

import com.mongodb.DBCursor;

import com.mongodb.DBObject;

import eu.dnetlib.data.information.oai.publisher.info.RecordInfo;

import eu.dnetlib.data.oai.store.Cursor;

import eu.dnetlib.miscutils.functional.UnaryFunction;

public class MongoCursor implements Cursor {

	/**

	 * Underlying mongo cursor.

	 */

	private DBCursor dbCursor;

	/**

	 * Function to apply to records before delivering.

	 */

	private UnaryFunction<String, String> function;

	/**

	 * true if the RecordInfo returned by this Cursor must include the record body, false otherwise.

	 */

	private boolean bodyIncluded;

	private RecordInfoGenerator recordInfoGenerator;

	private MetadataExtractor metadataExtractor;

	/**

	 * 

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.data.information.oai.publisher.store.Cursor#count()

	 */

	@Override

	public int count() {

		return this.dbCursor.count();

	}

	/**

	 * 

	 * {@inheritDoc}

	 * 

	 * @see java.lang.Iterable#iterator()

	 */

	@Override

	public Iterator<RecordInfo> iterator() {

		return new Iterator<RecordInfo>() {

			@Override

			public boolean hasNext() {

				return dbCursor.hasNext();

			}

			@Override

			public RecordInfo next() {

				DBObject res = dbCursor.next();

				RecordInfo info = recordInfoGenerator.transformDBObject(res, bodyIncluded);

				if ((function != null) && bodyIncluded) {

					info.setMetadata(function.evaluate(info.getMetadata()));

				}

				return info;

			}

			@Override

			public void remove() {

				throw new UnsupportedOperationException();

			}

		};

	}

	public MongoCursor() {

		super();

	}

	public MongoCursor(final DBCursor dbCursor, final boolean bodyIncluded, final RecordInfoGenerator recordInfoGenerator,

			final MetadataExtractor metadataExtractor) {

		this(dbCursor, null, bodyIncluded, recordInfoGenerator, metadataExtractor);

	}

	public MongoCursor(final DBCursor dbCursor, final UnaryFunction<String, String> function, final boolean bodyIncluded,

			final RecordInfoGenerator recordInfoGenerator, final MetadataExtractor metadataExtractor) {

		super();

		this.dbCursor = dbCursor;

		this.function = function;

		this.bodyIncluded = bodyIncluded;

		this.recordInfoGenerator = recordInfoGenerator;

		this.metadataExtractor = metadataExtractor;

	}

	public UnaryFunction<String, String> getFunction() {

		return function;

	}

	public void setFunction(final UnaryFunction<String, String> function) {

		this.function = function;

	}

	public DBCursor getDbCursor() {

		return dbCursor;

	}

	public void setDbCursor(final DBCursor dbCursor) {

		this.dbCursor = dbCursor;

	}

	@Override

	public boolean isBodyIncluded() {

		return this.bodyIncluded;

	}

	@Override

	public void setBodyIncluded(final boolean bodyIncluded) {

		this.bodyIncluded = bodyIncluded;

	}

	public RecordInfoGenerator getRecordInfoGenerator() {

		return recordInfoGenerator;

	}

	public void setRecordInfoGenerator(final RecordInfoGenerator recordInfoGenerator) {

		this.recordInfoGenerator = recordInfoGenerator;

	}

	public MetadataExtractor getMetadataExtractor() {

		return metadataExtractor;

	}

	public void setMetadataExtractor(final MetadataExtractor metadataExtractor) {

		this.metadataExtractor = metadataExtractor;

	}

}

package eu.dnetlib.data.oai.store.mongo;

import java.util.List;

import javax.annotation.Resource;

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

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

import com.google.common.collect.Lists;

import com.mongodb.BasicDBObject;

import com.mongodb.BasicDBObjectBuilder;

import com.mongodb.DB;

import com.mongodb.DBCollection;

import com.mongodb.DBCursor;

import com.mongodb.DBObject;

import com.mongodb.Mongo;

import eu.dnetlib.data.information.oai.publisher.OaiPublisherException;

import eu.dnetlib.data.information.oai.publisher.OaiPublisherRuntimeException;

import eu.dnetlib.data.information.oai.publisher.conf.OAIConfigurationReader;

import eu.dnetlib.data.information.oai.publisher.info.MDFInfo;

import eu.dnetlib.data.oai.store.PublisherStoreDAO;

import eu.dnetlib.data.oai.store.RecordChangeDetector;

import eu.dnetlib.data.oai.store.parser.MongoQueryParser;

import eu.dnetlib.data.oai.store.sets.MongoSetCollection;

public class MongoPublisherStoreDAO implements PublisherStoreDAO<MongoPublisherStore, MongoCursor> {

	@Autowired

	private Mongo publisherMongoServer;

	/** Name of the collection with information about the OAI stores. **/

	private String metadataCollection;

	@Resource

	private RecordInfoGenerator recordInfoGenerator;

	@Resource

	private MetadataExtractor metadataExtractor;

	@Resource

	private MongoQueryParser queryParser;

	@Resource(name = "oaiConfigurationExistReader")

	private OAIConfigurationReader configuration;

	@Resource

	private RecordChangeDetector recordChangeDetector;

	@Resource

	private MongoSetCollection mongoSetCollection;

	protected DB getDB(final String dbName) {

		return this.publisherMongoServer.getDB(dbName);

	}

	@Override

	public List<MongoPublisherStore> listPublisherStores(final String dbName) {

		List<MongoPublisherStore> stores = Lists.newArrayList();

		DB db = getDB(dbName);

		DBCursor cursor = db.getCollection(this.metadataCollection).find();

		for (DBObject storeInfo : cursor) {

			stores.add(this.createFromDBObject(storeInfo, db));

		}

		return stores;

	}

	@Override

	public MongoPublisherStore getStore(final String storeId, final String dbName) {

		DBObject storeInfo = getDB(dbName).getCollection(this.metadataCollection).findOne(new BasicDBObject("id", storeId));

		return this.createFromDBObject(storeInfo, getDB(dbName));

	}

	@Override

	public MongoPublisherStore getStore(final String mdfName, final String mdfInterpretation, final String mdfLayout, final String dbName) {

		return this.getStore(this.generateStoreId(mdfName, mdfInterpretation, mdfLayout), dbName);

	}

	@Override

	public MongoPublisherStore getStoreFor(final String targetMetadataPrefix, final String dbName) {

		MDFInfo info = this.configuration.getMetadataFormatInfo(targetMetadataPrefix);

		return this.getStore(info.getSourceFormatName(), info.getSourceFormatInterpretation(), info.getSourceFormatLayout(), dbName);

	}

	@Override

	public MongoPublisherStore createStore(final String mdfName, final String mdfInterpretation, final String mdfLayout, final String dbName)

			throws OaiPublisherException {

		DB db = getDB(dbName);

		DBObject store = createMetadataEntry(mdfName, mdfInterpretation, mdfLayout);

		DBCollection metadata = db.getCollection(this.metadataCollection);

		metadata.insert(store);

		MongoPublisherStore theStore = this.createFromDBObject(store, db);

		return theStore;

	}

	@Override

	public boolean deleteStore(final String storeId, final String dbName) {

		DB db = getDB(dbName);

		DBCollection metadata = db.getCollection(this.metadataCollection);

		DBObject storeInfo = metadata.findOne(new BasicDBObject("id", storeId));

		if (storeInfo == null) return false;

		else {

			db.getCollection(storeId).drop();

			metadata.remove(storeInfo);

			// TODO: should drop entries related to mdPrefix served by the store we are deleting, not all of them.

			this.mongoSetCollection.dropOAISets(dbName);

			return true;

		}

	}

	@Override

	public boolean deleteStore(final String mdfName, final String mdfInterpretation, final String mdfLayout, final String dbName) {

		return this.deleteStore(this.generateStoreId(mdfName, mdfInterpretation, mdfLayout), dbName);

	}

	public void ensureIndex(final MongoPublisherStore store) {

		if (store == null) throw new OaiPublisherRuntimeException("Can't ensure index on null store");

		Thread t = new Thread() {

			@Override

			public void run() {

				store.ensureIndices();

			}

		};

		t.start();

	}

	public void ensureIndex(final String dbName) {

		List<MongoPublisherStore> stores = this.listPublisherStores(dbName);

		for (final MongoPublisherStore s : stores) {

			s.ensureIndices();

		}

	}

	private MongoPublisherStore createFromDBObject(final DBObject storeInfo, final DB db) {

		if (storeInfo == null) return null;

		String storeId = (String) storeInfo.get("id");

		String mdFormat = (String) storeInfo.get("metadataFormat");

		String mdInterpreation = (String) storeInfo.get("interpretation");

		String mdLayout = (String) storeInfo.get("layout");

		MongoPublisherStore store = new MongoPublisherStore(storeId, mdFormat, mdInterpreation, mdLayout, db.getCollection(storeId),

				this.configuration.getFields(mdFormat, mdInterpreation, mdLayout), queryParser, recordInfoGenerator, this.configuration.getIdScheme(),

				this.configuration.getIdNamespace(), this.metadataExtractor, this.recordChangeDetector);

		store.setMongoSetCollection(mongoSetCollection);

		return store;

	}

	private DBObject createMetadataEntry(final String mdfName, final String mdfInterpretation, final String mdfLayout) {

		DBObject info = BasicDBObjectBuilder.start("id", generateStoreId(mdfName, mdfInterpretation, mdfLayout)).append("metadataFormat", mdfName)

				.append("interpretation", mdfInterpretation).append("layout", mdfLayout).get();

		return info;

	}

	private String generateStoreId(final String mdfName, final String mdfInterpretation, final String mdfLayout) {

		return mdfName + "-" + mdfLayout + "-" + mdfInterpretation;

	}

	public String getMetadataCollection() {

		return metadataCollection;

	}

	@Required

	public void setMetadataCollection(final String metadataCollection) {

		this.metadataCollection = metadataCollection;

	}

	public MongoQueryParser getQueryParser() {

		return queryParser;

	}

	public void setQueryParser(final MongoQueryParser queryParser) {

		this.queryParser = queryParser;

	}

	public OAIConfigurationReader getConfiguration() {

		return configuration;

	}

	public void setConfiguration(final OAIConfigurationReader configuration) {

		this.configuration = configuration;

	}

	public RecordInfoGenerator getRecordInfoGenerator() {

		return recordInfoGenerator;

	}

	public void setRecordInfoGenerator(final RecordInfoGenerator recordInfoGenerator) {

		this.recordInfoGenerator = recordInfoGenerator;