D-Net

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

import java.util.Map.Entry;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.datasources.LocalOpenaireDatasourceManager;

import eu.dnetlib.enabling.datasources.common.Api;

import eu.dnetlib.enabling.datasources.common.ApiParam;

import eu.dnetlib.enabling.datasources.common.DsmException;

import eu.dnetlib.miscutils.datetime.DateUtils;

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

public class UpsertHostedByApisJobNode extends SimpleJobNode {

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

	private static final String HOSTED_BY_COMPLIANCE = "hostedBy";

	@Autowired

	private LocalOpenaireDatasourceManager dsManager;

	private String countersParam;

	private String mdId;

	public static final String extraFieldsForTotal = "last_aggregation_total";

	public static final String extraFieldForDate = "last_aggregation_date";

	public static final String extraFieldForMdId = "last_aggregation_mdId";

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final HostedByCounters counters = (HostedByCounters) token.getEnv().getTransientAttribute(countersParam);

		final String date = DateUtils.now_ISO8601();

		final String namespacePrefix = token.getEnv().getAttribute("namespacePrefix");

		log.info(counters);

		for (final Entry<String, Integer> e : counters.getCounters().entrySet()) {

			updateHostedByApi(e.getKey(), namespacePrefix, date, e.getValue());

		}

		return Arc.DEFAULT_ARC;

	}

	private void updateHostedByApi(final String dsId,

		final String namepsacePrefix,

		final String date,

		final int size) {

		log.info("Verifying hostedBy api in ds: " + dsId);

		try {

			for (final Api<ApiParam> iface : dsManager.getApis(dsId)) {

				if (HOSTED_BY_COMPLIANCE.equals(iface.getCompatibility())) { return; }

			}

			final Api<ApiParam> api = new Api<>();

			api.setDatasource(dsId);

			api.setId("api_________::" + dsId + "::hostedBy");

			api.setCompatibility(HOSTED_BY_COMPLIANCE);

			api.setProtocol("UNKNOWN");

			api.setContentdescription("metadata");

			api.setBaseurl("");

			api.setActive(false);

			api.setRemovable(true);

			dsManager.addApi(api);

		} catch (final DsmException e) {

			log.warn("Error setting hostedBy api of ds: " + dsId, e);

		}

	}

	public String getMdId() {

		return mdId;

	}

	public void setMdId(final String mdId) {

		this.mdId = mdId;

	}

	public String getCountersParam() {

		return countersParam;

	}

	public void setCountersParam(final String countersParam) {

		this.countersParam = countersParam;

	}

}

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

