D-Net

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

import java.io.StringReader;

import java.util.Map;

import javax.annotation.Resource;

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

import com.google.common.collect.Maps;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.datasources.rmi.DatasourceManagerService;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

public class MDStoreToApiExtraFieldJobNode extends SimpleJobNode {

	private String mdId;

	private String datasourceId;

	private String datasourceInterface;

	private String extraFieldForTotal;

	private String extraFieldForDate;

	private String extraFieldForMdId;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String xq = "for $x in collection('/db/DRIVER/MDStoreDSResources/MDStoreDSResourceType') " +

				"where $x//RESOURCE_IDENTIFIER/@value='" + mdId + "' " +

				"return concat($x//NUMBER_OF_RECORDS, ' @=@ ', $x//LAST_STORAGE_DATE)";

		final String s = serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(xq);

		final String[] arr = s.split(" @=@ ");

		final Map<String, String> map = getCurrentExtraFields(datasourceId, datasourceInterface);

		map.put(extraFieldForTotal, arr[0].trim());

		map.put(extraFieldForDate, arr[1].trim());

		map.put(extraFieldForMdId, mdId);

		serviceLocator.getService(DatasourceManagerService.class).bulkUpdateApiExtraFields(datasourceId, datasourceInterface, map);

		return Arc.DEFAULT_ARC;

	}

	private Map<String, String> getCurrentExtraFields(final String repoId, final String ifaceId) throws Exception {

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

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

		final SAXReader reader = new SAXReader();

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

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

		if (ifcNode != null) {

			for (Object o : ifcNode.selectNodes("./INTERFACE_EXTRA_FIELD")) {

				res.put(((Node) o).valueOf("@name"), ((Node) o).getText());

			}

		}

		return res;

	}

	public String getMdId() {

		return mdId;

	}

	public void setMdId(final String mdId) {

		this.mdId = mdId;

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

	public String getExtraFieldForTotal() {

		return extraFieldForTotal;

	}

	public void setExtraFieldForTotal(final String extraFieldForTotal) {

		this.extraFieldForTotal = extraFieldForTotal;

	}

	public String getExtraFieldForDate() {

		return extraFieldForDate;

	}

	public void setExtraFieldForDate(final String extraFieldForDate) {

		this.extraFieldForDate = extraFieldForDate;

	}

	public String getExtraFieldForMdId() {

		return extraFieldForMdId;

	}

	public void setExtraFieldForMdId(final String extraFieldForMdId) {

		this.extraFieldForMdId = extraFieldForMdId;

	}

}

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

import javax.annotation.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.database.rmi.DatabaseService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

import org.apache.commons.lang.StringUtils;

public class ExecuteSqlFromEnvJobNode extends AsyncJobNode {

	private String db;

	private String dbParam;

	private String dbProperty;

	private String sqlParamName;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		String sql = token.getEnv().getAttribute(sqlParamName);

		if (StringUtils.isBlank(sql)) throw new IllegalArgumentException("Missing value in env attribute named: " + sqlParamName);

		serviceLocator.getService(DatabaseService.class).updateSQL(findDb(token), sql);

		return Arc.DEFAULT_ARC;

	}

	private String findDb(final NodeToken token) {

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

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

		} else if (dbProperty != null && !dbProperty.isEmpty()) {

			return getPropertyFetcher().getProperty(dbProperty);

		} else {

			return db;

		}

	}

	public String getDb() {

		return db;

	}

	public void setDb(final String db) {

		this.db = db;

	}

	public String getDbParam() {

		return dbParam;

	}

	public void setDbParam(final String dbParam) {

		this.dbParam = dbParam;

	}

	public String getDbProperty() {

		return dbProperty;

	}

	public void setDbProperty(final String dbProperty) {

		this.dbProperty = dbProperty;

	}

	public String getSqlParamName() {

		return sqlParamName;

	}

	public void setSqlParamName(final String sqlParamName) {

		this.sqlParamName = sqlParamName;

	}

}

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

import javax.annotation.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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.SimpleJobNode;

// TODO: Auto-generated Javadoc

/**

 * The Class UpdateObjectStoreSizeJobNode.

 */

public class UpdateObjectStoreSizeJobNode extends SimpleJobNode {

	/** The object store id. */

	private String objectStoreIdParam;

	/** The service locator. */

	@Resource

	private UniqueServiceLocator serviceLocator;

