D-Net

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

import java.util.List;

import javax.annotation.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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 GenerateFETH2020ContextJobNode extends SimpleJobNode implements ProgressJobNode {

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

	private String eprParam;

	private ResultsetProgressProvider progressProvider;

	@Resource

	private ResultSetClientFactory resultSetClientFactory;

	@Resource

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	public ProgressProvider getProgressProvider() {

		return this.progressProvider;

	}

	@Override

	protected String execute(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());

		BuildH2020FETTaxonomy builder = new BuildH2020FETTaxonomy();

		builder.setIterator(iter.iterator());

		String taxonomy = builder.parseProjects();

		final String xquery = "for $x in collection('/db/DRIVER/ContextDSResources/ContextDSResourceType') "

				+ "where $x//CONFIGURATION/context[@id='fet-h2020'] return $x//RESOURCE_IDENTIFIER/@value/string()";

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

		if (list.isEmpty()) {

			registerProfile(taxonomy);

		} else {

			deleteProfile(list.get(0));

			registerProfile(taxonomy);

		}

		return Arc.DEFAULT_ARC;

	}

	private void registerProfile(String profile) throws ISRegistryException {

		log.info("registering fet-h2020 profile");

		ISRegistryService is = serviceLocator.getService(ISRegistryService.class);

		String id = is.registerProfile(profile);

		log.info("Generating profile with id " + id);

	}

	private void deleteProfile(String profId) throws ISRegistryException {

		log.info("deleting fet-h2020 profile with id " + profId);

		ISRegistryService is = serviceLocator.getService(ISRegistryService.class);

		is.deleteProfile(profId);

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

	public void setProgressProvider(final ResultsetProgressProvider progressProvider) {

		this.progressProvider = progressProvider;

	}

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

	public ProcessCountingResultSetFactory getProcessCountingResultSetFactory() {

		return processCountingResultSetFactory;

	}

	public void setProcessCountingResultSetFactory(final ProcessCountingResultSetFactory processCountingResultSetFactory) {

		this.processCountingResultSetFactory = processCountingResultSetFactory;

	}

	public UniqueServiceLocator getServiceLocator() {

		return serviceLocator;

	}

	public void setServiceLocator(final UniqueServiceLocator serviceLocator) {

		this.serviceLocator = serviceLocator;

	}

}