import java.io.IOException;

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;

    public static final int ATTEMPTS = 3;

    public static final int SLEEP_MS_SOLR_CLIENT = 5000;

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

                int max_attempts = ATTEMPTS;

                count++;

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

                if (log.isDebugEnabled()) {

                    log.debug(String.format("Processing record %s, number: %d", id, count));

                }

                if (isRecordIndexed(idxClient, baseUrl, coll, id, max_attempts)) {

                    toDeleteFromCache.add(id);

                } else {

                    max_attempts = ATTEMPTS;

                    toFeed.add(prepareSolrDoc(idxClient, baseUrl, coll, id, record, indexDsId, xslt, max_attempts));

                }

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

            }

            if (!toFeed.isEmpty() || !toDeleteFromCache.isEmpty()) processLists(idxClient, baseUrl, coll, 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 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;

    }

    protected boolean isRecordIndexed(CloudIndexClient idxClient, String baseUrl, String coll, String id, int attempt) throws IOException, CloudIndexClientException, MSROException, InterruptedException {

        try {

            return idxClient.isRecordIndexed(id);

        } catch (CloudIndexClientException cie) {

            log.error(String.format("Error querying for %s, message: %s. Trying again, remaining attempts:", id, cie, attempt));

            idxClient = resetCloudIndexClient(idxClient, baseUrl, coll);

            if (attempt > 0) return isRecordIndexed(idxClient, baseUrl, coll, id, --attempt);

            else {

                String msg = String.format("Too many attempts %d to recreate the index client for checking if record %s exists.", ATTEMPTS, id);

                log.error(msg);

                throw new MSROException(cie);

            }

        }

    }

    protected SolrInputDocument prepareSolrDoc(CloudIndexClient idxClient, String baseUrl, String coll, String recordId, String record, String indexDsId, ApplyXslt xslt, int attempt) throws IOException, CloudIndexClientException, MSROException, InterruptedException {

        try {

            return idxClient.prepareSolrDocument(record, indexDsId, xslt);

        } catch (CloudIndexClientException cie) {

            log.error(String.format("Error preparing Solr doc for %s, message: %s. Trying again, remaining attempts:", recordId, cie, attempt));

            idxClient = resetCloudIndexClient(idxClient, baseUrl, coll);

            if (attempt > 0)

                return prepareSolrDoc(idxClient, baseUrl, coll, recordId, record, indexDsId, xslt, --attempt);

            else {

                String msg = String.format("Too many attempts %d to recreate the index client for preparing SolrDocument for %s", ATTEMPTS, id);

                log.error(msg);

                throw new MSROException(cie);

            }

        }

    }

    protected void tryToFeed(CloudIndexClient idxClient, String baseUrl, String coll, List<SolrInputDocument> toFeed, int attempt) throws MSROException, IOException, CloudIndexClientException, InterruptedException {

        try {

            idxClient.feed(toFeed, null);

        } catch (CloudIndexClientException cie) {

            log.error(String.format("Error feeding Solr in attempt number %d", attempt));

            idxClient = resetCloudIndexClient(idxClient, baseUrl, coll);

            if (attempt > 0) tryToFeed(idxClient, baseUrl, coll, toFeed, --attempt);

            else {

                String msg = String.format("Too many attempts %d to recreate the index client for feeding Solr", ATTEMPTS);

                log.error(msg);

                throw new MSROException(cie);

            }

        }

    }

    private CloudIndexClient resetCloudIndexClient(CloudIndexClient idxClient, String baseUrl, String coll) throws IOException, CloudIndexClientException, InterruptedException {

        if (idxClient != null) {

            idxClient.close();

        }

        Thread.sleep(SLEEP_MS_SOLR_CLIENT);

        CloudIndexClient newclient = CloudIndexClientFactory.newIndexClient(baseUrl, coll, false);

        log.info("Got new CloudIndexClient");

        return newclient;

    }

    private void processLists(final CloudIndexClient idxClient, String baseUrl, String coll, final List<SolrInputDocument> toFeed, final List<String> toDeleteFromCache) throws CloudIndexClientException, MSROException, IOException, InterruptedException {

        int max_attempts = ATTEMPTS;

        tryToFeed(idxClient, baseUrl, coll, toFeed, max_attempts);

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

    }

}

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

import java.io.StringReader;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.resultset.MappedResultSetFactory;

import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;

import eu.dnetlib.enabling.resultset.client.utils.EPRUtils;

import eu.dnetlib.miscutils.functional.UnaryFunction;

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

import org.dom4j.Document;

import org.dom4j.Element;

import org.dom4j.io.SAXReader;

// TODO: Auto-generated Javadoc

/**

 * The Class FindHostedByJonbNode.

 */

public class FindHostedByJobNode extends SimpleJobNode {

	/**

	 * The input epr param.

	 */

	private String inputEprParam;

	/**

	 * The output epr param.

	 */

	private String outputEprParam;

	/**

	 * The counters param.

	 */

	private String countersParam;

	/**

	 * The result set client factory.

	 */

	private ResultSetClientFactory resultSetClientFactory;

	/**

	 * The mapped result set factory.

	 */

	private MappedResultSetFactory mappedResultSetFactory;

	private final String unknown_repo_id = "openaire____::1256f046-bf1f-4afc-8b47-d0b147148b18";

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.msro.workflows.nodes.SimpleJobNode#execute(com.googlecode.sarasvati.NodeToken)

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final W3CEndpointReference inputEpr = new EPRUtils().getEpr(token.getEnv().getAttribute(inputEprParam));

		final HostedByCounters counters = new HostedByCounters();

		final SAXReader reader = new SAXReader();

