D-Net

package eu.dnetlib.data.oai.store;

/**

 * This is a basic dummy implementation that always returns true, regardless the actual content of the records to compare.

 * 

 * @author alessia

 * 

 */

public class DummyRecordChangeDetector implements RecordChangeDetector {

	/**

	 * 

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.data.oai.store.RecordChangeDetector#differs(java.lang.String, java.lang.String)

	 */

	@Override

	public boolean differs(final String record1, final String record2) {

		return true;

	}

}

package eu.dnetlib.data.oai.store;

public interface RecordChangeDetector {

	/**

	 * Checks if the two records have differences based on logics that vary on the actual implementor class.

	 * 

	 * @param record1

	 *            first record to compare

	 * @param record2

	 *            second record to compare

	 * @return true if the two records differ based on the implementor's logics

	 */

	boolean differs(final String record1, final String record2);

}

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

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

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

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

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

		p:metadataCollection="metadata" />

	<bean id="publisherMongoClient" class="com.mongodb.MongoClient">

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

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

				<constructor-arg index="0"

					value="${services.publisher.oai.host}" />

				<constructor-arg index="1"

					value="${services.publisher.oai.port}" />

			</bean>

		</constructor-arg>

	</bean>

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

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

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

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

	<bean id="provenanceExtractor" class="eu.dnetlib.data.oai.store.mongo.ProvenanceExtractor" />

	<bean id="theOAICacheManager" class="org.springframework.cache.ehcache.EhCacheCacheManager"

        p:cache-manager-ref="oaiStoreCacheManager" />

	<bean id="oaiStoreCacheManager"

		class="org.springframework.cache.ehcache.EhCacheManagerFactoryBean"

		p:cacheManagerName="oaiStoreCacheManger"

		p:configLocation="classpath:/eu/dnetlib/data/oai/store/mongo/cache/ehcache.xml" />

</beans>

services.publisher.oai.host=localhost

services.publisher.oai.port=27017

#services.publisher.oai.db=oaistore

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

import java.io.IOException;

import java.util.zip.ZipEntry;

import java.util.zip.ZipOutputStream;

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

import org.apache.commons.io.IOUtils;

import org.apache.commons.io.output.ByteArrayOutputStream;

import org.bson.types.Binary;

import org.junit.Before;

import org.junit.Test;

import org.springframework.core.io.ClassPathResource;

/**

 * RecordInfoGenerator Tester.

 *

 * @author <Authors name>

 * @version 1.0

 * @since <pre>Apr 6, 2016</pre>

 */

public class RecordInfoGeneratorTest {

	private RecordInfoGenerator gen = new RecordInfoGenerator();

	private String filePath = "/eu/dnetlib/data/oai/store/mongo/testRecord.xml";

	private Binary binaryXML;

	@Before

	public void before() throws Exception {

		String testRecord = IOUtils.toString(new ClassPathResource(filePath).getInputStream());

		binaryXML = createCompressRecord(testRecord);

	}

	/**

	 * Method: decompressRecord(final byte[] input)

	 */

	@Test

	public void testDecompressRecord() throws Exception {

		long timeStart = System.currentTimeMillis();

		String record = gen.decompressRecord(binaryXML.getData());

		long timeEnd = System.currentTimeMillis();

		System.out.println("Decompressed record in ms " + (timeEnd - timeStart));

		System.out.println(record);

	}

	private Binary createCompressRecord(final String record) throws IOException {

		ByteArrayOutputStream os = new ByteArrayOutputStream();

		ZipOutputStream zos = new ZipOutputStream(os);

		ZipEntry entry = new ZipEntry(OAIConfigurationReader.BODY_FIELD);

		zos.putNextEntry(entry);

		zos.write(record.getBytes());

		zos.closeEntry();

		//zos.flush();

		zos.close();

		return new Binary(os.toByteArray());

	}

}

