D-Net

package eu.dnetlib.oai.core;

import java.util.ArrayList;

import java.util.Iterator;

import java.util.List;

import java.util.function.Function;

import javax.annotation.Resource;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import eu.dnetlib.oai.BadResumptionTokenException;

import eu.dnetlib.oai.Cursor;

import eu.dnetlib.oai.NoRecordsMatchException;

import eu.dnetlib.oai.OAIError;

import eu.dnetlib.oai.PublisherStore;

import eu.dnetlib.oai.PublisherStoreDAO;

import eu.dnetlib.oai.conf.OAIConfigurationReader;

import eu.dnetlib.oai.info.ListDocumentsInfo;

import eu.dnetlib.oai.info.RecordInfo;

import eu.dnetlib.oai.info.ResumptionTokenImpl;

import eu.dnetlib.rmi.provision.MDFInfo;

import eu.dnetlib.rmi.provision.OaiPublisherException;

import eu.dnetlib.rmi.provision.OaiPublisherRuntimeException;

public class DNetOAICore extends AbstractOAICore {

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

	@Resource(name = "mongodbPublisherStoreDao")

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

	private String defaultDate = "2008-01-01T12:00:00Z";

	@Override

	protected RecordInfo getRecordById(final MDFInfo mdf, final String id) throws OaiPublisherException {

		final PublisherStore<Cursor> store = this.publisherStoreDAO.getStoreFor(mdf.getPrefix(), getCurrentDBName());

		if (store == null) { throw new OaiPublisherRuntimeException(

				"Missing store for metadata prefix " + mdf.getPrefix() + ". Please check OAI publisher configuration."); }

		RecordInfo record = null;

		if (StringUtils.isBlank(mdf.getTransformationRuleID())) {

			record = store.getRecord(id);

		} else {

			final Function<String, String> function = getLookupClient().getUnaryFunctionFromTDSRule(mdf.getTransformationRuleID());

			record = store.getRecord(id, function);

		}

		if (record != null) {

			record.setPrefix(mdf.getPrefix());

		}

		return record;

	}

	/**

	 * {@inheritDoc}

	 * <p>

	 * String, String)

	 */

	@Override

	protected ListDocumentsInfo getDocuments(final boolean onlyIdentifiers,

			final String set,

			final String metadataPrefix,

			final String from,

			final String until)

			throws OaiPublisherException {

		final MDFInfo mdf = obtainMDFInfo(metadataPrefix);

		final boolean hasDateRange = StringUtils.isNotBlank(from) || StringUtils.isNotBlank(until);

		final String query = this.generateQuery(mdf, set, from, until, hasDateRange);

		final int total = this.countTotal(hasDateRange, query, set, mdf);

		log.debug("Total number of records: " + total);

		final Cursor results = this.getCursor(query, onlyIdentifiers, mdf);

		final ListDocumentsInfo res = this.prepareListDocumentsInfo(results, mdf, query, set, 0, total, hasDateRange);

		log.debug("Delivering " + res.getDocs().size() + " in a page");

		return res;

	}

	@Override

	protected ListDocumentsInfo getDocuments(final boolean onlyIdentifiers, final String resumptionToken) throws OaiPublisherException {

		final ResumptionTokenImpl resToken = new ResumptionTokenImpl();

		resToken.deserialize(resumptionToken);

		log.debug(resToken.toString());

		final MDFInfo mdf = obtainMDFInfo(resToken.getMetadataPrefix());

		final String lastID = resToken.getLastObjIdentifier();

		final String query = resToken.getQuery();

		String newQuery = "";

		if (StringUtils.isNotBlank(query)) {

			newQuery = query + " AND ";

		}

		newQuery += " _id > \"" + lastID + "\"";

		log.debug("NEW QUERY BECAUSE of resumptionToken: " + newQuery);

		final int total = this.countTotal(resToken.hasDateRange(), query, resToken.getRequestedSet(), mdf);

		final Cursor results = this.getCursor(newQuery, onlyIdentifiers, mdf);

		final int oldCount = resToken.getnMaxElements();

		// if the number of records changed, then for sure we can invalidate the resumption token, unless we have a new total of -1 (date

		// range queries can't be counted for performance reasons)

		if ((total != -1) && (oldCount != total)) { throw new BadResumptionTokenException(resumptionToken); }

		final ListDocumentsInfo res =

				this.prepareListDocumentsInfo(results, mdf, query, resToken.getRequestedSet(), resToken.getnRead(), resToken.getnMaxElements(),

						resToken.hasDateRange());

		res.setCursor(resToken.getnRead());

		return res;

	}