		final UnaryFunction<String, String> hostedByMapFunction = new UnaryFunction<String, String>() {

			@Override

			public String evaluate(final String input) {

				try {

					final Document doc = reader.read(new StringReader(input));

					final Element node = (Element) doc.selectSingleNode("//*[local-name()='hostedBy']");

					if (node != null) {

						String hostedById = node.attributeValue("id");

						if (!hostedById.equals(unknown_repo_id)) {

							counters.increaseCounter(hostedById);

						}

					}

				} catch (Exception e) {

				}

				return input;

			}

		};

		final W3CEndpointReference epr = mappedResultSetFactory.createMappedResultSet(inputEpr, hostedByMapFunction);

		token.getEnv().setAttribute(outputEprParam, epr.toString());

		token.getEnv().setTransientAttribute(getCountersParam(), counters);

		return Arc.DEFAULT_ARC;

	}

	/**

	 * @return the inputEprParam

	 */

	public String getInputEprParam() {

		return inputEprParam;

	}

	/**

	 * @param inputEprParam the inputEprParam to set

	 */

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	/**

	 * @return the outputEprParam

	 */

	public String getOutputEprParam() {

		return outputEprParam;

	}

	/**

	 * @param outputEprParam the outputEprParam to set

	 */

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	/**

	 * @return the resultSetClientFactory

	 */

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	/**

	 * @param resultSetClientFactory the resultSetClientFactory to set

	 */

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

	/**

	 * @return the mappedResultSetFactory

	 */

	public MappedResultSetFactory getMappedResultSetFactory() {

		return mappedResultSetFactory;

	}

	/**

	 * @param mappedResultSetFactory the mappedResultSetFactory to set

	 */

	public void setMappedResultSetFactory(final MappedResultSetFactory mappedResultSetFactory) {

		this.mappedResultSetFactory = mappedResultSetFactory;

	}

	/**

	 * @return the countersParam

	 */

	public String getCountersParam() {

		return countersParam;

	}

	/**

	 * @param countersParam the countersParam to set

	 */

	public void setCountersParam(final String countersParam) {

		this.countersParam = countersParam;

	}

}

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

import java.util.List;

import javax.annotation.Resource;

import com.google.common.collect.Iterables;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

public class EntityGrouperConfigurationLoader {

	/**

	 * logger.

	 */

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

	private String xquery;

	@Resource

	private UniqueServiceLocator serviceLocator;

	public String load() throws ISLookUpException {

		log.info("loading EntityGrouperConfigurationDSResourceType: " + getXquery());

		List<String> conf = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(getXquery());

		if (conf == null || conf.isEmpty()) { throw new IllegalStateException("unable to read entity grouper configuration profile"); }

		if (conf.size() > 1) { throw new IllegalStateException("found more than one EntityGrouperConfigurationDSResourceType"); }

		return Iterables.getOnlyElement(conf).replaceAll("\\s+", " ").trim();

	}

	public String getXquery() {

		return xquery;

	}

	@Required

	public void setXquery(final String xquery) {

		this.xquery = xquery;

	}

}

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

import java.io.StringWriter;

import java.util.Map;

import java.util.Map.Entry;

import com.google.common.collect.Maps;

public class HostedByCounters {

	private Map<String, Integer> counters = Maps.newHashMap();

	public void increaseCounter(final String dsId) {

		int val = counters.containsKey(dsId) ? counters.get(dsId) + 1 : 1;

		counters.put(dsId, val);

	}

	public Map<String, Integer> getCounters() {

		return counters;

	}

	@Override

	public String toString() {

		final StringWriter sw = new StringWriter();

		sw.append("\n");

		sw.append("**************************************************\n");

		for (Entry<String, Integer> e : counters.entrySet()) {

			sw.append(e.getKey());

			sw.append(" : ");

			sw.append(Integer.toString(e.getValue()));

			sw.append("\n");

		}

		sw.append("**************************************************\n");

		return sw.toString();

	}

}

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

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.provision.index.rmi.IndexService;

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

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.functionality.index.solr.feed.InputDocumentFactory;

import eu.dnetlib.msro.rmi.MSROException;

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

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import static java.lang.String.format;

public class FinalizeIndexJobNode extends BlackboardJobNode {

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

	@Autowired

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String obtainServiceId(final NodeToken token) {

		return getServiceLocator().getServiceId(IndexService.class);

	}

	@Override