	/*

	 * (non-Javadoc)

	 * 

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

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

		int size = serviceLocator.getService(ObjectStoreService.class, objectStoreIdParam).getSize(objectStoreIdParam);

		String now = DateUtils.now_ISO8601();

		String mdstoreXUpdate = "for $x in //RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '" + objectStoreIdParam + "']"

				+ "return update value $x//LAST_STORAGE_DATE with '" + now + "'";

		registry.executeXUpdate(mdstoreXUpdate);

		String mdstoreNumberXUpdate = "for $x in //RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '" + objectStoreIdParam + "']"

				+ "return update value $x//COUNT_STORE with '" + size + "'";

		registry.executeXUpdate(mdstoreNumberXUpdate);

		mdstoreNumberXUpdate = "for $x in //RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '" + objectStoreIdParam + "']"

				+ "return update value $x//STORE_SIZE with '" + size + "'";

		registry.executeXUpdate(mdstoreNumberXUpdate);

		return Arc.DEFAULT_ARC;

	}

	/**

	 * Gets the object store id param.

	 *

	 * @return the objectStoreIdParam

	 */

	public String getObjectStoreIdParam() {

		return objectStoreIdParam;

	}

	/**

	 * Sets the object store id param.

	 *

	 * @param objectStoreIdParam

	 *            the new object store id param

	 */

	public void setObjectStoreIdParam(final String objectStoreIdParam) {

		this.objectStoreIdParam = objectStoreIdParam;

	}

}

<?xml version="1.0" encoding="UTF-8"?>

