D-Net

package eu.dnetlib.msro.openaireplus.workflows.nodes.index;

import java.io.StringReader;

import java.util.ArrayList;

import java.util.List;

import javax.annotation.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.index.CloudIndexClient;

import eu.dnetlib.data.index.CloudIndexClientFactory;

import eu.dnetlib.data.index.CloudIndexClientException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.functional.xml.ApplyXslt;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.nodes.AsyncJobNode;

import eu.dnetlib.openaire.directindex.api.RecentResultsQueue;

import eu.dnetlib.openaire.directindex.utils.OafToIndexRecordFactory;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.solr.common.SolrInputDocument;

import org.dom4j.io.SAXReader;

import org.springframework.beans.factory.annotation.Required;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.core.io.ClassPathResource;

/**

 * Created by michele on 15/12/15.

 */

public class FeedMissingClaimsJobNode extends AsyncJobNode {

	private static final Log log = LogFactory.getLog(FeedMissingClaimsJobNode.class);

	public static final int BATCH_SIZE = 1000;

	private RecentResultsQueue queue;

	private OafToIndexRecordFactory oafToIndexRecordFactory;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Value(value = "${openaire.api.directindex.findSolrIndexUrl.xquery}")

	private ClassPathResource findSolrIndexUrl;

	@Override

	protected String execute(final NodeToken nodeToken) throws Exception {

		final String format =

				nodeToken.getEnv().hasAttribute("format") ? nodeToken.getEnv().getAttribute("format") : nodeToken.getFullEnv().getAttribute("format");

		final String coll = format + "-index-openaire";

		final String indexDsId = nodeToken.getEnv().getAttribute("index_id");

		final String baseUrl = calculateIndexBaseUrl();

		CloudIndexClient idxClient = null;

		try {

			final List<SolrInputDocument> toFeed = new ArrayList<SolrInputDocument>();

			final List<String> toDeleteFromCache = new ArrayList<String>();

			final SAXReader reader = new SAXReader();

			final ApplyXslt xslt = oafToIndexRecordFactory.newTransformer(format);

			idxClient = CloudIndexClientFactory.newIndexClient(baseUrl, coll, false);

			log.info("Starting to feed claims in index collection "+coll);

			int count = 0;

			for (String record : queue) {

				try {

					final String id = reader.read(new StringReader(record)).valueOf("//*[local-name() = 'objIdentifier']");

					count++;

					if (log.isDebugEnabled()) {

						log.debug("Processing record " + count);

					}

					if (idxClient.isRecordIndexed(id)) {

						toDeleteFromCache.add(id);

					} else {

						toFeed.add(idxClient.prepareSolrDocument(record, indexDsId, xslt));

					}

					if (count % BATCH_SIZE == 0) processLists(idxClient, toFeed, toDeleteFromCache);

				}catch(CloudIndexClientException cie){

					log.error("Error feeding missing claims", cie);

					if (idxClient != null) {

						idxClient.close();

					}

					idxClient = CloudIndexClientFactory.newIndexClient(baseUrl, coll, false);

					log.info("Got new CloudIndexClient");

				}

			}

			if(!toFeed.isEmpty() || !toDeleteFromCache.isEmpty()) processLists(idxClient, toFeed, toDeleteFromCache);

			log.info(String.format("Finished feeding of claims in index collection %s, total: %d", coll, count));

		} catch(Throwable e) {

			log.error("Error feeding missing claims", e);

			throw new MSROException("Error feeding missing claims: " + e.getMessage(), e);

		} finally {

			if (idxClient != null) {

				idxClient.close();

			}

			log.info("Closed Solr index client");

		}

		log.info("Now proceeding to Arc.DEFAULT_ARC");

		return Arc.DEFAULT_ARC;

	}

	private void processLists(final CloudIndexClient idxClient, final List<SolrInputDocument> toFeed, final List<String> toDeleteFromCache) throws CloudIndexClientException{

		idxClient.feed(toFeed, null);

		queue.remove(toDeleteFromCache);

		log.info(String.format("%d claims fed and cache cleaned of %d records", toFeed.size(), toDeleteFromCache.size()));

		toFeed.clear();

		toDeleteFromCache.clear();

		log.info("Cleaned temporary lists");

	}

	public RecentResultsQueue getQueue() {

		return queue;

	}

	@Required

	public void setQueue(final RecentResultsQueue queue) {

		this.queue = queue;

	}

	public OafToIndexRecordFactory getOafToIndexRecordFactory() {

		return oafToIndexRecordFactory;

	}

	@Required

	public void setOafToIndexRecordFactory(final OafToIndexRecordFactory oafToIndexRecordFactory) {

		this.oafToIndexRecordFactory = oafToIndexRecordFactory;

	}

	private String calculateIndexBaseUrl() throws Exception {

		final String query = IOUtils.toString(findSolrIndexUrl.getInputStream());

		return serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(query);

	}

}