	protected ListDocumentsInfo prepareListDocumentsInfo(final Cursor results,

			final MDFInfo mdf,

			final String query,

			final String requestedSet,

			final int read,

			final int totalNumber,

			final boolean hasDateRange) throws OaiPublisherException {

		final ListDocumentsInfo documentList = new ListDocumentsInfo();

		documentList.setnMaxElements(totalNumber);

		documentList.setMetadataPrefix(mdf.getPrefix());

		documentList.setCursor(0);

		if (documentList.getnMaxElements() == 0) { throw new NoRecordsMatchException(OAIError.noRecordsMatch.getMessage()); }

		final List<RecordInfo> theRecords = this.generateOAIRecords(mdf, requestedSet, results);

		documentList.setDocs(theRecords);

		if ((theRecords == null) || theRecords.isEmpty()) { throw new NoRecordsMatchException("noRecordsMatch: 'documents' is null or empty"); }

		if ((documentList.getnMaxElements() > (read + theRecords.size())) || (documentList.getnMaxElements() == -1)) {

			final String lastID = theRecords.get(theRecords.size() - 1).getInternalId();

			final ResumptionTokenImpl nextToken = new ResumptionTokenImpl();

			nextToken.setDateRange(hasDateRange);

			nextToken.setLastObjIdentifier(lastID);

			nextToken.setMetadataPrefix(mdf.getPrefix());

			nextToken.setnMaxElements(totalNumber);

			nextToken.setnRead(read + theRecords.size());

			nextToken.setQuery(query);

			nextToken.setRequestedSet(requestedSet);

			documentList.setResumptionToken(nextToken);

		}

		return documentList;

	}

	protected Cursor getCursor(final String query, final boolean onlyIdentifiers, final MDFInfo mdfInfo) {

		final PublisherStore<Cursor> store = this.publisherStoreDAO.getStore(mdfInfo.getSourceFormat(), mdfInfo.getSourceInterpretation(),

				mdfInfo.getSourceLayout(), getCurrentDBName());

		if (store == null) { throw new OaiPublisherRuntimeException(

				"Missing store for metadata prefix " + mdfInfo.getPrefix() + ". Please check OAI publisher configuration."); }

		Cursor results = null;

		if (StringUtils.isBlank(mdfInfo.getTransformationRuleID())) {

			results = store.getRecords(query, !onlyIdentifiers, this.pageSize);

		} else {

			final Function<String, String> function = getLookupClient().getUnaryFunctionFromTDSRule(mdfInfo.getTransformationRuleID());

			results = store.getRecords(query, function, !onlyIdentifiers, this.pageSize);

		}

		return results;

	}

	/**

	 * Generates the List of RecordInfo to be delivered.

	 *

	 * @param mdf

	 *            MDFInfo, the requested metadata format information.

	 * @param requestedSet

	 *            set specified in the request. It is blank if no set was requested.

	 * @param cursor

	 *            Cursor instance to use to get the records.

	 * @return List of RecordInfo instances

	 */

	protected List<RecordInfo> generateOAIRecords(final MDFInfo mdf, final String requestedSet, final Cursor cursor) {

		final List<RecordInfo> documents = new ArrayList<>();

		final Iterator<RecordInfo> cursorIterator = cursor.iterator();

		while (cursorIterator.hasNext()) {

			final RecordInfo current = cursorIterator.next();

			current.addSetspec(requestedSet);

			current.setPrefix(mdf.getPrefix());

			documents.add(current);

		}

		return documents;

	}

	protected String generateQuery(final MDFInfo mdf, final String set, final String from, final String until, final boolean hasDateRange) {

		final String datestampIndexName = OAIConfigurationReader.DATESTAMP_FIELD;

		String query = mdf.getBaseQuery();

		if (!StringUtils.isBlank(set)) {

			if (!StringUtils.isBlank(query)) {

				query += " AND ";

			}

			query += getSetCollection().getSetQuery(set, getCurrentDBName());

		}

		if (hasDateRange) {

			if (!StringUtils.isBlank(query)) {

				query += " AND ";

			}

			if ((from != null) && (until != null)) {

				query += datestampIndexName + " >= " + from + " AND " + datestampIndexName + " <= " + until;

			} else if (from != null) {

				query += datestampIndexName + " >= " + from;

			} else if (until != null) {

				query += datestampIndexName + " <= " + until;

			}

		}

		log.info("QUERY GENERATED: \n" + query);

		return query;

	}

	private int countTotal(final boolean hasDateRange, final String query, final String set, final MDFInfo mdFormat) {

		int total = 0;

		if (hasDateRange) {

			// Counting in the store by date ranges is too expensive and delays to much the response

			total = -1;

		} else {

			String theSet = set;

			if (StringUtils.isBlank(set)) {

				theSet = "ALL";

			}

			log.debug("SET::: " + theSet);

			total = getSetCollection().count(theSet, mdFormat.getPrefix(), getCurrentDBName());

		}

		return total;

	}

	public String getDefaultDate() {

		return this.defaultDate;

	}

	public void setDefaultDate(final String defaultDate) {

		this.defaultDate = defaultDate;

	}

	public PublisherStoreDAO<PublisherStore<Cursor>, Cursor> getPublisherStoreDAO() {

		return this.publisherStoreDAO;

	}

	public void setPublisherStoreDAO(final PublisherStoreDAO<PublisherStore<Cursor>, Cursor> publisherStoreDAO) {

		this.publisherStoreDAO = publisherStoreDAO;

	}

	public int getPageSize() {

		return this.pageSize;

	}

	public void setPageSize(final int pageSize) {

		this.pageSize = pageSize;

	}

}