<?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:jaxws="http://cxf.apache.org/jaxws"

	xmlns:sec="http://cxf.apache.org/configuration/security" xmlns:wsa="http://cxf.apache.org/ws/addressing"

	xmlns:p="http://www.springframework.org/schema/p" xmlns:http="http://cxf.apache.org/transports/http/configuration"

	xmlns:t="http://dnetlib.eu/springbeans/t" xmlns:template="http://dnetlib.eu/springbeans/template"

	xmlns:util="http://www.springframework.org/schema/util"

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

    	http://cxf.apache.org/ws/addressing http://cxf.apache.org/schemas/ws-addr-conf.xsd

        http://cxf.apache.org/configuration/security http://cxf.apache.org/schemas/configuration/security.xsd

        http://cxf.apache.org/transports/http/configuration http://cxf.apache.org/schemas/configuration/http-conf.xsd

        http://cxf.apache.org/jaxws http://cxf.apache.org/schemas/jaxws.xsd

        http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd

        http://dnetlib.eu/springbeans/template http://dnetlib.eu/springbeans/template.xsd">

	<!-- endpoints -->

	<jaxws:endpoint id="oaiStoreServiceEndpoint"

		implementor="#oaiStoreService" implementorClass="eu.dnetlib.data.oai.store.OAIStoreService"

		address="/oaistore" />

	<template:instance name="serviceRegistrationManager"

		t:serviceRegistrationManagerClass="eu.dnetlib.enabling.tools.registration.ValidatingServiceRegistrationManagerImpl"

		t:name="oaiStoreServiceRegistrationManager" t:service="oaiStoreService"

		t:endpoint="oaiStoreServiceEndpoint" t:jobScheduler="jobScheduler"

		t:serviceRegistrator="blackboardServiceRegistrator" />

	<bean id="oaiStoreService" class="eu.dnetlib.data.oai.store.OAIStoreServiceImpl"

		init-method="start" destroy-method="stop" p:notificationHandler-ref="oaiStoreNotificationHandler" />

	<bean id="oaiStoreNotificationHandler"

		class="eu.dnetlib.enabling.tools.blackboard.BlackboardServerExecutorNotificationHandler"

		p:blackboardExecutor-ref="oaiStoreBlackboardExecutor" />

	<bean id="oaiStoreBlackboardExecutor"

		class="eu.dnetlib.enabling.tools.blackboard.BlackboardServerActionExecutor"

		p:blackboardHandler-ref="blackboardHandler"

		p:actionType="eu.dnetlib.data.oai.store.actions.OAIStoreActions"

		p:incomplete="false">

		<property name="actionMap">

			<map>

				<entry key="SYNC">

					<bean class="eu.dnetlib.data.oai.store.actions.SyncAction" />

				</entry>

				<entry key="COUNT_SETS">

					<bean class="eu.dnetlib.data.oai.store.actions.CountSetsAction"/>

				</entry>

				<entry key="REFRESH_CONFIG">

					<bean class="eu.dnetlib.data.oai.store.actions.RefreshConfigAction" />

				</entry>

				<entry key="ENSURE_INDEXES">

					<bean class="eu.dnetlib.data.oai.store.actions.EnsureIndexesAction" />

				</entry>

				<entry key="CREATE_STORE">

					<bean class="eu.dnetlib.data.oai.store.actions.CreateStoreAction" />

				</entry>

				<entry key="CREATE_OAI_INDEX">

					<bean class="eu.dnetlib.data.oai.store.actions.CreateOAIIndexAction" />

				</entry>

				<entry key="DROP_STORE">

					<bean class="eu.dnetlib.data.oai.store.actions.DropStoreAction" />

				</entry>

			</map>

		</property>

	</bean>

	<bean id="synchronizer" class="eu.dnetlib.data.oai.store.sync.OAIStoreSynchronizer" />

	<bean id="oaiHelper" class="eu.dnetlib.data.oai.store.conf.OAIHelper"/>

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

	<bean id="mongoCacheHelper" class="eu.dnetlib.data.oai.store.mongo.MongoPublisherCacheHelper"/>

</beans>