	protected void prepareJob(final BlackboardJob job, final NodeToken token) throws Exception {

		final String indexDsId = getEnvParam(token, "index_id");

		log.info("preparing blackboard job DELETE_BY_QUERY index: " + indexDsId);

		final String backendId = getBackendId(indexDsId);

		if (StringUtils.isBlank(backendId))

			throw new MSROException("empty index backend Id");

		job.setAction("DELETE_BY_QUERY");

		job.getParameters().put("id", indexDsId);

		job.getParameters().put("backend_Id", backendId);

		job.getParameters().put("query", buildQuery(getEnvParam(token, "index.feed.timestamp")));

	}

	private String buildQuery(final String version) {

		final String query =

				String.format("__dsversion:{* TO %s}", InputDocumentFactory.getParsedDateField(version));

		log.info("delete by query: " + query);

		return query;

	}

	private String getEnvParam(final NodeToken token, final String name) throws MSROException {

		final String value = token.getEnv().getAttribute(name);

		if (StringUtils.isBlank(value))

			throw new MSROException(format("unable to finalize index feeding, cannot find property '%s' in the workflow env.", name));

		return value;

	}

	public String getBackendId(final String indexDsId) throws ISLookUpDocumentNotFoundException, ISLookUpException {

		return getServiceLocator().getService(ISLookUpService.class).getResourceProfileByQuery(

				"//RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value='" + indexDsId + "']//BACKEND/text()");

	}

}

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

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

import eu.dnetlib.enabling.datasources.common.LocalDatasourceManager;

import eu.dnetlib.msro.workflows.nodes.repobye.DeleteMetaWfJobNode;

public class DeleteOpenaireMetaWfJobNode extends DeleteMetaWfJobNode {

	@Autowired

	private LocalDatasourceManager<?, ?> dsManager;

	@Override

	protected void setActivationStatus(final String dsId, final String ifaceId, final boolean active) throws Exception {

		dsManager.setActive(dsId, ifaceId, active);

	}

}