<?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>6.3.19</version>

	<scm>

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

	</scm>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

			<version>[2.0.0,3.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>[3.0.0,4.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>[1.0.0-SNAPSHOT,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

			<version>[6.0.0,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-index-solr-common</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>[6.0.0,7.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-solr-client</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-deduplication</artifactId>

			<version>[1.0.0,2.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>[1.0.0,2.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>

	</dependencies>

	<properties>

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

	</properties>

</project>

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;

	/**

	 * Stylesheet which transforms a layout to another stylesheet which converts a input record to a index record.

	 */

	private String layoutToRecordStylesheet;

	private String outputRecordsPathParam;

	private String rottenRecordsPathParam;

	private String hbaseTable;

	private String oafSchemaLocationProperty;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		log.info("start preparing job");

		final String xslt = prepareXslt(env("format", token), env("layout", token));

		token.getEnv().setAttribute("index.xslt", xslt);

		if (!StringUtils.isBlank(getOutputRecordsPathParam())) {

			token.getEnv().setAttribute(getOutputRecordsPathParam(), "/tmp" + getFileName(token, "indexrecords"));

		}

		if (!StringUtils.isBlank(getRottenRecordsPathParam())) {

			token.getEnv().setAttribute(getRottenRecordsPathParam(), "/tmp" + getFileName(token, "rottenrecords"));

		}

		token.getEnv().setAttribute("index.solr.url", getIndexSolrUrlZk());

		token.getEnv().setAttribute("index.solr.collection", getCollectionName(token));

		token.getEnv().setAttribute("index.shutdown.wait.time", getIndexSolrShutdownWait());

		token.getEnv().setAttribute("index.buffer.flush.threshold", getIndexBufferFlushTreshold());

		token.getEnv().setAttribute("index.solr.sim.mode", isFeedingSimulationMode());

		token.getEnv().setAttribute("index.feed.timestamp", DateUtils.now_ISO8601());

		token.getEnv().setAttribute(getOafSchemaLocationProperty(), getPropertyFetcher().getProperty(getOafSchemaLocationProperty()));

		return Arc.DEFAULT_ARC;

	}

	protected String tableName(final NodeToken token) {

		if (token.getEnv().hasAttribute("hbaseTable")) {

			String table = token.getEnv().getAttribute("hbaseTable");

			log.debug("found override value in wfEnv for 'hbaseTable' param: " + table);

			return table;

		}

		return getHbaseTable();

	}

	public String getIndexSolrUrlZk() throws ISLookUpException {

		return getServiceConfigValue(

				"for $x in /RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='IndexServiceResourceType'] return $x//PROTOCOL[./@name='solr']/@address/string()");

	}

	public String getIndexSolrShutdownWait() throws ISLookUpException {

		return queryForServiceProperty("solr:feedingShutdownTolerance");

	}

	public String getIndexBufferFlushTreshold() throws ISLookUpException {

		return queryForServiceProperty("solr:feedingBufferFlushThreshold");

	}

	public String isFeedingSimulationMode() throws ISLookUpException {

		return queryForServiceProperty("solr:feedingSimulationMode");

	}

	private String queryForServiceProperty(final String key) throws ISLookUpException {

		return getServiceConfigValue(

				"for $x in /RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='IndexServiceResourceType'] return $x//SERVICE_PROPERTIES/PROPERTY[./@ key='"

						+ key + "']/@value/string()");

	}

	private String getServiceConfigValue(final String xquery) throws ISLookUpException {

		log.debug("quering for service property: " + xquery);

		final String res = serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(xquery);

		if (StringUtils.isBlank(res)) { throw new IllegalStateException("unable to find unique service property, xquery: " + xquery); }

		return res;

	}

	private String getFileName(final NodeToken token, final String fileNamePrefix) {

		return "/" + fileNamePrefix + "_" + tableName(token) + "_" + token.getEnv().getAttribute("format") + ".seq";

	}

	private String getCollectionName(final NodeToken token) {

		return env("format", token) + "-" + env("layout", token) + "-" + env("interpretation", token);

	}

	private String env(final String s, final NodeToken token) {

		return token.getEnv().getAttribute(s);

	}

	/**

	 * Transforms each OAF record into a index record.

	 *

	 * @param format         format

	 * @param layout         layout

	 * @return resultset with transformed records

	 * @throws ISLookUpException    could happen

	 * @throws IOException          could happen

	 * @throws TransformerException could happen

	 */

	protected String prepareXslt(final String format, final String layout) throws ISLookUpException, IOException, TransformerException {

		final TransformerFactory factory = TransformerFactory.newInstance();

		final Transformer layoutTransformer = factory.newTransformer(new StreamSource(new StringReader(readXslt(getLayoutToRecordStylesheet()))));

		final StreamResult layoutToXsltXslt = new StreamResult(new StringWriter());

		layoutTransformer.setParameter("format", format);

		layoutTransformer.transform(new StreamSource(new StringReader(getLayoutSource(format, layout))), layoutToXsltXslt);

		return new String(Hashing.encodeBase64(layoutToXsltXslt.getWriter().toString()));

	}

	private String readXslt(final String s) throws IOException {

		ClassPathResource resource = new ClassPathResource(s);

		InputStream inputStream = resource.getInputStream();

		return IOUtils.toString(inputStream);

	}

	private String getLayoutSource(final String format, final String layout) throws ISLookUpDocumentNotFoundException, ISLookUpException {

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

				"collection('')//RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'MDFormatDSResourceType' and .//NAME='" + format + "']//LAYOUT[@name='" + layout

						+ "']");

	}

	public String getLayoutToRecordStylesheet() {

		return layoutToRecordStylesheet;

	}

	public void setLayoutToRecordStylesheet(final String layoutToRecordStylesheet) {

		this.layoutToRecordStylesheet = layoutToRecordStylesheet;

	}

	public String getHbaseTable() {

		return hbaseTable;

	}

	@Required

	public void setHbaseTable(final String hbaseTable) {

		this.hbaseTable = hbaseTable;

	}

	public String getOutputRecordsPathParam() {

		return outputRecordsPathParam;

	}

	public void setOutputRecordsPathParam(final String outputRecordsPathParam) {

		this.outputRecordsPathParam = outputRecordsPathParam;

	}

	public String getRottenRecordsPathParam() {

		return rottenRecordsPathParam;

	}

	public void setRottenRecordsPathParam(final String rottenRecordsPathParam) {

		this.rottenRecordsPathParam = rottenRecordsPathParam;

	}

	public String getOafSchemaLocationProperty() {

		return oafSchemaLocationProperty;

	}

	public void setOafSchemaLocationProperty(final String oafSchemaLocationProperty) {

		this.oafSchemaLocationProperty = oafSchemaLocationProperty;

	}

}

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

import java.util.Queue;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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;

/**

 * SwitchIndexesJobNode guides the index switches on all search services whose ids are in the queue available in the parameter named 'searchService_ids'.

 *

 * @author claudio, alessia

 * @see eu.dnetlib.msro.openaireplus.workflows.nodes.index.FindSearchServicesJobNode

 * @see eu.dnetlib.msro.openaireplus.workflows.nodes.index.SwitchIndexJobNode

 */

public class SwitchIndexesJobNode extends SimpleJobNode {

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

	@Override

	protected String execute(final NodeToken token) throws Exception {

		Queue<String> q = (Queue<String>) token.getEnv().getTransientAttribute("searchService_ids");

		log.debug("Got the searchService_ids queue: " + q.toString());

		if (q == null) throw new MSROException("Transient param 'searchService_ids' with queue of string could not be found");

		if (q.isEmpty()) {

			log.info("searchService_ids queue consumed, now ending cycle");

			return Arc.DEFAULT_ARC;

		} else {

			//we have something to do: setting the xqueryForServiceIdParam for the SwitchIndexJobNode

			String id = q.poll();

			log.debug("Polled id: " + id);

			log.debug("And now the queue is " + q);

			token.getEnv().setAttribute("search_service_ID", id);

			log.debug("Asking to switch on profile with id: " + id);

			//updating the queue for next iteration

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

			return "switch";

		}

	}

}

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

import com.googlecode.sarasvati.NodeToken;

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

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;

/**

 * SwitchSearchServiceJobNode switches the index used by a SearchService whose profile id is in 'search_service_ID' env param.

 *

 * @author claudio, alessia

 */

public class SwitchSearchServiceJobNode extends BlackboardJobNode {

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

	private static final String BB_ACTION_SWITCH_INDEX = "UpdateIndex";

	private String inputIndexIdParam;

	private String outputIndexIdParam;

	@Override

	protected String obtainServiceId(final NodeToken token) {

		final String id = token.getEnv().getAttribute("search_service_ID");

		if (StringUtils.isBlank(id)) throw new RuntimeException("No id found in env attribute 'search_service_ID'");

		else return id;

	}

	@Override

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

		job.setAction(BB_ACTION_SWITCH_INDEX);

		String indexId = token.getEnv().getAttribute(getInputIndexIdParam());

		checkParam(getInputIndexIdParam(), "output indexId param is missing");

		checkParam(indexId, "indexId is required to perform switch");

		log.info("Switching " + obtainServiceId(token) + " to index " + indexId);

		job.getParameters().put(getOutputIndexIdParam(), indexId);

	}

	private void checkParam(final String param, final String msg) throws MSROException {

		if (StringUtils.isBlank(param)) throw new MSROException(msg);

	}

	public String getInputIndexIdParam() {

		return inputIndexIdParam;

	}

	public void setInputIndexIdParam(final String inputIndexIdParam) {

		this.inputIndexIdParam = inputIndexIdParam;

	}

	public String getOutputIndexIdParam() {

		return outputIndexIdParam;

	}

	public void setOutputIndexIdParam(final String outputIndexIdParam) {

		this.outputIndexIdParam = outputIndexIdParam;

	}

}

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

import java.util.List;

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.tools.blackboard.BlackboardJob;

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;

/**

 * SwitchIndexJobNode performs index switch.

 *

 * @author claudio

 * @see eu.dnetlib.msro.openaireplus.workflows.nodes.index.SwitchSearchServiceJobNode

 * @see eu.dnetlib.msro.openaireplus.workflows.nodes.index.SwitchIndexesJobNode

 * @deprecated

 */

@Deprecated

public class SwitchIndexJobNode extends BlackboardJobNode {

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

	private static final String BB_ACTION_SWITCH_INDEX = "UpdateIndex";

	private String inputIndexIdParam;

	private String outputIndexIdParam;

	private String xqueryForServiceIdParam;

	@Override

	protected String obtainServiceId(final NodeToken token) {

		final String xquery = token.getEnv().getAttribute(getXqueryForServiceIdParam());

		List<String> searchServiceIds;

		try {

			searchServiceIds = getServiceLocator().getService(ISLookUpService.class).quickSearchProfile(xquery);

			if (searchServiceIds.size() > 1) throw new RuntimeException("Too many SearchService ids found using query: " + xquery);

			if (searchServiceIds.size() < 1) throw new RuntimeException("SearchService id not found using query: " + xquery);

			return searchServiceIds.get(0);

		} catch (ISLookUpException e) {

			throw new RuntimeException(e);

		}

	}

	@Override

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

		job.setAction(BB_ACTION_SWITCH_INDEX);

		String indexId = token.getEnv().getAttribute(getInputIndexIdParam());

		checkParam(getInputIndexIdParam(), "output indexId param is missing");

		checkParam(indexId, "indexId is required to perform switch");

		log.info("Switching " + obtainServiceId(token) + " to index " + indexId);

		job.getParameters().put(getOutputIndexIdParam(), indexId);

	}

	private void checkParam(final String param, final String msg) throws MSROException {

		if (StringUtils.isBlank(param)) throw new MSROException(msg);

	}

	public String getInputIndexIdParam() {

		return inputIndexIdParam;

	}

	public void setInputIndexIdParam(final String inputIndexIdParam) {

		this.inputIndexIdParam = inputIndexIdParam;

	}

	public String getOutputIndexIdParam() {

		return outputIndexIdParam;

	}

	public void setOutputIndexIdParam(final String outputIndexIdParam) {

		this.outputIndexIdParam = outputIndexIdParam;

	}

	public String getXqueryForServiceIdParam() {

		return xqueryForServiceIdParam;

	}

	public void setXqueryForServiceIdParam(final String xqueryForServiceIdParam) {

		this.xqueryForServiceIdParam = xqueryForServiceIdParam;

	}

}

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

import java.util.List;

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.miscutils.datetime.DateUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

public class IncrementalTransformationJobNode extends IncrementalOperationJobNode {

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

    @Autowired

    private UniqueServiceLocator locator;

    private static final DateUtils dateUtils = new DateUtils();

    @Override

    protected boolean forceRefresh(final NodeToken nodeToken, final Long lastSuccessEndDate, final String currentWfProfileId) throws Exception {

        if (lastSuccessEndDate < 0) {

            nodeToken.getFullEnv().setAttribute("OperationTypeInfo", "Last success date < 0, transformation forced to REFRESH");

            nodeToken.getFullEnv().setAttribute("operationType", "REFRESH");

            return true;

        }

        String trRuleId = getTransformationId(currentWfProfileId);

        final long lastUpdateDate = getLastTransformationRuleUpdate(trRuleId);

        log.info(String.format("Last update date of the transformation rule with id %s is %s", trRuleId, DateUtils.calculate_ISO8601(lastUpdateDate)));

        if (lastUpdateDate > lastSuccessEndDate) {

            nodeToken.getFullEnv().setAttribute("OperationTypeInfo", "Transformation Rule has been updated, transformation forced to REFRESH");

            nodeToken.getFullEnv().setAttribute("operationType", "REFRESH");

            return true;

        }

        return false;

    }

    private String getTransformationId(final String workflowId) throws ISLookUpException {

        final String query="for $x in collection('/db/DRIVER/WorkflowDSResources/WorkflowDSResourceType') where $x//RESOURCE_IDENTIFIER/@value='%s' " +

                "return $x//PARAM[./@category='TRANSFORMATION_RULE_ID']/text()";

        final ISLookUpService lookUpService = locator.getService(ISLookUpService.class);

        final String queryInstance = String.format(query, workflowId);

        log.debug("Query to find the Transformation Rule");

        List<String> transformationId = lookUpService.quickSearchProfile(queryInstance);

        if(transformationId== null || transformationId.isEmpty())

            throw new RuntimeException("Error unable to find the Transformation rule ID on workflow profile "+workflowId);

        return transformationId.get(0);

    }

    private Long getLastTransformationRuleUpdate(final String trId) throws ISLookUpException {

        final String query = "for $x in collection('/db/DRIVER/TransformationRuleDSResources/TransformationRuleDSResourceType') where $x//RESOURCE_IDENTIFIER/@value='%s' return $x//DATE_OF_CREATION/@value/string()";

        log.debug("retrieve creation date from transformation ID "+trId);

        final ISLookUpService lookUpService = locator.getService(ISLookUpService.class);

        final String queryInstance = String.format(query, trId);

        log.debug("Query to find the Transformation Rule");

        List<String> currentDate = lookUpService.quickSearchProfile(queryInstance);

        if(currentDate== null || currentDate.isEmpty())

            throw new RuntimeException("Error unable to find the creation date of the  Transformation rule "+trId);

        return dateUtils.parse(currentDate.get(0)).getTime();

    }

}

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

import java.io.StringReader;

import javax.annotation.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

public class RetrieveInterfaceInfoJobNode extends SimpleJobNode {

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		String datasourceId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ID);

		String interfaceID = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);

		final String profile = serviceLocator.getService(ISLookUpService.class).getResourceProfile(datasourceId);

		final Document doc = new SAXReader().read(new StringReader(profile));

		final Node ifcNode = doc.selectSingleNode("//INTERFACE[@id='" + interfaceID + "']");

		String contentDescription = ifcNode.valueOf("./@contentDescription");

		token.getEnv().setAttribute("objectStoreContentDescription", contentDescription);

		final Node acProtNode = doc.selectSingleNode("//INTERFACE[@id='" + interfaceID + "']/ACCESS_PROTOCOL");

		String basePath = acProtNode.valueOf("./@basePath");

		token.getEnv().setAttribute("objectStoreBasePath", basePath);

		return Arc.DEFAULT_ARC;

	}

}

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

