D-Net

package eu.dnetlib.data.information.oai.publisher.core;

import java.util.Iterator;

import java.util.List;

import javax.annotation.Resource;

import com.google.common.collect.Lists;

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

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

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

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

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

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

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

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

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

import eu.dnetlib.miscutils.functional.UnaryFunction;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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 {

		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 {

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

			record = store.getRecord(id, function);

		}

		if (record != null) {

			record.setPrefix(mdf.getPrefix());

		}

		return record;

	}

	/**

	 * 

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.data.information.oai.publisher.core.AbstractOAICore#getDocuments(boolean, java.lang.String, java.lang.String,

	 *      java.lang.String, java.lang.String)

	 */

	@Override

	protected ListDocumentsInfo getDocuments(final boolean onlyIdentifiers, final String set, final String metadataPrefix, final String from, final String until)

			throws OaiPublisherException {

		MDFInfo mdf = obtainMDFInfo(metadataPrefix);

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

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

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

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

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

		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 {

		ResumptionTokenImpl resToken = new ResumptionTokenImpl();

		resToken.deserialize(resumptionToken);

		log.debug(resToken.toString());

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

		String lastID = resToken.getLastObjIdentifier();

		String query = resToken.getQuery();

		String newQuery = "";

		if (StringUtils.isNotBlank(query)) {

			newQuery = query + " AND ";

		}

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

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

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

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

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

		long start = System.currentTimeMillis();

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

				resToken.hasDateRange());

		res.setCursor(resToken.getnRead());

		long end = System.currentTimeMillis();

		log.debug("ListDocumentInfo ready in (ms): " + (end - start));

		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 {

		ListDocumentsInfo documentList = new ListDocumentsInfo();

		documentList.setnMaxElements(totalNumber);

		documentList.setMetadataPrefix(mdf.getPrefix());

		documentList.setCursor(0);

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

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

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

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

		PublisherStore<Cursor> store = this.publisherStoreDAO.getStore(mdfInfo.getSourceFormatName(), mdfInfo.getSourceFormatInterpretation(),

				mdfInfo.getSourceFormatLayout(), 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, pageSize);

		} else {

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

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

		}

		log.debug("Got cursor");

		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 = Lists.newArrayList();

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

		while (cursorIterator.hasNext()) {

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

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

	}

	public void setDefaultDate(final String defaultDate) {

		this.defaultDate = defaultDate;

	}

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

		return publisherStoreDAO;

	}

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

		this.publisherStoreDAO = publisherStoreDAO;

	}

	public int getPageSize() {

		return pageSize;

	}

	public void setPageSize(final int pageSize) {

		this.pageSize = pageSize;

	}

}