<?xml version="1.0" ?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet45-parent</artifactId>

		<version>1.0.0</version>

		<relativePath />

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

	<artifactId>dnet-openaireplus-workflows</artifactId>

	<packaging>jar</packaging>

	<version>8.0.3</version>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-openaireplus-workflows/tags/dnet-openaireplus-workflows-8.0.3</developerConnection>

	</scm>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-data-flow-monitoring-core</artifactId>

			<version>[3.0.0, 4.0.0)</version>

			<exclusions>

				<exclusion>

					<artifactId>wstx-asl</artifactId>

					<groupId>org.codehaus.woodstox</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-enabling-database-api</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-enabling-database-service</artifactId>

			<version>[3.0.0,4.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-msro-service</artifactId>

			<version>[4.0.0,5.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-resultset-service</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaire-datasource-manager</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaireplus-mapping-utils</artifactId>

			<version>[6.3.24,7.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-hadoop-service-rmi</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-collector-plugins</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-actionmanager-api</artifactId>

			<version>[4.0.0,5.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-oai-common-workflows</artifactId>

			<version>[5.0.0,6.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-mongo-mdstore</artifactId>

			<version>[7.0.0,8.0.0)</version>

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-modular-objectstore-service</artifactId>

			<version>[4.2.1,5.0.0)</version>

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-index-client</artifactId>

			<version>[2.3.4,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-validator-workflows</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-deduplication</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>

		<!-- modular ui and servlet api are here because of the stats controller -->

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-directindex-api</artifactId>

			<version>[2.1.13,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>org.apache.velocity</groupId>

			<artifactId>velocity</artifactId>

			<version>1.7</version>

			<exclusions>

				<exclusion>

					<artifactId>antlr</artifactId>

					<groupId>antlr</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>org.apache.velocity</groupId>

			<artifactId>velocity-tools</artifactId>

			<version>2.0</version>

			<exclusions>

				<exclusion>

					<artifactId>antlr</artifactId>

					<groupId>antlr</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>javax.servlet</groupId>

			<artifactId>javax.servlet-api</artifactId>

			<version>${javax.servlet.version}</version>

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>${junit.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaireplus-profiles</artifactId>

			<version>[1.0.0,2.0.0)</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>io.springfox</groupId>

			<artifactId>springfox-swagger2</artifactId>

			<version>${springfox-version}</version>

		</dependency>

		<dependency>

			<groupId>org.mockito</groupId>

			<artifactId>mockito-core</artifactId>

			<version>1.9.5</version>

		</dependency>

		<dependency>

			<groupId>org.springframework</groupId>

			<artifactId>spring-jdbc</artifactId>

			<version>${spring.version}</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-misc-utils</artifactId>

			<version>[1.0.5, 2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib.dhp</groupId>

			<artifactId>dhp-common</artifactId>

			<version>1.2.4-branch_hadoop_aggregator</version>

			<exclusions>

				<exclusion>

					<groupId>net.sf.saxon</groupId>

					<artifactId>Saxon-HE</artifactId>

				</exclusion>

			</exclusions>

		</dependency>

	</dependencies>

	<properties>

		<springfox-version>2.5.0</springfox-version>

	</properties>

</project>

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

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

import eu.dnetlib.enabling.datasources.common.LocalDatasourceManager;

import eu.dnetlib.msro.workflows.nodes.repohi.UpdateMetaWfStatusJobNode;

public class UpdateOpenaireMetaWfStatusJobNode extends UpdateMetaWfStatusJobNode {

	@Autowired

	private LocalDatasourceManager<?, ?> dsManager;

	@Override

	protected void updateDatasource(final String dsId, final String ifaceId) throws Exception {

		if (!dsId.equals("openaire____::bootstrap")) {

			dsManager.setManaged(dsId, true);

			dsManager.setActive(dsId, ifaceId, true);

		}

	}

}

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

import java.util.List;

import java.util.Queue;

import javax.annotation.Resource;

import com.google.common.collect.Queues;

import com.googlecode.sarasvati.NodeToken;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.rmi.MSROException;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

/**

 * Supercedes eu.dnetlib.msro.openaireplus.workflows.nodes.index.FindSearchServiceJobNode.

 * Returns a queue of SearchService ids that match the xquery and saves it in the env with the name 'searchService_ids'.

 * Use with eu.dnetlib.msro.openaireplus.workflows.nodes.index.SwitchIndexesJobNode

 *

 * @author claudio, alessia

 */

public class FindSearchServicesJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Value(value = "${dnet.openaire.service.search.lookup.xquery}")

	private String xquery;

	/**

	 * {@inheritDoc}

	 *

	 * @throws ISLookUpException

	 * @throws MSROException

	 * @see com.googlecode.sarasvati.mem.MemNode#execute(com.googlecode.sarasvati.Engine, NodeToken)

	 */

	@Override

	public String execute(final NodeToken token) throws ISLookUpException, MSROException {

		log.info("lookup for search service: " + getXquery());

		List<String> searchServiceIds = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(getXquery());

		log.info("SearchService found: " + searchServiceIds.size());

		Queue<String> q = Queues.newLinkedBlockingQueue(searchServiceIds);

		if (searchServiceIds.isEmpty()) {

			return "notFound";

		} else {

			token.getEnv().setTransientAttribute("searchService_ids", q);

			return "found";

		}

	}

	public String getXquery() {

		return xquery;

	}

	public void setXquery(final String xquery) {

		this.xquery = xquery;

	}

}

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

import java.util.List;

import java.util.NoSuchElementException;

import javax.annotation.Resource;

import com.google.common.base.Joiner;

import com.google.common.base.Splitter;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.hadoop.config.ClusterName;

import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.rmi.MSROException;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;

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

public abstract class PrepareIISParamsV2 extends SimpleJobNode {

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

	@Resource

	protected ConfigurationEnumerator configurationEnumerator;

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String clusterName;

	private String clusterParam = "cluster";

	private String oozieWfAppPath;

	private String oozieWfAppPathParam = "oozie.wf.application.path";

	private String xqueryMdStoreService;

	private String mdStoreStoreLocationParam = "import_mdstore_service_location";

	private String xqueryObjectStoreService;

	private String objectStoreLocationParam = "import_content_object_store_location";

	private String xqueryIsLookupService;

	private String islookupLocationParam = "import_islookup_service_location";

	private String importProjectConceptsContextCSVParam = "import_project_concepts_context_ids_csv";

	private String importProjectConceptsContextCSV;

	private String importServicesCollectedFromParam = "import_infospace_eligible_service_collectedfrom_datasourceid";

	private String importServicesCollectedFrom;

	private String xqueryDatasetStore;

	private String mdStoreDatasetParam = "import_dataset_mdstore_ids_csv";

	private String objectStoreBlacklistCSV = "";

	protected void prepare(final NodeToken token) throws Exception {

		token.getEnv().setAttribute(getClusterParam(), getClusterName());

		// Assumes we only have one mdStore service instance

		token.getEnv().setAttribute(getMdStoreStoreLocationParam(), getServiceEndpoint(getXqueryMdStoreService()));

		// Assumes we only have one objectStore service instance

		token.getEnv().setAttribute(getObjectStoreLocationParam(), getServiceEndpoint(getXqueryObjectStoreService()));

		token.getEnv().setAttribute(getIslookupLocationParam(), getServiceEndpoint(getXqueryIsLookupService()));

		token.getEnv().setAttribute(getImportProjectConceptsContextCSVParam(), getImportProjectConceptsContextCSV());

		token.getEnv().setAttribute(getImportServicesCollectedFromParam(), getImportServicesCollectedFrom());

		Configuration conf = configurationEnumerator.get(ClusterName.valueOf(getClusterName()));

		String nameNode = conf.get("fs.defaultFS");

		token.getEnv().setAttribute(getOozieWfAppPathParam(), getURI(nameNode, getOozieWfAppPath()));

		token.getEnv().setAttribute(getMdStoreDatasetParam(), asCSV(getProfileIds(getXqueryDatasetStore())));

	}

	protected String getServiceEndpoint(final String xquery) throws MSROException {

		try {

			return Iterables.getOnlyElement(serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery));

		} catch (ISLookUpException e) {

			throw new MSROException("unable to fetch service endpoint", e);

		} catch (NoSuchElementException e) {

			throw new MSROException("unable to find service endpoint, xquery: " + getXqueryMdStoreService(), e);

		} catch (IllegalArgumentException e) {

			throw new MSROException("more than one services found, we assume to have only one available", e);

		}

	}

	protected String getProfileId(final String xquery) throws MSROException {

		try {

			return Iterables.getOnlyElement(serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery));

		} catch (ISLookUpException e) {

			throw new MSROException("unable to fetch profile id", e);

		} catch (NoSuchElementException e) {

			throw new MSROException("unable to find profile profile, xquery: " + xquery, e);

		} catch (IllegalArgumentException e) {

			throw new MSROException("more than one profile profiles was found, we assume to have only one available, xquery: " + xquery, e);

		}

	}

	protected List<String> getProfileIds(final String xquery) throws MSROException {

		try {

			List<String> ids = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery);

			if (ids.isEmpty()) {

				log.warn("couldn't find any profile, xquery: " + xquery);

			}

			return ids;

		} catch (ISLookUpException e) {

			throw new MSROException("unable to fetch profile ids, x query: " + xquery, e);

		}

	}

	protected String getFilteredObjectStoreCSV(final String xquery) throws MSROException {

		return asCSV(filter(getProfileIds(xquery), asList(getObjectStoreBlacklistCSV())));

	}

	protected List<String> filter(final List<String> list, final List<String> filter) {

		if (filter == null || filter.isEmpty()) { return list; }

		list.removeAll(filter);

		return list;

	}

	protected String asCSV(final List<String> list) {

		return Joiner.on(",").skipNulls().join(list);

	}

	protected List<String> asList(final String csv) {

		return Lists.newArrayList(Splitter.on(",").trimResults().omitEmptyStrings().split(csv));

	}

	private String getURI(final String nameNode, final String relative) {

		// TODO ensure to return a valid URI

		return nameNode + relative;

	}

	private String getZkQuorumCSV(final Configuration conf, final String zkPort) {

		return Joiner.on(":" + zkPort + ",").join(Splitter.on(",").omitEmptyStrings().split(conf.get("hbase.zookeeper.quorum")));

	}

	@Required

	public void setXqueryMdStoreService(final String xqueryMdStoreService) {

		this.xqueryMdStoreService = xqueryMdStoreService;

	}

	public String getXqueryMdStoreService() {

		return xqueryMdStoreService;

	}

	public String getMdStoreStoreLocationParam() {

		return mdStoreStoreLocationParam;

	}

	public void setMdStoreStoreLocationParam(final String mdStoreStoreLocationParam) {

		this.mdStoreStoreLocationParam = mdStoreStoreLocationParam;

	}

	public String getClusterName() {

		return clusterName;

	}

	public void setClusterName(final String clusterName) {

		this.clusterName = clusterName;

	}

	public String getClusterParam() {

		return clusterParam;

	}

	public void setClusterParam(final String clusterParam) {

		this.clusterParam = clusterParam;

	}

	public String getOozieWfAppPathParam() {

		return oozieWfAppPathParam;

	}

	public void setOozieWfAppPathParam(final String oozieWfAppPathParam) {

		this.oozieWfAppPathParam = oozieWfAppPathParam;

	}

	public String getOozieWfAppPath() {

		return oozieWfAppPath;

	}

	public void setOozieWfAppPath(final String oozieWfAppPath) {

		this.oozieWfAppPath = oozieWfAppPath;

	}

	@Required

	public String getXqueryDatasetStore() {

		return xqueryDatasetStore;

	}

	public void setXqueryDatasetStore(final String xqueryDatasetStore) {

		this.xqueryDatasetStore = xqueryDatasetStore;

	}

	public String getMdStoreDatasetParam() {

		return mdStoreDatasetParam;

	}

	public void setMdStoreDatasetParam(final String mdStoreDatasetParam) {

		this.mdStoreDatasetParam = mdStoreDatasetParam;

	}

	public String getXqueryObjectStoreService() {

		return xqueryObjectStoreService;

	}

	@Required

	public void setXqueryObjectStoreService(final String xqueryObjectStoreService) {

		this.xqueryObjectStoreService = xqueryObjectStoreService;

	}

	public String getObjectStoreLocationParam() {

		return objectStoreLocationParam;

	}

	public void setObjectStoreLocationParam(final String objectStoreLocationParam) {

		this.objectStoreLocationParam = objectStoreLocationParam;

	}

	public String getObjectStoreBlacklistCSV() {

		return objectStoreBlacklistCSV;

	}

	public void setObjectStoreBlacklistCSV(final String objectStoreBlacklistCSV) {

		this.objectStoreBlacklistCSV = objectStoreBlacklistCSV;

	}

	public String getXqueryIsLookupService() {

		return xqueryIsLookupService;

	}

	@Required

	public void setXqueryIsLookupService(final String xqueryIsLookupService) {

		this.xqueryIsLookupService = xqueryIsLookupService;

	}

	public String getIslookupLocationParam() {

		return islookupLocationParam;

	}

	public void setIslookupLocationParam(final String islookupLocationParam) {

		this.islookupLocationParam = islookupLocationParam;

	}

	public String getImportProjectConceptsContextCSVParam() {

		return importProjectConceptsContextCSVParam;

	}

	public void setImportProjectConceptsContextCSVParam(final String importProjectConceptsContextCSVParam) {

		this.importProjectConceptsContextCSVParam = importProjectConceptsContextCSVParam;

	}

	public String getImportProjectConceptsContextCSV() {

		return importProjectConceptsContextCSV;

	}

	public void setImportProjectConceptsContextCSV(final String importProjectConceptsContextCSV) {

		this.importProjectConceptsContextCSV = importProjectConceptsContextCSV;

	}

	public String getImportServicesCollectedFromParam() {

		return importServicesCollectedFromParam;

	}

	public void setImportServicesCollectedFromParam(String importServicesCollectedFromParam) {

		this.importServicesCollectedFromParam = importServicesCollectedFromParam;

	}

	public String getImportServicesCollectedFrom() {

		return importServicesCollectedFrom;

	}

	public void setImportServicesCollectedFrom(String importServicesCollectedFrom) {

		this.importServicesCollectedFrom = importServicesCollectedFrom;

	}

}

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

import javax.annotation.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;

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

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

import eu.dnetlib.msro.workflows.resultset.ProcessCountingResultSetFactory;

import eu.dnetlib.msro.workflows.util.ProgressProvider;

import eu.dnetlib.msro.workflows.util.ResultsetProgressProvider;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class ProcessContextsJobNode extends SimpleJobNode implements ProgressJobNode {

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

	private String eprParam;

	private String contextObj;

	private ResultsetProgressProvider progressProvider;

	private String contextId;

	private String contextLabel;

	private String contextType;

	private String contextParams;

	private String dashboardVisibility;

	@Resource

	private ResultSetClientFactory resultSetClientFactory;

	@Resource

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String epr = token.getEnv().getAttribute(eprParam);

		this.progressProvider = processCountingResultSetFactory.createProgressProvider(token.getProcess(), epr);

		final Iterable<String> iter = resultSetClientFactory.getClient(progressProvider.getEpr());

		final ContextDesc context = ContextUtils.getContext(iter, getContextId(), getContextLabel(), getContextType(), getContextParams(), getDashboardVisibility());

		token.getEnv().setTransientAttribute(contextObj, context);

		return Arc.DEFAULT_ARC;

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

	public String getContextObj() {

		return contextObj;

	}

	public void setContextObj(final String contextObj) {

		this.contextObj = contextObj;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return progressProvider;

	}

	public String getContextId() {

		return contextId;

	}

	public void setContextId(final String contextId) {

		this.contextId = contextId;

	}

	public String getContextLabel() {

		return contextLabel;

	}

	public void setContextLabel(final String contextLabel) {

		this.contextLabel = contextLabel;

	}

	public String getContextType() {

		return contextType;

	}

	public void setContextType(final String contextType) {

		this.contextType = contextType;

	}

	public String getContextParams() {

		return contextParams;

	}

	public void setContextParams(final String contextParams) {

		this.contextParams = contextParams;

	}

	public String getDashboardVisibility() {

		return dashboardVisibility;

	}

	public void setDashboardVisibility(final String dashboardVisibility) {

		this.dashboardVisibility = dashboardVisibility;

	}

}

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

import java.util.Map;

import java.util.Map.Entry;

import com.google.common.collect.Maps;

import org.dom4j.DocumentHelper;

import org.dom4j.Element;

public class ContextPart {

	private String id;

	private String label;

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

	private Map<String, ContextPart> parts = Maps.newLinkedHashMap();

	public ContextPart() {

	}

	public ContextPart(final String id, final String label) {

		this.id = id;

		this.label = label;

	}

	public String getId() {

		return id;

	}

	public void setId(final String id) {

		this.id = id;

	}

	public String getLabel() {

		return label;

	}

	public void setLabel(final String label) {

		this.label = label;

	}

	public Map<String, String> getParams() {

		return params;

	}

	public void setParams(final Map<String, String> params) {

		this.params = params;

	}

	public Map<String, ContextPart> getParts() {

		return parts;

	}

	public void setParts(final Map<String, ContextPart> parts) {

		this.parts = parts;

	}

	public void addPart(final ContextPart part) {

		if (parts.containsKey(part.getId())) {

			final ContextPart localChild = getParts().get(part.getId());

			for (ContextPart child : part.getParts().values()) {

				localChild.addPart(child);

			}

		} else {

			parts.put(part.getId(), part);

		}

	}

	public Element asDomElement(final String name) {

		final Element elem = DocumentHelper.createElement(name);

		elem.addAttribute("id", id);

		elem.addAttribute("label", label);

		elem.addAttribute("claim", "false");

		for (Entry<String, String> e : params.entrySet()) {

			final Element p = elem.addElement("param");

			p.addAttribute("name", e.getKey());

			p.setText(e.getValue());

		}

		for (ContextPart child : parts.values()) {

			elem.add(child.asDomElement("concept"));

		}

		return elem;

	}

}

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

import java.io.IOException;

import java.io.InputStream;

import java.io.StringReader;

import java.io.StringWriter;

import javax.annotation.Resource;

import javax.xml.transform.Transformer;

import javax.xml.transform.TransformerException;

import javax.xml.transform.TransformerFactory;

import javax.xml.transform.stream.StreamResult;

import javax.xml.transform.stream.StreamSource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.miscutils.functional.hash.Hashing;

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

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.core.io.ClassPathResource;

public class PrepareIndexDataJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	/**