import javax.annotation.Resource;

import com.google.common.base.Joiner;

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.functional.xml.IndentXmlString;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

public class ContextLoader {

	/**

	 * logger.

	 */

	private static final Log log = LogFactory.getLog(ContextLoader.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 ContextDSResources: " + getXquery());

		StringBuilder sb = new StringBuilder("<ContextDSResources>");

		Joiner.on("").appendTo(sb, serviceLocator.getService(ISLookUpService.class).quickSearchProfile(getXquery()));

		sb.append("</ContextDSResources>");

		log.debug("got ContextDSResources: \n" + IndentXmlString.apply(sb.toString()));

		return sb.toString();

	}

	public String getXquery() {

		return xquery;

	}

	@Required

	public void setXquery(final String xquery) {

		this.xquery = xquery;

	}

}

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

import java.util.Iterator;

import java.util.Map;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.common.logging.DnetLogger;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.rmi.MSROException;

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

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

import org.apache.commons.lang.math.NumberUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

public class IncrementalOperationJobNode extends SimpleJobNode {

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

    //incremental or refresh

    private String operationType;

    @Autowired

    private DnetLogger dnetLogger;

    @Override

    protected String execute(NodeToken nodeToken) throws Exception {

        if ("incremental".equalsIgnoreCase(operationType)) {

            final String currentWfProfileId = findCurrentWfProfileId(nodeToken);

            final Long lastSuccessEndDate = findLastSuccessEndDate(currentWfProfileId);

            if(forceRefresh(nodeToken, lastSuccessEndDate, currentWfProfileId)) return Arc.DEFAULT_ARC;

            log.info("Last success date "+ lastSuccessEndDate);

            nodeToken.getFullEnv().setAttribute("OperationTypeInfo", "Operation type set to INCREMENTAL with date "+DateUtils.calculate_ISO8601(lastSuccessEndDate));

            nodeToken.getFullEnv().setAttribute("operationType", "INCREMENTAL");

            nodeToken.getFullEnv().setAttribute("DateFromFilter", lastSuccessEndDate);            

            return Arc.DEFAULT_ARC;

        }

        nodeToken.getFullEnv().setAttribute("operationType", "REFRESH");

        nodeToken.getFullEnv().setAttribute("OperationTypeInfo", "Operation type manually set to REFRESH");

        return Arc.DEFAULT_ARC;

    }

