D-Net

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

import java.util.List;

import javax.annotation.Resource;

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;

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;

import eu.dnetlib.miscutils.cache.EhCache;

public class MongoPublisherStoreDAO implements PublisherStoreDAO<MongoPublisherStore, MongoCursor> {

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

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

	/**

	 * Cache for oaistores. Keys are dbName-storeId, values are objects of type MongoPublisherStore.

	 */

	@Resource

	private EhCache<String, MongoPublisherStore> mongoOaistoreCache;

	/**

	 * Cache for oaistores. Keys is OAI metadata prefixes, values are objects of type MongoPublisherStore.

	 */

	@Resource

	private EhCache<String, MongoPublisherStore> mongoOaistoreCacheByMdPrefix;

	private boolean alwaysNewRecord;

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

		if (mongoOaistoreCacheByMdPrefix.containsKey(targetMetadataPrefix)) return mongoOaistoreCacheByMdPrefix.get(targetMetadataPrefix);

		else {

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

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

			mongoOaistoreCacheByMdPrefix.put(targetMetadataPrefix, store);

			return store;

		}

	}

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

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

			return true;

		}

	}

	@Override

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

		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).remove(new BasicDBObject(OAIConfigurationReader.SET_FIELD, set));

			this.mongoSetCollection.dropSet(dbName, set);

			log.debug("Deleted set " + set + " 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 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");

		String k = db.getName() + "-" + storeId;

		if (mongoOaistoreCache.containsKey(k)) {

			log.debug("Store retreived from cache and alwaysNewRecord is" + alwaysNewRecord);

			MongoPublisherStore store = mongoOaistoreCache.get(k);

			store.setAlwaysNewRecord(alwaysNewRecord);

			return store;

		} else {

			log.debug("Store retreived, cache miss,  alwaysNewRecord is" + alwaysNewRecord);

			log.fatal("Not using cache to create oaistore from dbObject: " + k);

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

			store.setMongoSetCollection(mongoSetCollection);

			mongoOaistoreCache.put(k, store);

			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;

	}

	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;

	}

	public MongoSetCollection getMongoSetCollection() {

		return mongoSetCollection;

	}

	public void setMongoSetCollection(final MongoSetCollection mongoSetCollection) {

		this.mongoSetCollection = mongoSetCollection;

	}

	public Mongo getPublisherMongoServer() {

		return publisherMongoServer;

	}

	public void setPublisherMongoServer(final Mongo publisherMongoServer) {

		this.publisherMongoServer = publisherMongoServer;

	}

	public boolean isAlwaysNewRecord() {

		return alwaysNewRecord;

	}

	public void setAlwaysNewRecord(final boolean alwaysNewRecord) {

		this.alwaysNewRecord = alwaysNewRecord;

	}

}