<?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>

		<relativePath />

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

	<artifactId>dnet-oai-store-service</artifactId>

	<version>8.0.2</version>

	<packaging>jar</packaging>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-oai-store-service/tags/dnet-oai-store-service-8.0.2</developerConnection>

	</scm>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-blackboard-common</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-cql-utils</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-oai-store-service-rmi</artifactId>

			<version>[4.0.0,5.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-resultset-client</artifactId>

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

		</dependency>

		<dependency>

			<groupId>org.mongodb</groupId>

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

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

		</dependency>

		<dependency>

			<groupId>joda-time</groupId>

			<artifactId>joda-time</artifactId>

			<version>[2.3,3.0)</version>

		</dependency>

		<dependency>

			<groupId>org.z3950.zing</groupId>

			<artifactId>cql-java</artifactId>

			<version>[1.7]</version>

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>${junit.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>

	</dependencies>

</project>

/**

 *

 */

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

import java.io.IOException;

import org.bson.types.Binary;

import org.junit.Assert;

import org.junit.Test;

/**

 * @author sandro

 *

 */

public class MongoPublisherStoreTest {

	/**

	 * Test method for {@link eu.dnetlib.data.oai.store.mongo.MongoPublisherStore#createCompressRecord(java.lang.String)}.

	 * @throws IOException

	 */

	@Test

	public void testCreateCompressRecord() throws IOException {

		MongoPublisherStore store = new MongoPublisherStore();

		RecordInfoGenerator gen = new RecordInfoGenerator();

		StringBuffer buffer = new StringBuffer();

		String  input =" CIAO MONDO DA SANDRO!";

		for (int i=0; i< 10000; i++) {

			buffer.append(input);

		}

		Binary data = store.createCompressRecord(buffer.toString());

		Assert.assertEquals(buffer.toString(),gen.decompressRecord(data.getData()));

	}

}

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

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

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

import java.io.ByteArrayInputStream;

import java.io.IOException;

import java.io.StringReader;

import java.time.LocalDateTime;

import java.time.ZoneId;

import java.time.format.DateTimeFormatter;

import java.util.Date;

import java.util.List;

import java.util.zip.ZipEntry;

import java.util.zip.ZipInputStream;

import javax.annotation.Resource;

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;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang3.StringEscapeUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.io.SAXReader;

/**

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

 *

 * @author alessia

 *

 */

public class RecordInfoGenerator {

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

	@Resource

	private MetadataExtractor metadataExtractor;

	@Resource

	private ProvenanceExtractor provenanceExtractor;

	@SuppressWarnings("unchecked")

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

		if ((object == null) || object.keySet().isEmpty()) return null;

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

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

		id = StringEscapeUtils.escapeXml11(id);

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

		RecordInfo record = new RecordInfo();

		record.setIdentifier(id);

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

		Object datestampObject = object.get(OAIConfigurationReader.DATESTAMP_FIELD);

		if(datestampObject instanceof Date){

			record.setDatestamp((Date) datestampObject);

		}

		else {

			//assuming we have a string in the form 2017-12-05T14:24:48.61Z to convert to Date

			log.debug("Why the hell there are string instead of Dates in datestamp???");

			try {

				LocalDateTime d = LocalDateTime.parse(datestampObject.toString(), DateTimeFormatter.ISO_INSTANT);

				Date utilDate = Date.from(d.atZone(ZoneId.systemDefault()).toInstant());

				record.setDatestamp(utilDate);

			}catch(Exception dateException){

				log.warn("Error setting date from a string datestamp, using current date instead: "+dateException.getMessage());

				record.setDatestamp(new Date());

			}

		}

		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 = decompressRecord((byte[]) 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));

				record.setProvenance(this.provenanceExtractor.evaluate(doc));

			} catch (DocumentException e) {

				throw new OaiPublisherRuntimeException(e);

			}

		}

		return record;

	}

	public  String decompressRecord(final byte[] input)  {

		try {

			ByteArrayInputStream bis = new ByteArrayInputStream(input);

			ZipInputStream zis = new ZipInputStream(bis);

			ZipEntry ze;

			ze = zis.getNextEntry();

			if (ze== null)

				throw new OaiPublisherRuntimeException("cannot decompress null zip entry ");

			if (!ze.getName().equals(OAIConfigurationReader.BODY_FIELD))

				throw new OaiPublisherRuntimeException("cannot decompress zip entry name :"+ze.getName());

			return IOUtils.toString(zis);

		} catch (IOException e) {

			throw new OaiPublisherRuntimeException(e);

		}

	}

	public MetadataExtractor getMetadataExtractor() {

		return metadataExtractor;

	}

	public void setMetadataExtractor(final MetadataExtractor metadataExtractor) {

		this.metadataExtractor = metadataExtractor;

	}

	public ProvenanceExtractor getProvenanceExtractor() {

		return provenanceExtractor;

	}

	public void setProvenanceExtractor(final ProvenanceExtractor provenanceExtractor) {

		this.provenanceExtractor = provenanceExtractor;

	}

}

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