    protected boolean forceRefresh(final NodeToken nodeToken, final Long lastSuccessEndDate, final String currentWfProfileId) throws Exception {

        if (lastSuccessEndDate < 0) {

            nodeToken.getFullEnv().setAttribute("OperationTypeInfo", "Last success date < 0, operation forced to REFRESH");

            nodeToken.getFullEnv().setAttribute("operationType", "REFRESH");

            return true;

        }

        return false;

    }

    private Long findLastSuccessEndDate(String profId) {

        long res = -1;

        final Iterator<Map<String, String>> iter = dnetLogger.find(WorkflowsConstants.SYSTEM_WF_PROFILE_ID, profId);

        while (iter.hasNext()) {

            final Map<String, String> map = iter.next();

            log.debug("Iterating on the logs");

            if ("true".equalsIgnoreCase(map.get(WorkflowsConstants.SYSTEM_COMPLETED_SUCCESSFULLY))) {

                final long curr = NumberUtils.toLong(map.get(WorkflowsConstants.SYSTEM_END_DATE), -1);

                if (curr > res) {

                    res = curr;

                }

            }

        }

        return res;

    }

    private String findCurrentWfProfileId(NodeToken token) throws MSROException {

        log.debug("Start to find the current profile Id");

        final String p1 = token.getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

        if (p1 != null && !p1.isEmpty()) {

            log.debug("The profile Id found is "+p1);

            return p1;

        }

        final String p2 = token.getFullEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

        if (p2 != null && !p2.isEmpty()) {

            log.debug("The profile Id found is "+p2);

            return p2;

        }

        final String p3 = token.getProcess().getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

        if (p3 != null && !p3.isEmpty()) {

            log.debug("The profile Id found is "+p3);

            return p3;

        }

        throw new MSROException("Missing property in env: " + WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

    }

    public String getOperationType() {

        return operationType;

    }

    public void setOperationType(final String operationType) {

        this.operationType = operationType;

    }

}

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

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

public class DeleteOpenaireMetaWfJobNode extends DeleteMetaWfJobNode {

