D-Net

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

import java.io.StringReader;

import java.util.ArrayList;

import java.util.List;

import javax.annotation.Resource;

import com.google.common.base.Function;

import eu.dnetlib.data.index.CloudIndexClient;

import eu.dnetlib.data.index.CloudIndexClientException;

import eu.dnetlib.data.index.CloudIndexClientFactory;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.workflows.graph.Arc;

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

import eu.dnetlib.msro.workflows.procs.Env;

import eu.dnetlib.openaire.api.RecentPublicationsQueue;

import eu.dnetlib.rmi.enabling.ISLookUpService;

import eu.dnetlib.rmi.manager.MSROException;

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

	private RecentPublicationsQueue queue;

	private OafToIndexRecordFactory oafToIndexRecordFactory;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Value(value = "${openaireplus.msro.api.findSolrIndexUrl.xquery}")

	private ClassPathResource findSolrIndexUrl;

	@Override

	protected String execute(final Env env) throws Exception {

		final String format = env.getAttribute("format", String.class);

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

		final String indexDsId = env.getAttribute("index_id", String.class);

		final String baseUrl = calculateIndexBaseUrl();

		CloudIndexClient idxClient = null;

		try {

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

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

			final SAXReader reader = new SAXReader();

			final Function<String, String> xslt = (Function<String, String>) oafToIndexRecordFactory.newTransformer(format);

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

			for (final String record : this.queue) {

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

				if (idxClient.isRecordIndexed(id)) {

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

				} else {

					toDeleteFromCache.add(id);

				}

			}

			idxClient.feed(toFeed, null);

			this.queue.remove(toDeleteFromCache);

		} catch (final 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();

			}

		}

		return Arc.DEFAULT_ARC;

	}

	public RecentPublicationsQueue getQueue() {

		return this.queue;

	}

	@Required

	public void setQueue(final RecentPublicationsQueue queue) {

		this.queue = queue;

	}

	public OafToIndexRecordFactory getOafToIndexRecordFactory() {

		return this.oafToIndexRecordFactory;

	}

	@Required

	public void setOafToIndexRecordFactory(final OafToIndexRecordFactory oafToIndexRecordFactory) {

		this.oafToIndexRecordFactory = oafToIndexRecordFactory;

	}

	private String calculateIndexBaseUrl() throws Exception {

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

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

	}

	private class IsNotIndexed implements com.google.common.base.Predicate<String> {

		private final CloudIndexClient idxClient;

		private final SAXReader reader = new SAXReader();

		public IsNotIndexed(final CloudIndexClient idxClient) {

			this.idxClient = idxClient;

		}

		@Override

		public boolean apply(final String s) {

			try {

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

				return !this.idxClient.isRecordIndexed(id);

			} catch (final Throwable e) {

				log.error("Error searching record: " + s, e);

				throw new RuntimeException(e);

			}

		}

	}

	private class CreateSolrDocument implements Function<String, SolrInputDocument> {

		private final CloudIndexClient idxClient;

		private final String indexDsId;

		private final Function<String, String> toIndexRecord;

		public CreateSolrDocument(final CloudIndexClient idxClient, final String indexDsId, final Function<String, String> toIndexRecord) {

			this.idxClient = idxClient;

			this.indexDsId = indexDsId;

			this.toIndexRecord = toIndexRecord;

		}

		@Override

		public SolrInputDocument apply(final String s) {

			try {

				return this.idxClient.prepareSolrDocument(s, this.indexDsId, this.toIndexRecord);

			} catch (final CloudIndexClientException e) {

				throw new RuntimeException(e);

			}

		}

	}

}