import java.io.IOException;

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 java.util.zip.ZipEntry;

import java.util.zip.ZipOutputStream;

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

import com.google.common.collect.Multimap;

import com.mongodb.BasicDBObject;

import com.mongodb.BasicDBObjectBuilder;

import com.mongodb.DBObject;

import com.mongodb.WriteConcern;

import com.mongodb.client.FindIterable;

import com.mongodb.client.ListIndexesIterable;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoDatabase;

import com.mongodb.client.model.Filters;

import com.mongodb.client.model.IndexOptions;

import com.mongodb.client.model.Sorts;

import com.mongodb.client.model.UpdateOptions;

import com.mongodb.client.result.DeleteResult;

import com.mongodb.client.result.UpdateResult;

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

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

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

import eu.dnetlib.functionality.cql.mongo.MongoCqlTranslator;

import eu.dnetlib.miscutils.functional.UnaryFunction;

import org.apache.commons.io.output.ByteArrayOutputStream;

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.bson.types.Binary;

public class MongoPublisherStore implements PublisherStore<DNetOAIMongoCursor> {

	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 MongoCollection<DBObject> collection;

	private MongoCollection<DBObject> discardedCollection;

	private RecordInfoGenerator recordInfoGenerator;

	private MetadataExtractor metadataExtractor;

	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.

	 */

	private String idScheme;

	/**

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

	 * http://www.openarchives.org/OAI/openarchivesprotocol.html#UniqueIdentifier.

	 */

	private String idNamespace;

	private boolean alwaysNewRecord;

	public MongoPublisherStore() {

		super();

	}

	public MongoPublisherStore(final String id,

			final String metadataFormat,

			final String interpretation,

			final String layout,

			final MongoCollection<DBObject> collection,

			final List<PublisherField> mongoFields,

			final RecordInfoGenerator recordInfoGenerator,

			final String idScheme,

			final String idNamespace,

			final MetadataExtractor metadataExtractor,

			final RecordChangeDetector recordChangeDetector,

			final boolean alwaysNewRecord,

			final MongoDatabase mongodb) {

		super();

		this.id = id;

		this.metadataFormat = metadataFormat;

		this.interpretation = interpretation;

		this.layout = layout;

		this.collection = collection;

		this.discardedCollection = mongodb.getCollection("discarded-" + collection.getNamespace().getCollectionName(), DBObject.class);

		this.mongoFields = mongoFields;

		this.recordInfoGenerator = recordInfoGenerator;

		this.idScheme = idScheme;

		this.idNamespace = idNamespace;

		this.recordChangeDetector = recordChangeDetector;

		this.alwaysNewRecord = alwaysNewRecord;

	}

	@Override

	public RecordInfo getRecord(final String recordId) {

		Bson query = Filters.eq(OAIConfigurationReader.ID_FIELD, recordId);

		DBObject result = this.collection.find(query).first();

		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 DNetOAIMongoCursor getRecords(final String queryString, final boolean bodyIncluded, final int limit) {

		FindIterable<DBObject> iter = loggedFindByQuery(queryString, limit);

		return new DNetOAIMongoCursor(iter.iterator(), bodyIncluded, this.recordInfoGenerator, this.metadataExtractor);

	}

	@Override

	public DNetOAIMongoCursor getRecords(final String queryString,

			final UnaryFunction<String, String> unaryFunction,

			final boolean bodyIncluded,

			final int limit) {

		FindIterable<DBObject> iter = loggedFindByQuery(queryString, limit);

		return new DNetOAIMongoCursor(iter.iterator(), unaryFunction, bodyIncluded, this.recordInfoGenerator, this.metadataExtractor);

	}

	private FindIterable<DBObject> loggedFindByQuery(final String queryString, final int limit) {

		final Bson query = parseQuery(queryString);

		long start = System.currentTimeMillis();

		Bson sortByIdAsc = Sorts.orderBy(Sorts.ascending("_id"));

		FindIterable<DBObject> iter = this.collection.find(query).sort(sortByIdAsc).limit(limit);

		long end = System.currentTimeMillis();

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

		return iter;

	}

	private Bson parseQuery(final String query) {

		try {

			return MongoCqlTranslator.toMongo(query);

		} catch(Exception e ) {

			throw new OaiPublisherRuntimeException(e);

		}

	}

	@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}