	@Autowired

	private UniqueServiceLocator serviceLocator;

	@Autowired

	private LocalDatasourceManager dsManager;

	@Override

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

		final String openaireDsId = serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(

				"/*[.//RESOURCE_IDENTIFIER/@value = '" + dsId + "']//FIELD/value[../key='OpenAireDataSourceId']/text()");

		if (openaireDsId.equals("openaire____::bootstrap")) {

			dsManager.setActive(dsId, ifaceId, active);

		} else {

			dsManager.setActive(openaireDsId, ifaceId, active);

		}

	}

}

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

import java.io.ByteArrayInputStream;

import java.io.InputStream;

import java.io.StringWriter;

import java.nio.charset.StandardCharsets;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import java.util.Queue;

import javax.xml.XMLConstants;

import javax.xml.parsers.DocumentBuilderFactory;

import javax.xml.transform.Transformer;

import javax.xml.transform.TransformerFactory;

import javax.xml.transform.dom.DOMSource;

import javax.xml.transform.stream.StreamResult;

import javax.xml.xpath.XPath;

import javax.xml.xpath.XPathConstants;

import javax.xml.xpath.XPathFactory;

import com.google.common.collect.Maps;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.w3c.dom.*;

/**

 * The Class SplitterDatasetsIterator.

 */