<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>dnet-parent</artifactId>

		<version>1.0.0</version>

		<relativePath />

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

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

	<packaging>jar</packaging>

	<version>3.1.21</version>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-msro-service/tags/dnet-msro-service-3.1.21</developerConnection>

	</scm>

	<dependencies>

		<dependency>

			<groupId>opensymphony</groupId>

			<artifactId>quartz</artifactId>

			<version>1.6.6</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

			<version>[2.0.0,3.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>cnr-data-utility-cleaner-rmi</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-resultset-client</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-blackboard-common</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-mongo-logging</artifactId>

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

		</dependency>

		<dependency>

			<groupId>com.googlecode</groupId>

			<artifactId>sarasvati</artifactId>

			<version>1.0.3</version>

		</dependency>

		<dependency>

			<groupId>com.googlecode</groupId>

			<artifactId>sarasvati-visual</artifactId>

			<version>1.0.3</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-modular-collector-service-rmi</artifactId>

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

		</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>dnet-objectstore-rmi</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-data-transformation-service-rmi</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-data-provision-rmi</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-runtime</artifactId>

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

		</dependency>

		<dependency>

			<groupId>javax.mail</groupId>

			<artifactId>mail</artifactId>

			<version>1.4</version>

		</dependency>

		<dependency>

			<groupId>org.codehaus.groovy</groupId>

			<artifactId>groovy-all</artifactId>

			<version>2.1.6</version>

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

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

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>joda-time</groupId>

			<artifactId>joda-time</artifactId>

			<version>2.3</version>

		</dependency>

	</dependencies>

	<properties>

		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

	</properties>

</project>

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

import java.io.IOException;

import java.net.URLEncoder;

import java.util.List;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import com.googlecode.sarasvati.Arc;

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.enabling.resultset.XSLTMappedResultSetFactory;

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

import eu.dnetlib.msro.rmi.MSROException;

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

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.core.io.Resource;

public class MdBuilderJobNode extends SimpleJobNode {

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

	private Resource mdBuilderTemplateXslt;

	private String inputEprParam;

	private String outputEprParam;

	private String datasourceId;

	private String datasourceInterface;

	private XSLTMappedResultSetFactory xsltMappedResultSetFactory;

	@javax.annotation.Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws MSROException {

		final String inputEpr = token.getEnv().getAttribute(inputEprParam);

		if ((inputEpr == null) || inputEpr.isEmpty()) throw new MSROException("InputEprParam (" + inputEprParam + ") not found in ENV");

		StringTemplate st = null;

		try {

			st = new StringTemplate(IOUtils.toString(getMdBuilderTemplateXslt().getInputStream()));

			st.setAttribute("datasourceId", datasourceId);

			st.setAttribute("xpath", getMetadataIdentifierPath().replace("\"", "'"));

			st.setAttribute("baseurl", URLEncoder.encode(getBaseUrl(), "UTF-8"));

			st.setAttribute("metadatanamespace", getMetadataNamespace());

			/*

			 * If namespacePrefix has been already pushed to env by some custom JobNode e.g. ObtainOpenaireDataSourceParamsJobNode then push

			 * it to ST. Else: a) try to get it from EXTRAFIELDS of the datasource b) try to get it from DATASOURCE_ORIGINAL_ID of the

			 * datasource c) if any of the is present, then push to ST the datasourceId

			 */

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

				st.setAttribute("namespacePrefix", token.getEnv().getAttribute("namespacePrefix"));

			} else {

				List<String> namespacePrefix;

				String xQuery = "/*[.//RESOURCE_IDENTIFIER/@value='" + datasourceId + "']//EXTRA_FIELDS/FIELD/value[../key='NamespacePrefix']/string()";

				namespacePrefix = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xQuery);

				if (namespacePrefix.size() != 0) {

					st.setAttribute("namespacePrefix", namespacePrefix.get(0));

				} else {

					xQuery = "/*[.//RESOURCE_IDENTIFIER/@value='" + datasourceId + "']//DATASOURCE_ORIGINAL_ID/string()";

					namespacePrefix = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xQuery);

					if (namespacePrefix.size() != 0) {

						st.setAttribute("namespacePrefix", namespacePrefix.get(0));

					} else {

						st.setAttribute("namespacePrefix", datasourceId);

					}

				}

			}

			final W3CEndpointReference epr = xsltMappedResultSetFactory.createMappedResultSet(new EPRUtils().getEpr(inputEpr), st.toString());

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

			return Arc.DEFAULT_ARC;

		} catch (ISLookUpException e) {

			throw new MSROException("Error while initializing mdBuilder template (" + getMdBuilderTemplateXslt().getFilename() + ") for datasource "

					+ datasourceId, e);

		} catch (IOException e) {

			throw new MSROException("Error parsing template: " + getMdBuilderTemplateXslt().getFilename(), e);

		} catch (Exception e) {

			log.error(st.toString());

			throw new MSROException(e);

		}

	}

	private String getBaseUrl() throws ISLookUpException {

		String xQuery = "/*[.//RESOURCE_IDENTIFIER/@value='{datasourceId}']//INTERFACE[@id='{interfaceId}']//BASE_URL/string()";

		xQuery = xQuery.replace("{interfaceId}", datasourceInterface).replace("{datasourceId}", datasourceId);

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

	}

	private String getMetadataIdentifierPath() throws ISLookUpException {

		String xQuery = "for $x in collection('/db/DRIVER/RepositoryServiceResources/RepositoryServiceResourceType') "

				+ "return $x//INTERFACE[@id='{interfaceId}']/INTERFACE_EXTRA_FIELD[@name='metadata_identifier_path']/string()";

		xQuery = xQuery.replace("{interfaceId}", datasourceInterface);

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

	}

	private String getMetadataNamespace() {

		try {

			String xQuery = "let $x := /*[.//RESOURCE_IDENTIFIER/@value='{datasourceId}']//INTERFACE[@id='{interfaceId}']/ACCESS_PROTOCOL/@format/string() "

					+ "return /*[.//RESOURCE_TYPE/@value='MetadataFormatDSResourceType']//METADATAFORMAT[@Prefix=$x]/@NameSpace/string()";

			xQuery = xQuery.replace("{interfaceId}", datasourceInterface).replace("{datasourceId}", datasourceId);

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

		} catch (ISLookUpException e) {

			log.error("The interface is not OAI or the format is not found in the MetadataFormatDSResourceType, thus metadata format in the <about> section "

					+ "cannot managed here and it will be leaved empty (for the time being)");

			return "";

		}

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public XSLTMappedResultSetFactory getXsltMappedResultSetFactory() {

		return xsltMappedResultSetFactory;

	}

	@Required

	public void setXsltMappedResultSetFactory(final XSLTMappedResultSetFactory xsltMappedResultSetFactory) {

		this.xsltMappedResultSetFactory = xsltMappedResultSetFactory;

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

	public Resource getMdBuilderTemplateXslt() {

		return mdBuilderTemplateXslt;

	}

	@Required

	public void setMdBuilderTemplateXslt(final Resource mdBuilderTemplateXslt) {

		this.mdBuilderTemplateXslt = mdBuilderTemplateXslt;

	}

}

package eu.dnetlib.msro.cron;

import java.util.Date;

import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.quartz.CronExpression;

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

import com.googlecode.sarasvati.GraphProcess;

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.msro.workflows.sarasvati.loader.WorkflowExecutor;

import eu.dnetlib.msro.workflows.sarasvati.registry.GraphProcessRegistry;

public class ScheduledWorkflowLauncher {

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

	private static final DateUtils dateUtils = new DateUtils();

	private WorkflowExecutor workflowExecutor;

	private GraphProcessRegistry graphProcessRegistry;

	private int windowSize = 1800000; // 30 minutes

	@Resource

	private UniqueServiceLocator serviceLocator;

	public void verifySheduledWorkflows() {

		log.debug("Verifying scheduled workflows - START");

		final String query = "for $x in collection('/db/DRIVER/MetaWorkflowDSResources/MetaWorkflowDSResourceType') " +

				"where $x//CONFIGURATION/@status='EXECUTABLE' " +

				"and $x//SCHEDULING/@enabled='true' " +

				"return concat($x//RESOURCE_IDENTIFIER/@value, ' @@@ ', $x//SCHEDULING/CRON, ' @@@ ', $x//SCHEDULING/MININTERVAL)";

		try {

			for (final String s : serviceLocator.getService(ISLookUpService.class).quickSearchProfile(query)) {

				final String[] arr = s.split("@@@");

				final String id = arr[0].trim();

				final String cron = arr[1].trim();

				final int minInterval = Math.max(NumberUtils.toInt(arr[2].trim(), 0), windowSize);

				final Date lastExecutionDate = calculateLastExecutionDate(id);

				if (isReady(id, cron, minInterval, lastExecutionDate, new Date()) && !isAlreadyRunning(id)) {

					try {

						workflowExecutor.startMetaWorkflow(id, false);

					} catch (final Exception e) {

						log.debug("Error launching scheduled wf: " + id, e);

					}

				}

			}

		} catch (final ISLookUpException e) {

			log.error("Error executing query " + query);

		}

		log.debug("Verifying scheduled workflows - END");

	}

	protected boolean isReady(final String id, final String cron, final int minInterval, final Date lastExecutionDate, final Date now) {

		if (CronExpression.isValidExpression(cron)) {

			final int minIntervalMillis = minInterval * 60000; // minutes to millis;

			final boolean res;

			if (lastExecutionDate != null) {

				final int elapsed = Math.round(now.getTime() - lastExecutionDate.getTime());

				res = (elapsed > minIntervalMillis) && verifyCron(cron, now);

			} else {

				res = verifyCron(cron, now);

			}

			if (log.isDebugEnabled()) {

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

				log.debug("META WORKFLOW ID       : " + id);

				log.debug("NOW                    : " + now);

				log.debug("LAST EXECUTION DATE    : " + lastExecutionDate);

				log.debug("MIN INTERVAL (minutes) : " + minInterval);

				log.debug("REAL MIN INTERVAL (ms) : " + minIntervalMillis);

				log.debug("WINDOW SIZE (ms)       : " + windowSize);

				log.debug("MUST BE EXECUTED       : " + res);

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

			}

			return res;

		}

		return false;

	}

	private boolean verifyCron(final String cronExpression, final Date now) {

		try {

			final CronExpression cron = new CronExpression(cronExpression);

			final Date date = new Date(now.getTime() - windowSize);

			final Date cronDate = cron.getNextValidTimeAfter(date);

			if (log.isDebugEnabled()) {

				log.debug("NEXT EXECUTION DATE: " + cronDate);

				log.debug("FIRED              : " + (cronDate.getTime() < now.getTime()));

			}

			return cronDate.getTime() < now.getTime();

		} catch (final Exception e) {

			log.error("Error calculating next cron event: " + cronExpression, e);

			return false;

		}

	}

	private boolean isAlreadyRunning(final String metaWfId) {

		final String query = "doc('/db/DRIVER/MetaWorkflowDSResources/MetaWorkflowDSResourceType/" + StringUtils.substringBefore(metaWfId, "_")

				+ "')//WORKFLOW/@id/string()";

		try {

			for (final String profileId : serviceLocator.getService(ISLookUpService.class).quickSearchProfile(query)) {

				if (profileId.length() > 0) {

					for (final GraphProcess p : graphProcessRegistry.findProcessesByResource(profileId)) {

						switch (p.getState()) {

						case Created:

							return true;

						case Executing:

							return true;

						default:

							break;

						}

					}

				}

			}

		} catch (final ISLookUpException e) {

			log.error("Error executing query " + query);

		}

		return false;

	}

	private Date calculateLastExecutionDate(final String id) {

		final String query = "for $id in doc('/db/DRIVER/MetaWorkflowDSResources/MetaWorkflowDSResourceType/" + StringUtils.substringBefore(id, "_")

				+ "')//WORKFLOW/@id/string() " +

				"for $x in doc(concat('/db/DRIVER/WorkflowDSResources/WorkflowDSResourceType/', substring-before($id, '_'))) " +

				"where $x//LAST_EXECUTION_STATUS = 'SUCCESS' " +

				"return $x//LAST_EXECUTION_DATE/text() ";

		long time = 0;

		try {

			for (final String s : serviceLocator.getService(ISLookUpService.class).quickSearchProfile(query)) {

				if (s.length() > 0) {

					final Date d = dateUtils.parse(s);

					if (time < d.getTime()) {

						time = d.getTime();

					}

				}

			}

		} catch (final ISLookUpException e) {

			log.error("Error executing query " + query);

		} catch (final Exception e) {

			log.error("Error calculating date", e);

		}

		return time > 0 ? new Date(time) : null;

	}

	public WorkflowExecutor getWorkflowExecutor() {

		return workflowExecutor;

	}

	@Required

	public void setWorkflowExecutor(final WorkflowExecutor workflowExecutor) {

		this.workflowExecutor = workflowExecutor;

	}

	public GraphProcessRegistry getGraphProcessRegistry() {

		return graphProcessRegistry;

	}

	@Required

	public void setGraphProcessRegistry(final GraphProcessRegistry graphProcessRegistry) {

		this.graphProcessRegistry = graphProcessRegistry;

	}

	public int getWindowSize() {

		return windowSize;

	}

	@Required

	public void setWindowSize(final int windowSize) {

		this.windowSize = windowSize;

	}

}

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

import java.io.IOException;

import java.util.Map;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.google.common.collect.Maps;

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.enabling.resultset.MappedResultSetFactory;

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

import eu.dnetlib.msro.rmi.MSROException;

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

import groovy.lang.GroovyShell;

import groovy.util.GroovyScriptEngine;

public class GroovyJobNode extends SimpleJobNode {

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

	/**

	 * used to transform the records using Groovy.

	 */

	private MappedResultSetFactory mappedResultSetFactory;

	private String inputEprParam;

	private String outputEprParam;

	private String transformationRuleId;

	// private String groovyParams;

	@Resource

	private UniqueServiceLocator serviceLocator;

	private Map<String, String> retrieveGroovyParameter() {

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

		String query = "for $x in collection('/db/DRIVER/GroovyProcessingDSResource/GroovyProcessingDSResourceType')"

				+ "where $x[.//RESOURCE_IDENTIFIER/@value='" + transformationRuleId + "']"

				+ "return concat($x//GROOVY_CLASSPATH/text(),':::',$x//GROOVY_DNETCLASS/text())";

		try {

			String result = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(query).get(0);

			if (result == null) { return null; }

			String[] data = result.trim().split(":::");

			if (data.length == 2) {

				out.put("classpath", data[0]);

				out.put("mainClass", data[1]);

			}

			return out;

		} catch (Exception e) {

			log.error(e);

			return null;

		}

	}

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String inputEprString = token.getEnv().getAttribute(inputEprParam);

		if (inputEprString == null || inputEprString.isEmpty()) { throw new MSROException("InputEprParam (" + inputEprParam + ") not found in ENV"); }

		final W3CEndpointReference inputEpr = new EPRUtils().getEpr(inputEprString);

		String groovyClasspath, groovyDnetClass;

		Map<String, String> prop = retrieveGroovyParameter();

		groovyClasspath = prop.get("classpath");

		groovyDnetClass = prop.get("mainClass");

		W3CEndpointReference epr = transformGroovy(inputEpr, groovyClasspath, groovyDnetClass, parseJsonParameters(token));

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

		return Arc.DEFAULT_ARC;

	}

	private W3CEndpointReference transformGroovy(final W3CEndpointReference source,

			final String groovyClasspath,

			final String groovyDnetClass,

			final Map<String, String> params) throws ClassNotFoundException, IOException {

		GroovyScriptEngine gse = new GroovyScriptEngine(groovyClasspath);

		gse.getGroovyClassLoader().loadClass(groovyDnetClass);

		log.info("***********************************************");

		log.info("Loaded Groovy classes:");

		for (Class<?> c : gse.getGroovyClassLoader().getLoadedClasses()) {

			log.info(c.getCanonicalName());

		}

		log.info("***********************************************");

		GroovyShell groovyShell = new GroovyShell(gse.getGroovyClassLoader());

		Object go = groovyShell.evaluate("new " + groovyDnetClass + "()");

		if (go instanceof GroovyUnaryFunction) {

			GroovyUnaryFunction groovyUnaryFunction = (GroovyUnaryFunction) go;

			if (params != null) {

				groovyUnaryFunction.setParams(params);

			}

			return mappedResultSetFactory.createMappedResultSet(source, groovyUnaryFunction);

		} else {

			throw new RuntimeException("Groovy object " + go + " is not supported");

		}

	}

	public MappedResultSetFactory getMappedResultSetFactory() {

		return mappedResultSetFactory;

	}

	@Required

	public void setMappedResultSetFactory(final MappedResultSetFactory mappedResultSetFactory) {

		this.mappedResultSetFactory = mappedResultSetFactory;

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public String getTransformationRuleId() {

		return transformationRuleId;

	}

	public void setTransformationRuleId(final String transformationRuleId) {

		this.transformationRuleId = transformationRuleId;

	}

}

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

import java.util.HashMap;

import java.util.Map;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

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

import org.springframework.core.io.ClassPathResource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.resultset.XSLTMappedResultSetFactory;

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

import eu.dnetlib.msro.rmi.MSROException;

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

public class ApplyXsltJobNode extends SimpleJobNode {

	private String inputEprParam;

	private String outputEprParam;

	private String xsltClasspath;

	private XSLTMappedResultSetFactory xsltMappedResultSetFactory;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String inputEpr = token.getEnv().getAttribute(inputEprParam);

		if ((inputEpr == null) || inputEpr.isEmpty()) throw new MSROException("InputEprParam (" + inputEprParam + ") not found in ENV");

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

		for (String name : token.getFullEnv().getAttributeNames()) {

			xsltParams.put(name.replaceAll(":", "_"), token.getFullEnv().getAttribute(name));

		}

		for (String name : token.getEnv().getAttributeNames()) {

			xsltParams.put(name.replaceAll(":", "_"), token.getEnv().getAttribute(name));

		}

		xsltParams.putAll(parseJsonParameters(token));

		final W3CEndpointReference epr = xsltMappedResultSetFactory.createMappedResultSet(new EPRUtils().getEpr(inputEpr),

				(new ClassPathResource(xsltClasspath)), xsltParams);

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

		return Arc.DEFAULT_ARC;

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public String getXsltClasspath() {

		return xsltClasspath;

	}

	public void setXsltClasspath(final String xsltClasspath) {

		this.xsltClasspath = xsltClasspath;

	}

	public XSLTMappedResultSetFactory getXsltMappedResultSetFactory() {

		return xsltMappedResultSetFactory;

	}

	@Required

	public void setXsltMappedResultSetFactory(final XSLTMappedResultSetFactory xsltMappedResultSetFactory) {

		this.xsltMappedResultSetFactory = xsltMappedResultSetFactory;

	}

}

package eu.dnetlib.msro.workflows.sarasvati.loader;

import java.io.File;

import java.util.Comparator;

import java.util.List;

import java.util.Map;

import java.util.concurrent.Executors;

import java.util.concurrent.PriorityBlockingQueue;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import com.googlecode.sarasvati.Graph;

import com.googlecode.sarasvati.GraphProcess;

import com.googlecode.sarasvati.mem.MemEngine;

import com.googlecode.sarasvati.mem.MemGraphProcess;

import eu.dnetlib.enabling.common.Stoppable;

import eu.dnetlib.enabling.common.StoppableDetails;

import eu.dnetlib.enabling.common.StoppableDetails.StopStatus;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.sarasvati.registry.GraphProcessRegistry;

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.Required;

public class WorkflowExecutor implements Stoppable {

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

	private MemEngine engine;

	private GraphLoader graphLoader;

	private GraphProcessRegistry graphProcessRegistry;

	private ProfileToSarasvatiConverter profileToSarasvatiConverter;

	private ScheduledExecutorService queueConsumers;

	private boolean paused = false;

	private int maxRunningWorkflows = WorkflowsConstants.MAX_WF_THREADS;

	@Resource

	private UniqueServiceLocator serviceLocator;

	private PriorityBlockingQueue<GraphProcess> pendingProcs = new PriorityBlockingQueue<GraphProcess>(20, new Comparator<GraphProcess>() {

		@Override

		public int compare(final GraphProcess p1, final GraphProcess p2) {

			int n1 = NumberUtils.toInt(p1.getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PRIORITY), WorkflowsConstants.DEFAULT_WF_PRIORITY);

			int n2 = NumberUtils.toInt(p2.getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PRIORITY), WorkflowsConstants.DEFAULT_WF_PRIORITY);

			return NumberUtils.compare(n1, n2);

		}

	});

	public boolean isPaused() {

		return paused;

	}

	public void setPaused(final boolean paused) {

		this.paused = paused;

	}

	public void init() {

		this.queueConsumers = Executors.newScheduledThreadPool(WorkflowsConstants.MAX_WF_THREADS);

		final int period = 60;

		final int step = period / WorkflowsConstants.MAX_WF_THREADS;

		for (int i = 0; i < WorkflowsConstants.MAX_WF_THREADS; i++) {

			this.queueConsumers.scheduleAtFixedRate(new Runnable() {

				@Override

				public void run() {

					if (isPaused()) {

						return;

					}

					int running = graphProcessRegistry.countRunningWfs(false);

					if (running >= getMaxRunningWorkflows()) {

						if (log.isDebugEnabled()) {

							log.debug("reached max running workflows: " + running);

						}

						return;

					}

					final GraphProcess process = pendingProcs.poll();

					if (process != null) {

						log.info("Starting workflow: " + process);

						final long now = DateUtils.now();

						process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_START_DATE, now);

						process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_START_HUMAN_DATE, DateUtils.calculate_ISO8601(now));

						engine.startProcess(process);

					} else {

						log.debug("Process queue is empty");

					}

				}

			}, i * step, period, TimeUnit.SECONDS);

		}

	}

	public String startProcess(final String profileId) throws Exception {

		return startProcess(profileId, null);

	}

	public String startProcess(final String profileId, final Map<String, Object> params) throws Exception {

		final WfProfileDescriptor desc = profileToSarasvatiConverter.getSarasvatiWorkflow(profileId);

		if (isPaused()) {

			log.warn("Wf " + profileId + " not launched, because WorkflowExecutor is preparing for shutdown");

			throw new MSROException("WorkflowExecutor is preparing for shutdown");

		}

		if (!desc.isReady()) {

			log.warn("Wf " + profileId + " not launched, because it is not ready to start");

			throw new MSROException("Workflow " + profileId + " is not ready to start");

		}

		if (pendingProcs.size() > WorkflowsConstants.MAX_PENDING_PROCS_SIZE) {

			log.warn("Wf " + profileId + " not launched, Max number of pending procs reached: " + WorkflowsConstants.MAX_PENDING_PROCS_SIZE);

			throw new MSROException("Max number of pending procs reached: " + WorkflowsConstants.MAX_PENDING_PROCS_SIZE);

		}

		final File tmpFile = File.createTempFile("wftfs", null);

		try {

			final Graph graph = graphLoader.loadGraph(desc.getWorkflowXml());

			final GraphProcess process = new MemGraphProcess(graph);

			final String procId = graphProcessRegistry.registerProcess(process);

			graphProcessRegistry.associateProcessWithResource(process, profileId);

			process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_PROCESS_ID, procId);

			process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID, profileId);

			process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_NAME, desc.getName());

			process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_NAME, graph.getName());

			process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_FAMILY, desc.getType());

			process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_PRIORITY, desc.getPriority());

			if (params != null) {

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

					process.getEnv().setAttribute(e.getKey(), e.getValue());

				}

			}

			log.info("Process " + process + " in queue, priority=" + desc.getPriority());

			pendingProcs.put(process);

			return procId;

		} catch (Exception e) {

			log.error("Error parsing workflow xml: " + desc.getWorkflowXml(), e);

			throw new IllegalArgumentException("Error parsing workflow");

		} finally {

			tmpFile.delete();

		}

	}

	public void startMetaWorkflow(final String id, final boolean manual) throws Exception {

		final String query = "/*[.//RESOURCE_IDENTIFIER/@value='" + id + "']//CONFIGURATION[@status='EXECUTABLE']/WORKFLOW/@id/string()";

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

		final List<String> list = lookup.quickSearchProfile(query);

		if (list == null || list.isEmpty()) { throw new MSROException("Metaworkflow " + id + " not launched"); }

		for (String wfId : list) {

			final String q = "/*[.//RESOURCE_IDENTIFIER/@value='" + wfId + "']//CONFIGURATION/@start/string()";

			if (manual || lookup.getResourceProfileByQuery(q).equals("auto")) {

				startProcess(wfId);

			} else {

				log.debug("Worflow " + wfId + " can not be launched AUTOMATICALLY");

			}

		}

	}

	public GraphLoader getGraphLoader() {

		return graphLoader;

	}

	@Required

	public void setGraphLoader(final GraphLoader graphLoader) {

		this.graphLoader = graphLoader;

	}

	public MemEngine getEngine() {

		return engine;

	}

	@Required

	public void setEngine(final MemEngine engine) {

		this.engine = engine;

	}

	public GraphProcessRegistry getGraphProcessRegistry() {

		return graphProcessRegistry;

	}

	@Required

	public void setGraphProcessRegistry(final GraphProcessRegistry graphProcessRegistry) {

		this.graphProcessRegistry = graphProcessRegistry;

	}

	public ProfileToSarasvatiConverter getProfileToSarasvatiConverter() {

		return profileToSarasvatiConverter;

	}

	@Required

	public void setProfileToSarasvatiConverter(final ProfileToSarasvatiConverter profileToSarasvatiConverter) {

		this.profileToSarasvatiConverter = profileToSarasvatiConverter;

	}

	@Override

	public void stop() {

		this.paused = true;

	}

	@Override

	public void resume() {

		this.paused = false;

	}

	@Override

	public StoppableDetails getStopDetails() {

		final int count = graphProcessRegistry.countRunningWfs();

		StoppableDetails.StopStatus status = StopStatus.RUNNING;

		if (isPaused()) {

			if (count == 0) {

				status = StopStatus.STOPPED;

			} else {

				status = StopStatus.STOPPING;

			}

		}

		graphProcessRegistry.listIdentifiers();

		return new StoppableDetails("D-NET workflow manager", "Running workflows: " + count, status);

	}

	public int getMaxRunningWorkflows() {

		return maxRunningWorkflows;

	}

	public void setMaxRunningWorkflows(final int maxRunningWorkflows) {

		this.maxRunningWorkflows = maxRunningWorkflows;

	}

}

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