	 *

	 */

	@Override

	public void ensureIndices() {

		final ListIndexesIterable<BasicDBObject> indexesIterable = this.collection.listIndexes(BasicDBObject.class);

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

		Stopwatch sw = Stopwatch.createUnstarted();

		sw.start();

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

		// I DO NOT NEED TO DO THIS. If the indexes are there, they are there.

		/*log.debug("Ensuring currently defined composite indexes on store "+id+": ");

		for (BasicDBObject o : indexesIterable) {

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

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

				log.debug(o);

				this.collection.createIndex(fieldIndexed, indexOptions);

			}

		}*/

		// Indexes on single fields.

		for (PublisherField field : this.mongoFields) {

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

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

			this.collection.createIndex(mongoIdx, indexOptions);

		}

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

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

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

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

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

		}

		BasicDBObjectBuilder theIndexBuilder = BasicDBObjectBuilder.start();

		for (String f : fieldNames) {

			theIndexBuilder.add(f, 1);

		}

		BasicDBObject theIndex = (BasicDBObject) theIndexBuilder.get();

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

		this.getCollection().createIndex(theIndex, new IndexOptions().background(true));

	}

	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.deleteMany(Filters.eq(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() {

				//For fast feeding we want to use a collection with unack write concern

				final MongoCollection<DBObject> unackCollection = collection.withWriteConcern(WriteConcern.UNACKNOWLEDGED);

				while (true) {

					try {

						Object record = queue.take();

						if (record == sentinel) {

							break;

						}

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

					} 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) {

		//get the collection with ACKNOWLEDGE Write concern

		final MongoCollection<DBObject> ackCollection = collection.withWriteConcern(WriteConcern.ACKNOWLEDGED);

		Thread deletedSetter = new Thread(new Runnable() {

			@Override

			public void run() {

				Bson filter = Filters.and(Filters.eq(OAIConfigurationReader.DELETED_FIELD, false),

						Filters.lt(OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD, feedDate));

				if (!StringUtils.isBlank(source)) {

					filter = Filters.and(filter, Filters.eq(OAIConfigurationReader.SET_FIELD, source));

				}

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

				BasicDBObject 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());

				final UpdateResult updateResult = ackCollection.updateMany(filter, update, new UpdateOptions().upsert(false));

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

			}

		});

		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) {

		Bson query = parseQuery(queryString);

		final DeleteResult deleteResult = this.collection.deleteMany(query);

		log.debug("Deleted by query: " + queryString + " #deleted: " + deleteResult.getDeletedCount());

	}

	@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();

		Bson query = parseQuery(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...");

		return Lists.newArrayList(this.collection.distinct(OAIConfigurationReader.SET_FIELD, String.class));

	}

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

	// Feed utilities

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

	private boolean safeFeedRecord(final String record, final String source, final Date feedDate, final MongoCollection<DBObject> unackCollection) {

		try {

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

		} catch (final Throwable e) {

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

			discardedCollection.insertOne(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, final MongoCollection<DBObject> unackCollection) {

		PublisherRecordParser parser = new PublisherRecordParser(this.mongoFields);

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

		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, unackCollection);

				return true;

			} else {

				if (isChanged(oaiID, record)) {

					updateRecord(oaiID, record, recordProperties, feedDate, unackCollection);

				} else {

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

					handleRecord(oaiID, feedDate, unackCollection);

				}

			}

		} else {

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

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

					OAIConfigurationReader.DATESTAMP_FIELD, feedDate));

		}

		return false;

	}

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

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

					}

				}

			}

		}

		try {

			obj.put(OAIConfigurationReader.BODY_FIELD, createCompressRecord(record));

			obj.put(OAIConfigurationReader.DELETED_FIELD, false);

			return obj;

		} catch (IOException e) {

			throw new OaiPublisherRuntimeException(e);

		}

	}

	/**

	 * @param record

	 * @throws IOException

	 */

	public Binary createCompressRecord(final String record) throws IOException {

		ByteArrayOutputStream os = new ByteArrayOutputStream();

		ZipOutputStream zos = new ZipOutputStream(os);

		ZipEntry entry = new ZipEntry(OAIConfigurationReader.BODY_FIELD);

		zos.putNextEntry(entry);

		zos.write(record.getBytes());

		zos.closeEntry();

		zos.flush();

		zos.close();

		return new Binary(os.toByteArray());

	}

	private void feedNew(final String oaiID,

			final String record,

			final Multimap<String, String> recordProperties,

			final Date feedDate,

			final MongoCollection<DBObject> unackCollection) {

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

		unackCollection.insertOne(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,

			final MongoCollection<DBObject> unackCollection) {

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

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

		Bson oldObj = Filters.eq(OAIConfigurationReader.ID_FIELD, oaiID);

		unackCollection.replaceOne(oldObj, obj, new UpdateOptions().upsert(true));

		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().getNamespace().getDatabaseName());

				}

			}

		}

	}

	private void handleRecord(final String oaiID, final Date lastCollectionDate, final MongoCollection<DBObject> unackCollection) {

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

		Bson oldObj = Filters.eq(OAIConfigurationReader.ID_FIELD, oaiID);

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

		unackCollection.updateOne(oldObj, update, new UpdateOptions().upsert(true));

	}

	private boolean isNewRecord(final String oaiIdentifier) {

		if (alwaysNewRecord || (collection.count() == 0)) return true;

		return this.collection.find(Filters.eq(OAIConfigurationReader.ID_FIELD, oaiIdentifier)).first() == null;

	}

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

	// Setters / Getters / Basic utilities

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

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

		RecordInfo oldRecord = getRecord(oaiID);

		if (oldRecord == null) return StringUtils.isBlank(record);

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

	}

	private String getOAIIdentifier(final String id) {

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

	}

	@Override

	public String toString() {

		return "MongoPublisherStore{" +

				"id='" + id + '\'' +

				", metadataFormat='" + metadataFormat + '\'' +

				", interpretation='" + interpretation + '\'' +

				", layout='" + layout + '\'' +

				", idScheme='" + idScheme + '\'' +

				", idNamespace='" + idNamespace + '\'' +

				", alwaysNewRecord=" + alwaysNewRecord +

				'}';

	}

	@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 MongoCollection<DBObject> getCollection() {

		return collection;

	}

	public void setCollection(final MongoCollection<DBObject> collection) {

		this.collection = collection;

	}

	public MongoCollection<DBObject> getDiscardedCollection() {

		return discardedCollection;

	}

	public void setDiscardedCollection(final MongoCollection<DBObject> 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;

	}

	public void setId(final String id) {

		this.id = id;

	}

	@Override

	public String getMetadataFormat() {

		return this.metadataFormat;

	}

	public void setMetadataFormat(final String metadataFormat) {

		this.metadataFormat = metadataFormat;

	}

	@Override

	public String getInterpretation() {

		return this.interpretation;

	}

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

	@Override

	public String getLayout() {

		return this.layout;

	}

	public void setLayout(final String layout) {

		this.layout = 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;

	}

	public boolean isAlwaysNewRecord() {

		return alwaysNewRecord;

	}

	public void setAlwaysNewRecord(final boolean alwaysNewRecord) {

		this.alwaysNewRecord = alwaysNewRecord;

	}

}

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