public class SplitterDatasetsIterator {

	/**

	 * The Constant log.

	 */

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

	/**

	 * The end queue.

	 */

	public static String END_QUEUE = "END_QUEUE";

	/**

	 * The publications.

	 */

	private Queue<String> publications;

	/**

	 * The input epr.

	 */

	private Iterable<String> inputEPR;

	/**

	 * The root name.

	 */

	private String rootName;

	/**

	 * Instantiates a new splitter datasets iterator.

	 *

	 * @param publicationsQueue the publications queue

	 * @param inputEPR          the input epr

	 * @param rootName          the root name

	 */

	public SplitterDatasetsIterator(final Queue<String> publicationsQueue, final Iterable<String> inputEPR, final String rootName) {

		this.publications = publicationsQueue;

		this.inputEPR = inputEPR;

		this.rootName = rootName;

	}

	/**

	 * Populate queues.

	 */

	public void populateQueues() {

		if (this.inputEPR == null) return;

		for (String inputXML : inputEPR) {

			final ByteArrayInputStream bais = new ByteArrayInputStream(inputXML.getBytes(StandardCharsets.UTF_8));

			final List<String> publication_extracted = extractByTag(bais, "publication");

			if (publication_extracted != null) {

				publications.addAll(publication_extracted);

			}

		}

		publications.add(END_QUEUE);

	}

	/**

	 * Extract by tag.

	 *

	 * @param inputXML the input xml

	 * @param tag      the tag

	 * @return the list

	 */

	private List<String> extractByTag(final InputStream inputXML, final String tag) {

		try {

			DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();

			Document doc = dbf.newDocumentBuilder().parse(inputXML);

			XPath xpath = XPathFactory.newInstance().newXPath();

			Node rootNode = (Node) xpath.evaluate("//*[local-name()='" + this.rootName + "']", doc, XPathConstants.NODE);

			NamedNodeMap attributes = rootNode.getAttributes();

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

			for (int i = 0; i < attributes.getLength(); i++) {

				Node node = attributes.item(i);

				String name = node.getNodeName();

				if (name.startsWith("xmlns:")) {

					nameSpaces.put(StringUtils.substringAfter(name, "xmlns:"), node.getNodeValue());

				}

			}

			xpath = XPathFactory.newInstance().newXPath();

			NodeList nodes = (NodeList) xpath.evaluate("//*[local-name()='" + tag + "']/*[local-name()='record']", doc, XPathConstants.NODESET);

			if ((nodes != null) && (nodes.getLength() > 0)) {

				List<String> result = new ArrayList<>();

				for (int i = 0; i < nodes.getLength(); i++) {

					Document currentDoc = dbf.newDocumentBuilder().newDocument();

					Node imported = currentDoc.importNode(nodes.item(i), true);

					for (String key : nameSpaces.keySet()) {

						Element element = (Element) imported;

						element.setAttributeNS(XMLConstants.XMLNS_ATTRIBUTE_NS_URI, "xmlns:" + key, nameSpaces.get(key));

					}

					Transformer transformer = TransformerFactory.newInstance().newTransformer();

					DOMSource mydoc = new DOMSource(imported);

					StringWriter writer = new StringWriter();

					transformer.transform(mydoc, new StreamResult(writer));

					String record = writer.toString();

					result.add(record);

				}

				return result;

			}

		} catch (Exception e) {

			log.error("Error on extracting " + tag, e);

			return null;

		}

		return null;

	}

}

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