import java.io.IOException;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.StringUtils;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.database.rmi.DatabaseService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

public class QueryDbJobNode extends AsyncJobNode {

	private String db;

	private String dbParam;

	private String dbProperty;

	private String sql;

	private String sqlForSize;

	private String xslt;

	private String outputEprParam;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String sqlText = fetchSqlAsText(sql);

		W3CEndpointReference epr = null;

		final DatabaseService dbService = serviceLocator.getService(DatabaseService.class);

		if (StringUtils.isNotBlank(xslt)) {

			final String xsltText = IOUtils.toString(getClass().getResourceAsStream(xslt));

			if (StringUtils.isBlank(sqlForSize)) {

				epr = dbService.xsltSearchSQL(findDb(token), sqlText, xsltText);

			} else {

				epr = dbService.alternativeXsltSearchSQL(findDb(token), sqlText, fetchSqlAsText(sqlForSize), xsltText);

			}

		} else {

			if (StringUtils.isBlank(sqlForSize)) {

				epr = dbService.searchSQL(findDb(token), sqlText);

			} else {

				epr = dbService.alternativeSearchSQL(findDb(token), sqlText, fetchSqlAsText(sqlForSize));

			}

		}

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

		return Arc.DEFAULT_ARC;

	}

	private String fetchSqlAsText(final String path) throws IOException {

		return IOUtils.toString(getClass().getResourceAsStream(path));

	}

	private String findDb(final NodeToken token) {

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

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

		} else if (dbProperty != null && !dbProperty.isEmpty()) {

			return getPropertyFetcher().getProperty(dbProperty);

		} else {

			return db;

		}

	}

	public String getDb() {

		return db;

	}

	public void setDb(final String db) {

		this.db = db;

	}

	public String getDbParam() {

		return dbParam;

	}

	public void setDbParam(final String dbParam) {

		this.dbParam = dbParam;

	}

	public String getSql() {

		return sql;

	}

	public void setSql(final String sql) {

		this.sql = sql;

	}

	public String getXslt() {

		return xslt;

	}

	public void setXslt(final String xslt) {

		this.xslt = xslt;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public String getDbProperty() {

		return dbProperty;

	}

	public void setDbProperty(final String dbProperty) {

		this.dbProperty = dbProperty;

	}

	public String getSqlForSize() {

		return sqlForSize;

	}

	public void setSqlForSize(final String sqlForSize) {

		this.sqlForSize = sqlForSize;

	}

}

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

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

public class PrepareCreateIndexJobNode extends SimpleJobNode {

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

	private String layout;

	private String format;

	private String interpretation;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		log.info("Preparing env for CreateIndexJobNode");

		token.getEnv().setAttribute("layout", layout);

		token.getEnv().setAttribute("format", format);

		token.getEnv().setAttribute("interpretation", interpretation);

		return null;

	}

	public String getLayout() {

		return layout;

	}

	public void setLayout(final String layout) {

		this.layout = layout;

	}

	public String getFormat() {

		return format;

	}

	public void setFormat(final String format) {

		this.format = format;

	}

	public String getInterpretation() {

		return interpretation;

	}

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

}

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

import javax.annotation.Resource;

import com.google.common.base.Splitter;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.datasources.rmi.DatasourceConstants;

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.msro.rmi.MSROException;

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

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

public class VerifyDatasourceJobNode extends SimpleJobNode {

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String expectedInterfaceTypologyPrefixes;

	private String expectedCompliancePrefixes;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String dsId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ID);

		final String ifaceId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);

		final ISLookUpService lookupService = serviceLocator.getService(ISLookUpService.class);