import java.util.List;

import javax.annotation.Resource;

import com.mongodb.BasicDBObjectBuilder;

import com.mongodb.DBObject;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoDatabase;

import com.mongodb.client.model.Filters;

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.sets.MongoSetCollection;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

public class MongoPublisherStoreDAO implements PublisherStoreDAO<MongoPublisherStore, DNetOAIMongoCursor> {

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

	@Autowired

	private MongoPublisherCacheHelper mongoCacheHelper;

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

	private String metadataCollection;

	@Resource

	private MongoSetCollection mongoSetCollection;

	private boolean alwaysNewRecord;

	@Resource(name = "oaiConfigurationExistReader")

	private OAIConfigurationReader configuration;

	@Override

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

		return mongoCacheHelper.listPublisherStores(dbName, metadataCollection, alwaysNewRecord, mongoSetCollection);

	}

	@Override

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

		return mongoCacheHelper.getStoreById(storeId, dbName, metadataCollection, alwaysNewRecord,  mongoSetCollection);

	}

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

		MongoPublisherStore store = this.getStore(info.getSourceFormatName(), info.getSourceFormatInterpretation(), info.getSourceFormatLayout(), dbName);

		return store;

	}

	@Override

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

			throws OaiPublisherException {

		log.debug("CREATING OAI STORE");

		MongoDatabase db = this.mongoCacheHelper.getDB(dbName);

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

		MongoCollection<DBObject> metadata = db.getCollection(this.metadataCollection, DBObject.class);

		String id = (String) store.get("id");

		metadata.insertOne(store);

		log.debug("metadata insert done for store "+id);

		MongoPublisherStore theStore = this.mongoCacheHelper.getStoreById(id, dbName, this.metadataCollection, this.alwaysNewRecord, this.mongoSetCollection);

		return theStore;

	}

	@Override

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

		log.debug("DELETING OAI STORE "+storeId+" -- db: "+dbName);

		this.mongoCacheHelper.deleteFromCache(storeId, dbName);

		MongoDatabase db = this.mongoCacheHelper.getDB(dbName);

		MongoCollection<DBObject> metadata = db.getCollection(this.metadataCollection, DBObject.class);

		final DBObject storeDeleted = metadata.findOneAndDelete(Filters.eq("id", storeId));

		if (storeDeleted == null) return false;

		else {

			db.getCollection(storeId).drop();

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

			this.mongoSetCollection.dropOAISets(dbName);

			log.debug("Deleted oaistore " + storeId + ", db: " + dbName);

			return true;

		}

	}

	@Override

	public boolean deleteFromStore(final String storeId, final String dbName, final String set) {

		String setSpec = mongoSetCollection.normalizeSetSpec(set);

		log.debug(String.format("DELETING OAI STORE %s BY SET %s from db %s", storeId, setSpec, dbName));

		this.mongoCacheHelper.deleteFromCache(storeId, dbName);

		MongoDatabase db = this.mongoCacheHelper.getDB(dbName);

		MongoCollection<DBObject> metadata = db.getCollection(this.metadataCollection, DBObject.class);

		DBObject storeInfo = metadata.find(Filters.eq("id", storeId)).first();

		if (storeInfo == null) return false;

		else {

			db.getCollection(storeId).deleteMany(Filters.eq(OAIConfigurationReader.SET_FIELD, setSpec));

			this.mongoSetCollection.dropSet(dbName, setSpec);

			log.debug("Deleted set " + setSpec + " from oaistore " + storeId + ", db: " + dbName);

			return true;

		}

	}

	@Override

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

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

	}

	@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 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();

		log.debug("Created DBObject for OAIStore metadata entry: "+info.toString());

		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 OAIConfigurationReader getConfiguration() {

		return configuration;

	}

	public void setConfiguration(final OAIConfigurationReader configuration) {

		this.configuration = configuration;

	}

	public MongoSetCollection getMongoSetCollection() {

		return mongoSetCollection;

	}

	public void setMongoSetCollection(final MongoSetCollection mongoSetCollection) {

		this.mongoSetCollection = mongoSetCollection;

	}

	public boolean isAlwaysNewRecord() {

		return alwaysNewRecord;

	}

	public void setAlwaysNewRecord(final boolean alwaysNewRecord) {

		this.alwaysNewRecord = alwaysNewRecord;

	}

}

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

import java.util.List;