import java.io.IOException;

import java.io.StringReader;

import java.util.Date;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Objects;

import java.util.Set;

import java.util.stream.Collectors;

import javax.annotation.Resource;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.math.NumberUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.io.SAXReader;

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.mdstore.MDStoreServiceException;

import eu.dnetlib.data.mdstore.modular.connector.MDStoreDao;

import eu.dnetlib.data.objectstore.modular.connector.ObjectStoreDao;

import eu.dnetlib.data.objectstore.rmi.ObjectStoreServiceException;

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.enabling.is.lookup.rmi.ISLookUpException;

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

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

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

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

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

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

public class FixRepoMdstoreSizesJobNode extends SimpleJobNode implements ProgressJobNode {

	@Autowired

	private LocalOpenaireDatasourceManager dsManager;

	@Autowired

	private UniqueServiceLocator serviceLocator;

	@Resource(name = "mongodbMDStoreDao")

	private MDStoreDao mdstoreDao;

	@Autowired

	private ObjectStoreDao objectStoreDao;

	private final DateUtils dateUtils = new DateUtils();

	private int current = 0;

	private int total = 0;

	private ISRegistryService registry;

	private ISLookUpService lookup;

	private final Map<String, String> openaireIds = new HashMap<>();

	private boolean alwaysUpdate = false;

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

	public void init(final int total) {

		this.current = 0;

		this.total = total;

		this.lookup = serviceLocator.getService(ISLookUpService.class);

		this.registry = serviceLocator.getService(ISRegistryService.class);

		try {

			openaireIds.putAll(lookup.quickSearchProfile(

					"for $x in collection('/db/DRIVER/RepositoryServiceResources/RepositoryServiceResourceType') return concat($x//DATASOURCE_ORIGINAL_ID, ' @@@ ', $x//RESOURCE_IDENTIFIER/@value)")

					.stream()

					.collect(Collectors.toMap(

							s -> StringUtils.substringBefore(s, "@@@").trim(),

							s -> StringUtils.substringAfter(s, "@@@").trim())));

		} catch (final ISLookUpException e) {

			// TODO Auto-generated catch block

			e.printStackTrace();

		}

	}

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final Set<String> list = dsManager.listManagedDatasourceIds();

		init(list.size());

