D-Net

package eu.dnetlib.index.actors;

import java.io.IOException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.function.Function;

import eu.dnetlib.clients.index.model.document.IndexDocument;

import eu.dnetlib.clients.index.utils.IndexFieldUtility;

import eu.dnetlib.utils.MetadataReference;

import eu.dnetlib.clients.index.utils.ServiceTools;

import eu.dnetlib.enabling.tools.DnetStreamSupport;

import eu.dnetlib.index.IndexCollection;

import eu.dnetlib.index.IndexServerDAO;

import eu.dnetlib.index.IndexServerDAOMap;

import eu.dnetlib.index.cql.CqlTranslator;

import eu.dnetlib.index.feed.DocumentFeeder;

import eu.dnetlib.index.feed.DocumentMapperFactory;

import eu.dnetlib.index.feed.FeedMode;

import eu.dnetlib.index.feed.FeedResult;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.rmi.provision.IndexServiceException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.z3950.zing.cql.CQLParseException;

/**

 * The Class IndexFeedActorImpl.

 */

public class IndexFeedActorImpl implements IndexFeedActor {

	/**

	 * The Constant log.

	 */

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

	/**

	 * The index server dao map.

	 */

	private final transient IndexServerDAOMap indexServerDAOMap;

	/**

	 * The service tools.

	 */

	private final transient ServiceTools serviceTools;

	/**

	 * Thread pool used for the feeding process.

	 */

	private final transient ExecutorService threadPool = Executors.newCachedThreadPool();

	/**

	 * CqlTranslator.

	 */

	private CqlTranslator translator;

	/**

	 * Instantiates a new index feed actor impl.

	 *

	 * @param indexServerDAOMap the index server dao map

	 * @param serviceTools      the service tools

	 */

	public IndexFeedActorImpl(final IndexServerDAOMap indexServerDAOMap, final ServiceTools serviceTools, final CqlTranslator translator) {

		super();

		this.indexServerDAOMap = indexServerDAOMap;

		this.serviceTools = serviceTools;

		this.translator = translator;

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public void feedIndex(final String dsId,

			final FeedMode feedMode,

			final Iterable<String> docIterator,

			final ResultsetKeepAliveCallback startCallback,

			final BlackboardActorCallback endCallback,

			final String backendId, final boolean emptyResult) {

		IndexCollection indexCollection = null;

		try {

			startCallback.unschedule();

			final MetadataReference mdref = serviceTools.getMetadataRef(dsId);

			final IndexServerDAO serverDAO = indexServerDAOMap.getIndexServerDAO(backendId);

			final DocumentMapperFactory docMapperFactory = serverDAO.getDocumentMapperFactory();

			final String version = DateUtils.now_ISO8601();

			final Function<String, IndexDocument> docMapper = docMapperFactory.getRecordMapper(serverDAO.getSchema(mdref), mdref, dsId, version, emptyResult);

			Iterable<IndexDocument> result = DnetStreamSupport.generateStreamFromIterator(docIterator.iterator()).map(docMapper)::iterator;

			indexCollection = serverDAO.getIndexCollection(mdref);

			final FeedResult res = threadPool.submit(new DocumentFeeder(indexCollection, result)).get();

			cleanMarkedDocuments(indexCollection, dsId);

			if (feedMode.equals(FeedMode.REFRESH)) {

				deleteByVersion(indexCollection, dsId, version);

			}

			indexCollection.commit();

			indexCollection.shutdown();

			log.info("FeedResult: " + res.setTimeElapsed(System.currentTimeMillis()));

			endCallback.setJobDone();

		} catch (final Exception e) {

			endCallback.setJobFailed(e);

			log.error("feed index job failed", e);

		} finally {

			if (indexCollection != null) {

				indexCollection.shutdown();

			}

		}

	}

	/**

	 * mdFormatVersion.

	 *

	 * @param indexCollection the server dao

	 * @param dsId            the ds id

	 * @throws IndexServiceException the index service exception

	 */

	private void deleteByVersion(final IndexCollection indexCollection, final String dsId, final String cqlQuery) throws IndexServiceException {

		//final String cqlQuery = IndexFieldUtility.DS_VERSION + " < \"" + IndexDateUtility.getParsedDateField(version) + "\"";

		try {

			String luceneQuery = translator.getTranslatedQuery(cqlQuery).asLucene();

			indexCollection.deleteByQuery(luceneQuery, dsId);

		} catch (CQLParseException e) {

			throw new IndexServiceException("Cannot parse CQL query into lucene query: " + cqlQuery, e);

		} catch (IOException e) {

			throw new IndexServiceException("Cannot parse CQL query into lucene query: " + cqlQuery, e);

		}

	}

	/**

	 * method delete documents where IndexMap.DELETE_DOCUMENT field is true

	 *

	 * @param indexCollection the server dao

	 * @param dsId            the ds id

	 * @throws IndexServiceException the index service exception

	 */

	public void cleanMarkedDocuments(final IndexCollection indexCollection, final String dsId) throws IndexServiceException {

		final String cqlQuery = IndexFieldUtility.DELETE_DOCUMENT + " all true ";

		deleteByVersion(indexCollection, dsId, cqlQuery);

	}

}