D-Net

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;

	}

}