		for (final String dsId : list) {

			log.info("Processing ds: " + dsId);

			current++;

			try {

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

					verifyApi(dsId, api);

				}

			} catch (final Throwable e) {

				log.error("Error processing ds: " + dsId, e);

				token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_ERROR + "::" + dsId, e.getMessage());

			}

		}

		return Arc.DEFAULT_ARC;

	}

	private void verifyApi(final String dsId, final Api<ApiParam> api)

			throws DsmException, ISRegistryException, IOException, ISLookUpException, MDStoreServiceException, ObjectStoreServiceException {

		for (final Document doc : listCollectionMdStores(dsId, api.getId())) {

			final String mdId = doc.valueOf("//RESOURCE_IDENTIFIER/@value");

			final int actualSize = NumberUtils.toInt(doc.valueOf("//NUMBER_OF_RECORDS"), 0);

			final int size = mdstoreDao.getMDStore(mdId).getSize();

			if (alwaysUpdate || size != actualSize) {

				log.info("  -- Updating size of api " + api.getId() + ", new value = " + size);

				updateMdStoreProfile(mdId, doc, size);

				dsManager.setLastCollectionInfo(dsId, api.getId(), mdId, size, calculateLastDate(doc));

			}

		}

		for (final Document doc : listTransformationMdStores(dsId, api.getId())) {

			final String mdId = doc.valueOf("//RESOURCE_IDENTIFIER/@value");

			final int actualSize = NumberUtils.toInt(doc.valueOf("//NUMBER_OF_RECORDS"), 0);

			final int size = mdstoreDao.getMDStore(mdId).getSize();

			if (alwaysUpdate || size != actualSize) {

				updateMdStoreProfile(mdId, doc, size);

				dsManager.setLastAggregationInfo(dsId, api.getId(), mdId, size, calculateLastDate(doc));

			}

		}

		for (final Document doc : listDownloadObjectStores(dsId, api.getId())) {

			final String objId = doc.valueOf("//RESOURCE_IDENTIFIER/@value");

			final int actualSize = NumberUtils.toInt(doc.valueOf("//STORE_SIZE"), 0);

			final int size = objectStoreDao.getObjectStore(objId).getSize();

			if (alwaysUpdate || size != actualSize) {

				updateObjStoreProfile(objId, doc, size);

				dsManager.setLastDownloadInfo(dsId, api.getId(), objId, size, calculateLastDate(doc));

			}

		}

	}

	private Date calculateLastDate(final Document doc) {

		final String dateS = doc.valueOf("//LAST_STORAGE_DATE");

		final Date date = StringUtils.isNoneBlank(dateS) ? dateUtils.parse(dateS) : new Date();

		return date;

	}

	private List<Document> listCollectionMdStores(final String dsId, final String apiId) throws ISLookUpException, IOException {

		return executeXquery("listCollectionMdStores.xquery.st", dsId, apiId);

	}

	private List<Document> listTransformationMdStores(final String dsId, final String apiId) throws ISLookUpException, IOException {

		return executeXquery("listTransformationMdStores.xquery.st", dsId, apiId);

	}

	private List<Document> listDownloadObjectStores(final String dsId, final String apiId) throws ISLookUpException, IOException {

		return executeXquery("listDownloadObjectStores.xquery.st", dsId, apiId);

	}

	private List<Document> executeXquery(final String template, final String dsId, final String apiId) throws ISLookUpException, IOException {

		final StringTemplate st = new StringTemplate(IOUtils.toString(getClass().getResourceAsStream(template)));

		st.setAttribute("dsId", openaireIds.get(dsId));

		st.setAttribute("apiId", apiId);

		final SAXReader reader = new SAXReader();

		return lookup.quickSearchProfile(st.toString())

				.stream()

				.map(s -> {

					try {

						return reader.read(new StringReader(s));

					} catch (final DocumentException e) {

						return null;

					}

				})

				.filter(Objects::nonNull)

				.collect(Collectors.toList());

	}

	private void updateMdStoreProfile(final String mdId, final Document doc, final int size) throws ISRegistryException {

		doc.selectSingleNode("//NUMBER_OF_RECORDS").setText(Integer.toString(size));

		registry.updateProfile(mdId, doc.asXML(), "MDStoreDSResourceType");

	}

	private void updateObjStoreProfile(final String objId, final Document doc, final int size) throws ISRegistryException {

		doc.selectSingleNode("//COUNT_STORE").setText(Integer.toString(size));

		doc.selectSingleNode("//STORE_SIZE").setText(Integer.toString(size));

		registry.updateProfile(objId, doc.asXML(), "ObjectStoreDSResourceType");

	}

	public boolean isAlwaysUpdate() {

		return alwaysUpdate;

	}

	public void setAlwaysUpdate(final boolean alwaysUpdate) {

		this.alwaysUpdate = alwaysUpdate;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return new ProgressProvider() {

			@Override

			public int getTotalValue() {

				return total;

			}

			@Override

			public int getCurrentValue() {

				return current;

			}

			@Override

			public boolean isInaccurate() {

				return false;

			}

		};

	}

}

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

import java.io.StringReader;

import java.util.Map;

import eu.dnetlib.miscutils.functional.UnaryFunction;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.Element;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

/**

 * The Class PatchHostedBy.

 */

public class PatchHostedBy implements UnaryFunction<String, String> {

	/**

	 * The set spec hosted by map.

	 */

	private Map<String, HostedByEntry> setSpecHostedByMap;

	/**

	 * The counters.

	 */

	private HostedByCounters counters;

	/**

	 * The xpath.

	 */

	private String xpath;

	/**

	 * The reader.

	 */

	private final SAXReader reader = new SAXReader();

	/**

	 * The Constant log.

	 */

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

	/**

	 * Instantiates a new patch hosted by.

	 *

	 * @param setSpecHostedByMap the set spec hosted by map

	 * @param xpath              the xpath

	 * @param counters           the counters

	 */

	public PatchHostedBy(final Map<String, HostedByEntry> setSpecHostedByMap, final String xpath, final HostedByCounters counters) {

		this.setSpecHostedByMap = setSpecHostedByMap;

		this.xpath = xpath;

		this.counters = counters;

		if (log.isDebugEnabled()) {

			log.debug("****************************************");

			log.debug("SetSpec/hostedBy map:");