D-Net

package eu.dnetlib.index.actors;

import java.util.Map;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.function.Function;

import java.util.stream.Stream;

import java.util.stream.StreamSupport;

import com.google.common.collect.Maps;

import eu.dnetlib.clients.index.model.Any.ValueType;

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

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

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

import eu.dnetlib.cql.CqlTranslator;

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.feed.DocumentFeeder;

import eu.dnetlib.index.feed.DocumentMapperFactory;

import eu.dnetlib.index.feed.FeedMode;

import eu.dnetlib.index.feed.FeedResult;

import eu.dnetlib.index.utils.IndexDateUtility;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.rmi.provision.IndexServiceException;

import eu.dnetlib.utils.MetadataReference;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

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

			final MetadataReference mdref = serviceTools.getMetadataRef(dsId);

			final IndexServerDAO serverDAO = indexServerDAOMap.getIndexServerDAO(backendId);

			final DocumentMapperFactory docMapperFactory = serverDAO.getDocumentMapperFactory();

            final String version = IndexDateUtility.getParsedDateField(DateUtils.now_ISO8601());

            Map<String, ValueType> fields = serverDAO.getSchema(mdref);

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

			Stream<IndexDocument> docStream = DnetStreamSupport.generateStreamFromIterator(docIterator.iterator()).map(docMapper);

			indexCollection = serverDAO.getIndexCollection(mdref);

			final FeedResult res = threadPool.submit(new DocumentFeeder(indexCollection, docStream)).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(buildParams(res));

		} catch (final Exception e) {

			endCallback.setJobFailed(e);

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

		} finally {

			if (indexCollection != null) {

				indexCollection.shutdown();

			}

		}

	}

	private Map<String, String> buildParams(final FeedResult feedResult) {

		Map<String, String> params = Maps.newHashMap();

		params.put("added", String.valueOf(feedResult.getAdded()));

		params.put("skipped", String.valueOf(feedResult.getSkipped()));

		params.put("marked", String.valueOf(feedResult.getMarked()));

		params.put("time", String.valueOf(feedResult.getTime()));

		return params;

	}

	/**

	 * 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 version) throws IndexServiceException {

        final String query = String.format("%s:[* TO \"%s\"}", IndexFieldUtility.DS_VERSION, version);

        indexCollection.deleteByQuery(query, dsId);

	}

	/**

	 * 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 query = IndexFieldUtility.DELETE_DOCUMENT + ":true ";

		indexCollection.deleteByQuery(query, dsId);

	}

}