D-Net

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

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-msro-service/tags/dnet-msro-service-2.1.2</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.0.0,2.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>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-datasource-common</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

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

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.mdstore.MDStoreService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.rmi.MSROException;

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

public class FetchMDStoreRecordsJobNode extends SimpleJobNode {

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String mdId;

	private String mdFormat;

	private String eprParam;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		if (getMdId() == null) {

			setMdId(token.getEnv().getAttribute("mdId"));

		}

		if (getMdFormat() == null) {

			setMdFormat(token.getEnv().getAttribute("mdFormat"));

		}

		log.info("getting MDRecords from: " + getMdId());

		final W3CEndpointReference epr = serviceLocator.getService(MDStoreService.class, getMdId()).deliverMDRecords(getMdId(), "", "", "");

		if (epr == null) { throw new MSROException("unable to read MDRecords from: " + getMdId()); }

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

		return Arc.DEFAULT_ARC;

	}

	public String getMdId() {

		return mdId;

	}

	public void setMdId(final String mdId) {

		this.mdId = mdId;

	}

	public String getMdFormat() {

		return mdFormat;

	}

	public void setMdFormat(final String mdFormat) {

		this.mdFormat = mdFormat;

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

}

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

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

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

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

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

import eu.dnetlib.msro.workflows.nodes.blackboard.BlackboardWorkflowJobListener;

public class CreateObjectStoreJobNode extends BlackboardJobNode {

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

	private String interpretation;

	private String outputPrefix = "objectStore_";

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		log.info("preparing blackboard job for the creation of the objectStore ");

		job.setAction("CREATE");

		job.getParameters().put("interpretation", interpretation);

	}

	public String getInterpretation() {

		return interpretation;

	}

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

	public String getOutputPrefix() {

		return outputPrefix;

	}

	public void setOutputPrefix(final String outputPrefix) {

		this.outputPrefix = outputPrefix;

	}

	@Override

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Engine engine, final NodeToken token) {

		return new BlackboardWorkflowJobListener(engine, token) {

			@Override

			protected void populateEnv(final Env env, final Map<String, String> responseParams) {

				;

				env.setAttribute(getOutputPrefix() + "interpretation", interpretation);

				env.setAttribute(getOutputPrefix() + "id", responseParams.get("objectStoreId"));

			}

		};

	}

}

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

<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"

	elementFormDefault="qualified">

	<xs:element name="RESOURCE_PROFILE">

		<xs:complexType>

			<xs:sequence>

				<xs:element ref="HEADER" />

				<xs:element ref="BODY" />

			</xs:sequence>

		</xs:complexType>

	</xs:element>

	<xs:element name="HEADER">

		<xs:complexType>

			<xs:sequence>

				<xs:element ref="RESOURCE_IDENTIFIER" />

				<xs:element ref="RESOURCE_TYPE" />

				<xs:element ref="RESOURCE_KIND" />

				<xs:element ref="RESOURCE_URI" />

				<xs:element ref="DATE_OF_CREATION" />

			</xs:sequence>

		</xs:complexType>

	</xs:element>

	<xs:element name="RESOURCE_IDENTIFIER">

		<xs:complexType>

			<xs:attribute name="value" use="required" />

		</xs:complexType>

	</xs:element>

	<xs:element name="RESOURCE_TYPE">

		<xs:complexType>

			<xs:attribute name="value" use="required" fixed="WorkflowDSResourceType" />

		</xs:complexType>

	</xs:element>

	<xs:element name="RESOURCE_KIND">

		<xs:complexType>

			<xs:attribute name="value" use="required" fixed="WorkflowDSResources" />

		</xs:complexType>

	</xs:element>

	<xs:element name="RESOURCE_URI">

		<xs:complexType>

			<xs:attribute name="value" use="required" />

		</xs:complexType>

	</xs:element>

	<xs:element name="DATE_OF_CREATION">

		<xs:complexType>

			<xs:attribute name="value" use="required" type="xs:dateTime" />

		</xs:complexType>

	</xs:element>

	<xs:element name="BODY">

		<xs:complexType>

			<xs:sequence>

				<xs:element name="WORKFLOW_NAME" type="xs:string" />

				<xs:element name="WORKFLOW_INFO" minOccurs="0" maxOccurs="1">

					<xs:complexType>

						<xs:sequence>

							<xs:element name="FIELD" minOccurs="0" maxOccurs="unbounded">

								<xs:complexType>

									<xs:simpleContent>

										<xs:extension base="xs:string">

											<xs:attribute name="name" use="required" type="xs:string"/>

										</xs:extension>

									</xs:simpleContent>

								</xs:complexType>

							</xs:element>

						</xs:sequence>	

					</xs:complexType>

				</xs:element>

				<xs:element name="WORKFLOW_TYPE" type="xs:string" />

				<xs:element name="WORKFLOW_PRIORITY">

					<xs:simpleType>

						<xs:restriction base="xs:integer">

							<xs:minInclusive value="0" />

							<xs:maxInclusive value="100" />

						</xs:restriction>

					</xs:simpleType>

				</xs:element>

				<xs:element ref="CONFIGURATION" />

				<xs:element ref="STATUS" />

			</xs:sequence>

		</xs:complexType>

	</xs:element>

	<xs:element name="WORKFLOW_NAME" type="xs:string" />

	<xs:element name="CONFIGURATION">

		<xs:complexType>

			<xs:sequence>

				<xs:element maxOccurs="unbounded" ref="NODE" />

			</xs:sequence>

			<xs:attribute name="start" use="required">

				<xs:simpleType>

					<xs:restriction base="xs:NCName">

						<xs:enumeration value="auto" />

						<xs:enumeration value="manual" />

						<xs:enumeration value="disabled" />

					</xs:restriction>

				</xs:simpleType>

			</xs:attribute>

		</xs:complexType>

	</xs:element>

	<xs:element name="NODE">

		<xs:complexType>

			<xs:sequence>

				<xs:element ref="DESCRIPTION" />

				<xs:element ref="PARAMETERS" />

				<xs:element ref="ARCS" />

			</xs:sequence>

			<xs:attribute name="isStart" use="optional" type="xs:boolean" />

			<xs:attribute name="isJoin" use="optional" type="xs:boolean" />

			<xs:attribute name="name" use="required" type="xs:NCName" />

			<xs:attribute name="type" use="optional" type="xs:NCName" />

		</xs:complexType>

	</xs:element>

	<xs:element name="DESCRIPTION" type="xs:string" />

	<xs:element name="PARAMETERS">

		<xs:complexType>

			<xs:sequence>

				<xs:element minOccurs="0" maxOccurs="unbounded" ref="PARAM" />

			</xs:sequence>

		</xs:complexType>

	</xs:element>

	<xs:element name="PARAM">

		<xs:complexType mixed="true">

			<xs:simpleContent>

				<xs:extension base="xs:string">

					<xs:attribute name="name" use="required" type="xs:NCName" />

					<xs:attribute name="required" use="optional" type="xs:boolean" />

					<xs:attribute name="managedBy" use="required">

						<xs:simpleType>

							<xs:restriction base="xs:NCName">

								<xs:enumeration value="user" />

								<xs:enumeration value="system" />

							</xs:restriction>

						</xs:simpleType>

					</xs:attribute>

					<xs:attribute name="type" use="optional">

						<xs:simpleType>

							<xs:restriction base="xs:NCName">

								<xs:enumeration value="int" />

								<xs:enumeration value="boolean" />

								<xs:enumeration value="float" />

								<xs:enumeration value="string" />

								<xs:enumeration value="date" />

								<xs:enumeration value="datetime" />

								<xs:enumeration value="property" />

							</xs:restriction>

						</xs:simpleType>

					</xs:attribute>

					<xs:attribute name="function" use="optional" type="xs:string" />

					<xs:attribute name="category" use="optional" type="xs:string" />

				</xs:extension>

			</xs:simpleContent>

		</xs:complexType>

	</xs:element>

	<xs:element name="ARCS">

		<xs:complexType>

			<xs:sequence>

				<xs:element maxOccurs="unbounded" ref="ARC" />

			</xs:sequence>

		</xs:complexType>

	</xs:element>

	<xs:element name="ARC">

		<xs:complexType>

			<xs:attribute name="name" use="optional" type="xs:string" />

			<xs:attribute name="to" use="required" type="xs:NCName" />

		</xs:complexType>

	</xs:element>

	<xs:element name="STATUS">

		<xs:complexType>

			<xs:sequence minOccurs="0">

				<xs:element name="LAST_EXECUTION_ID" type="xs:string" />

				<xs:element name="LAST_EXECUTION_DATE" type="xs:dateTime" />

				<xs:element name="LAST_EXECUTION_STATUS" type="xs:NCName" />

				<xs:element name="LAST_EXECUTION_ERROR" type="xs:string" />

				<xs:element name="LAST_EXECUTION_OUTPUT" minOccurs="0" maxOccurs="unbounded">

					<xs:complexType>

						<xs:simpleContent>

							<xs:extension base="xs:string">

								<xs:attribute name="name" type="xs:string" use="required" />

							</xs:extension>

						</xs:simpleContent>

					</xs:complexType>

				</xs:element>

			</xs:sequence>

		</xs:complexType>

	</xs:element>

</xs:schema>

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

import java.io.StringReader;

import java.io.StringWriter;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

public class DeleteMetaWfJobNode extends SimpleJobNode {

	private String metaWfId;

	@Resource

	private UniqueServiceLocator serviceLocator;

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

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

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

		final String dsId = doc.valueOf("//DATAPROVIDER/@id");

		final String dsName = doc.valueOf("//DATAPROVIDER/text()");

		final String ifaceId = doc.valueOf("//DATAPROVIDER/@interface");

		final String destroyWfId = doc.valueOf("//CONFIGURATION/@destroyWorkflow");

		log.info("Removing a MetaWf of dataprovider: " + dsId);

		token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_ID, dsId);

		token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_NAME, dsName);

		token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE, ifaceId);

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

		for (Object o : doc.selectNodes("//WORKFLOW")) {

			final String wfId = ((Node) o).valueOf("@id");

			try {

				registry.deleteProfile(wfId);

				log.info(" - Deleted Workflow: " + wfId);

			} catch (Exception e) {

				log.error(" - (ERR) Error deleting profile " + wfId);

			}

		}

		registry.deleteProfile(metaWfId);

		log.info(" - Deleted MetaWorkflow: " + metaWfId);

		registry.deleteProfile(destroyWfId);

		log.info(" - Deleted destroy workflow: " + destroyWfId);

		verifyDatasource(dsId, ifaceId);

		return Arc.DEFAULT_ARC;

	}

	private void verifyDatasource(final String dsId, final String ifaceId) throws Exception {

		final StringWriter sw = new StringWriter();

		sw.append("for $x in collection('/db/DRIVER/MetaWorkflowDSResources/MetaWorkflowDSResourceType') where");

		sw.append("  $x//DATAPROVIDER/@id = '" + dsId + "' and ");

		sw.append("  $x//DATAPROVIDER/@interface = '" + ifaceId + "' and ");

		sw.append("  $x//RESOURCE_IDENTIFIER/@value != '" + metaWfId + "' ");

		sw.append("return $x//RESOURCE_IDENTIFIER/@value/string()");

		final boolean active = !serviceLocator.getService(ISLookUpService.class).quickSearchProfile(sw.toString()).isEmpty();

		log.info(" - Updating iface, active status: " + active);

		updateIfaceActivationStatus(dsId, ifaceId, active);

	}

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

		serviceLocator.getService(ISRegistryService.class).updateProfileNode(dsId, "//INTERFACE[@id = '" + ifaceId + "']/@active", "'" + active + "'");

	}

	public String getMetaWfId() {

		return metaWfId;

	}

	public void setMetaWfId(final String metaWfId) {

		this.metaWfId = metaWfId;

	}

}

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

import java.util.Map;

import java.util.Map.Entry;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

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

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

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

public class BlackboardWorkflowJobListener extends AbstractBlackboardJobListener {

	/**

	 * logger.

	 */

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

	/**

	 * workflow engine.

	 */

	private Engine engine;

	/**

	 * workflow node token.

	 */

	private NodeToken token;

	@Override

	protected void onDone(final BlackboardJob job) {

		log.debug("Blackboard workflow node DONE");

		complete(job, Arc.DEFAULT_ARC);

	}

	public BlackboardWorkflowJobListener(final Engine engine, final NodeToken token) {

		super();

		this.engine = engine;

		this.token = token;

	}

	@Override

	final public void processJob(final BlackboardJob job) {

		token.getEnv().setTransientAttribute(WorkflowsConstants.BLACKBOARD_JOB, job);

		super.processJob(job);

	}

	@Override

	final protected void onFailed(final BlackboardJob job) {

		log.warn("Blackboard workflow node FAILED: " + job.getError());

		token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_HAS_FAILED, true);

		token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_ERROR, job.getError());

		complete(job, "failed");

	}

	final protected void complete(final BlackboardJob job, final String arc) {

		final Env env = token.getEnv();

		populateEnv(env, job.getParameters());

		engine.complete(token, arc);

		engine.executeQueuedArcTokens(token.getProcess());

	}

	protected void populateEnv(final Env env, Map<String, String> responseParams) {

		for (Entry<String, String> entry : responseParams.entrySet()) {

			env.setAttribute(WorkflowsConstants.BLACKBOARD_PARAM_PREFIX + entry.getKey(), entry.getValue());

		}

	}

	@Override

	protected void onOngoing(final BlackboardJob job) {

		token.getEnv().setAttribute(WorkflowsConstants.BLACKBOARD_IS_GOING, true);

	}

	public Engine getEngine() {

		return engine;

	}

	public void setEngine(final Engine engine) {

		this.engine = engine;

	}

	public NodeToken getToken() {

		return token;

	}

	public void setToken(final NodeToken token) {

		this.token = token;

	}

}

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

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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.msro.workflows.nodes.SimpleJobNode;

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

public class ValidateDatasourceJobNode extends SimpleJobNode {

	@Resource

	private UniqueServiceLocator serviceLocator;

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

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

		String newId = registerDatasourceWorkflow(oldId);

		token.getFullEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_ID, newId);

		token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_ID, newId);

		log.info("Validated datasource - OLD ID: " + oldId);

		log.info("Validated datasource - NEW ID: " + newId);

		return Arc.DEFAULT_ARC;

	}

	public String registerDatasourceWorkflow(final String oldId) throws ISRegistryException {

		return serviceLocator.getService(ISRegistryService.class).validateProfile(oldId);

	}

}

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

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

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

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

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

import eu.dnetlib.msro.workflows.nodes.blackboard.BlackboardWorkflowJobListener;

// TODO: Auto-generated Javadoc

/**

 * The Class CreateIndexJobNode.

 */

public class CreateIndexJobNode extends BlackboardJobNode {

	/** The Constant log. */

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

	/** The output prefix. */

	private String outputPrefix = "index_";

	/** The default backend id. */

	private String defaultBackendId;

	/** The format. */

	private String format;

	/** The layout. */

	private String layout;

	/** The interpretation. */

	private String interpretation;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.workflows.nodes.BlackboardJobNode#prepareJob(eu.dnetlib.enabling.tools.blackboard.BlackboardJob,

	 * com.googlecode.sarasvati.NodeToken)

	 */

	@Override

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

		final String env_format = token.getFullEnv().getAttribute("format");

		final String env_layout = token.getFullEnv().getAttribute("layout");

		final String env_interp = token.getFullEnv().getAttribute("interpretation");

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

			this.format = env_format;

		}

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

			this.layout = env_layout;

		}

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

			this.interpretation = env_interp;

		}

		log.info("preparing CREATE blackboard job: " + format + "-" + layout + "-" + interpretation);

		job.setAction("CREATE");

		job.getParameters().put("format", format);

		job.getParameters().put("layout", layout);

		job.getParameters().put("interpretation", interpretation);

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

	}

	/**

	 * Gets the output prefix.

	 * 

	 * @return the output prefix

	 */

	public String getOutputPrefix() {

		return outputPrefix;

	}

	/**

	 * Sets the output prefix.

	 * 

	 * @param outputPrefix

	 *            the output prefix

	 */

	public void setOutputPrefix(final String outputPrefix) {

		this.outputPrefix = outputPrefix;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.workflows.nodes.BlackboardJobNode#generateBlackboardListener(com.googlecode.sarasvati.Engine,

	 * com.googlecode.sarasvati.NodeToken)

	 */

	@Override

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Engine engine, final NodeToken token) {

		return new BlackboardWorkflowJobListener(engine, token) {

			@Override

			protected void populateEnv(final Env env, final Map<String, String> responseParams) {

				env.setAttribute(getOutputPrefix() + "format", getFormat());

				env.setAttribute(getOutputPrefix() + "layout", getLayout());

				env.setAttribute(getOutputPrefix() + "interpretation", getInterpretation());

				env.setAttribute(getOutputPrefix() + "id", responseParams.get("id"));

			}

		};

	}

	/**

	 * Gets the default backend id.

	 * 

	 * @return the default backend id

	 */

	public String getDefaultBackendId() {

		return defaultBackendId;

	}

	/**

	 * Sets the default backend id.

	 * 

	 * @param defaultBackendId

	 *            the default backend id

	 */

	@Required

	public void setDefaultBackendId(final String defaultBackendId) {

		this.defaultBackendId = defaultBackendId;

	}

	/**

	 * Gets the format.

	 * 

	 * @return the format

	 */

	public String getFormat() {

		return format;

	}

	/**

	 * Sets the format.

	 * 

	 * @param format

	 *            the format

	 */

	public void setFormat(final String format) {

		this.format = format;

	}

	/**

	 * Gets the layout.

	 * 

	 * @return the layout

	 */

	public String getLayout() {

		return layout;

	}

	/**

	 * Sets the layout.

	 * 

	 * @param layout

	 *            the layout

	 */

	public void setLayout(final String layout) {

		this.layout = layout;

	}

	/**

	 * Gets the interpretation.

	 * 

	 * @return the interpretation

	 */

	public String getInterpretation() {

		return interpretation;

	}

	/**

	 * Sets the interpretation.

	 * 

	 * @param interpretation

	 *            the interpretation

	 */

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

}

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

import java.io.IOException;

import java.io.StringWriter;

import javax.annotation.Resource;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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.msro.workflows.nodes.SimpleJobNode;

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

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

public class RegisterMetaWfJobNode extends SimpleJobNode {

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String wfName;

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

		final String dsName = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_NAME);

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

		final String metaWfId = registerDatasourceWorkflow(dsId, dsName, ifaceId);

		token.getFullEnv().setAttribute("META_WORKFLOW_ID", metaWfId);

		token.getFullEnv().setAttribute("META_WORKFLOW_STATUS", WorkflowStatus.ASSIGNED.toString());

		return Arc.DEFAULT_ARC;

	}

	public String registerDatasourceWorkflow(final String dsId, final String dsName, final String ifaceId) throws ISRegistryException, IOException {

		final StringWriter sw = new StringWriter();

		IOUtils.copy(getClass().getResourceAsStream("/eu/dnetlib/msro/workflows/templates/meta-workflow.xml.st"), sw);

		final StringTemplate st = new StringTemplate(sw.toString());

		st.setAttribute("dsId", dsId);

		st.setAttribute("ifaceId", ifaceId);

		st.setAttribute("dsName", dsName);

		st.setAttribute("section", "dataproviders");

		st.setAttribute("wfName", getWfName());

		st.setAttribute("wfFamily", getWfName());

		st.setAttribute("status", WorkflowStatus.ASSIGNED.toString());

		return serviceLocator.getService(ISRegistryService.class).registerProfile(st.toString());

	}

	public String getWfName() {

		return wfName;

	}

	public void setWfName(final String wfName) {

		this.wfName = wfName;

	}

}

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

import java.util.Iterator;

import java.util.List;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

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

import eu.dnetlib.data.mdstore.MDStoreService;

import eu.dnetlib.data.mdstore.MDStoreServiceException;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

// TODO: Auto-generated Javadoc

/**

 * The Class MultipleMdStoreIterator.

 */

public class MultipleMdStoreIterator implements Iterable<String>, Iterator<String> {

	/** The service locator. */

	@Resource

	private UniqueServiceLocator serviceLocator;

	/** The md i ds. */

	private List<String> mdIDs;

	/** The current id. */

	private String currentId = null;

	/** The current iterator. */

	private Iterator<String> currentIterator;

	/** The result set client factory. */

	@Autowired

	private ResultSetClientFactory resultSetClientFactory;

	/**

	 * Instantiates a new multiple md store iterator.

	 * 

	 * @param mdstoreLocator

	 *            the mdstore locator

	 * @param mdIds

	 *            the md ids

	 */

	public MultipleMdStoreIterator(final UniqueServiceLocator serviceLocator, final List<String> mdIds,

			final ResultSetClientFactory resultSetClientFactory) {

		this.serviceLocator = serviceLocator;

		this.mdIDs = mdIds;

		this.resultSetClientFactory = resultSetClientFactory;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see java.lang.Iterable#iterator()

	 */

	@Override

	public Iterator<String> iterator() {

		getNextMDStoreRecords();

		return this;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see java.util.Iterator#hasNext()

	 */

	@Override

	public boolean hasNext() {

		if (currentId == null || currentIterator == null) { return false; }

		if (!currentIterator.hasNext()) {

			getNextMDStoreRecords();

		}

		return currentIterator.hasNext();

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see java.util.Iterator#next()

	 */

	@Override

	public String next() {

		return currentIterator.next();

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see java.util.Iterator#remove()

	 */

	@Override

	public void remove() {

		currentIterator.remove();

	}

	/**

	 * Gets the next md store records.

	 * 

	 * @return the next md store records

	 */

	private void getNextMDStoreRecords() {

		if (mdIDs.size() > 0) {

			currentId = mdIDs.remove(0);

			currentIterator = getIterableResultset(currentId);

		}

	}

	/**

	 * Gets the iterable resultset.

	 * 

	 * @param id

	 *            the id

	 * @return the iterable resultset

	 */

	private Iterator<String> getIterableResultset(final String id) {

		try {

			W3CEndpointReference epr = serviceLocator.getService(MDStoreService.class, id).deliverMDRecords(id, "", "", "");

			Iterable<String> input = resultSetClientFactory.getClient(epr);

			return input.iterator();

		} catch (MDStoreServiceException e) {

			return null;

		}

	}

}

package eu.dnetlib.msro.workflows.nodes;

import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

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

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

import eu.dnetlib.msro.workflows.nodes.blackboard.BlackboardWorkflowJobListener;

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

public abstract class BlackboardJobNode extends SarasvatiJobNode {

	@Resource

	private UniqueServiceLocator serviceLocator;

	/**

	 * logger.

	 */

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

	/**

	 * blackboard handler.

	 */

	@Resource

	private BlackboardClientHandler blackboardClientHandler;

	/**

	 * blackboard job registry.

	 */

	@Resource

	private BlackboardJobRegistry jobRegistry;

	@Override

	public void execute(final Engine engine, final NodeToken token) {

		super.execute(engine, token);

		log.info("executing blackboard node");

		try {

			token.getEnv().setAttribute(WorkflowsConstants.BLACKBOARD_IS_BLACKBOARD, true);

			final String serviceId = obtainServiceId(token);

			if (StringUtils.isBlank(serviceId)) {

				token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_HAS_FAILED, true);

				final String msg = "cannot locate target service profile: " + serviceId;

				token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_ERROR, msg);

				log.error(msg);

				engine.complete(token, "failed");

				return;

			}

			final BlackboardJob job = blackboardClientHandler.newJob(serviceId);

			token.getEnv().setTransientAttribute(WorkflowsConstants.BLACKBOARD_JOB, job);

			token.getEnv().setAttribute(WorkflowsConstants.BLACKBOARD_SERVICE_ID, ((BlackboardJobImpl) job).getServiceId());

			prepareJob(job, token);

			jobRegistry.registerJobListener(job, generateBlackboardListener(engine, token));

			blackboardClientHandler.assign(job);

		} catch (final Throwable e) {

			token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_HAS_FAILED, true);

			token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_ERROR, "cannot prepare blackboard job: " + e);

			engine.complete(token, "failed");

			log.error("cannot prepare blackboard job", e);

		}

	}

	abstract protected String obtainServiceId(NodeToken token);

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

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Engine engine, final NodeToken token) {

		return new BlackboardWorkflowJobListener(engine, token);

	}

	public UniqueServiceLocator getServiceLocator() {

		return serviceLocator;

	}

}

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

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Iterator;

import java.util.Map;

import javax.annotation.Resource;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.common.logging.DnetLogger;

import eu.dnetlib.msro.rmi.MSROException;

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

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

public class FindDateRangeForIncrementalHarvestingJobNode extends SimpleJobNode {

	private String fromDateParam;

	private String untilDateParam;

	private long ONE_DAY = 1000 * 60 * 60 * 24;	

	@Resource(name = "msroWorkflowLogger")

	private DnetLogger dnetLogger;

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

	@Override

	protected String execute(NodeToken token) throws Exception {

		final String profId = findCurrentWfProfileId(token);

		final String fromDate = calculateFromDate(profId); 

		final String untilDate = null;

		log.info("Incremental Harv Details - profId: " + profId + " - from: " + fromDate + " - until: " + untilDate);

		if (fromDate != null) {

			token.getEnv().setAttribute(getFromDateParam(), fromDate);

		}

		//if (untilDate != null) {

		//	token.getEnv().setAttribute(getUntilDateParam(), untilDate);

		//}

		return Arc.DEFAULT_ARC;

	}	

	private String calculateFromDate(final String profId) {

		final long d = findLastSuccessStartDate(profId);

		return (d > 0) ? (new SimpleDateFormat("yyyy-MM-dd")).format(new Date(d - ONE_DAY)) : null;

	}

	private long findLastSuccessStartDate(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();

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

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

				if (curr > res) {

					res = curr;

				}

			}

		}

		return res;

	}

	private String findCurrentWfProfileId(NodeToken token) throws MSROException {

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

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

			return p1;

		}

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

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

			return p2;

		}

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

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

			return p3;

		}

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

	}

	public String getFromDateParam() {

		return fromDateParam;

	}

	public void setFromDateParam(String fromDateParam) {

		this.fromDateParam = fromDateParam;

	}

	public String getUntilDateParam() {

		return untilDateParam;

	}

	public void setUntilDateParam(String untilDateParam) {

		this.untilDateParam = untilDateParam;

	}

}

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

<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema" elementFormDefault="qualified">

	<xs:element name="RESOURCE_PROFILE">

		<xs:complexType>

			<xs:sequence>

				<xs:element name="HEADER" type="HEADERType" />

				<xs:element name="BODY" type="BODYType" />

			</xs:sequence>

		</xs:complexType>

	</xs:element>

	<xs:complexType name="HEADERType">

		<xs:all>

			<xs:element name="RESOURCE_IDENTIFIER" type="RESOURCE_IDENTIFIERType" />

			<xs:element name="RESOURCE_TYPE" type="RESOURCE_TYPEType" />

			<xs:element name="RESOURCE_KIND" type="RESOURCE_KINDType" />

			<xs:element name="RESOURCE_URI" type="RESOURCE_URIType" />

			<xs:element name="DATE_OF_CREATION" type="DATEType" />

		</xs:all>

	</xs:complexType>

	<xs:complexType name="BODYType">

		<xs:sequence>

			<xs:element name="DATAPROVIDER" minOccurs="0" maxOccurs="1">

				<xs:complexType>

					<xs:simpleContent>

						<xs:extension base="xs:string">

							<xs:attribute name="id" type="xs:string" use="required" />

							<xs:attribute name="interface" type="xs:string" use="required" />

						</xs:extension>

					</xs:simpleContent>

				</xs:complexType>			

			</xs:element>

			<xs:element name="METAWORKFLOW_NAME">

				<xs:complexType>

					<xs:simpleContent>

						<xs:extension base="xs:string">

	        				<xs:attribute name="family" type="xs:string" use="required"/>

	        			</xs:extension>

	        		</xs:simpleContent>

        		</xs:complexType>

			</xs:element>

			<xs:element name="METAWORKFLOW_DESCRIPTION" type="xs:string" />

			<xs:element name="METAWORKFLOW_SECTION" type="xs:string"/>

			<xs:element name="ADMIN_EMAIL" type="xs:string"/>

			<xs:element name="CONFIGURATION" type="CONFIGURATIONType" />

			<xs:element name="SCHEDULING" type="SCHEDULINGType" />

		</xs:sequence>

	</xs:complexType>

	<xs:complexType name="RESOURCE_IDENTIFIERType">

		<xs:attribute name="value" type="xs:string" use="required" />

	</xs:complexType>

	<xs:complexType name="RESOURCE_TYPEType">

		<xs:attribute name="value" use="required">

			<xs:simpleType>

				<xs:restriction base="xs:string">

					<xs:enumeration value="MetaWorkflowDSResourceType" />

				</xs:restriction>

			</xs:simpleType>

		</xs:attribute>

	</xs:complexType>

	<xs:complexType name="RESOURCE_KINDType">

		<xs:attribute name="value" use="required">

			<xs:simpleType>

				<xs:restriction base="xs:string">

					<xs:enumeration value="MetaWorkflowDSResources" />

				</xs:restriction>

			</xs:simpleType>

		</xs:attribute>

	</xs:complexType>

	<xs:complexType name="RESOURCE_URIType">

		<xs:attribute name="value" type="xs:string" use="required" />

	</xs:complexType>

	<xs:complexType name="DATEType">

		<xs:attribute name="value" type="xs:dateTime" use="required" />

	</xs:complexType>

	<xs:complexType name="CONFIGURATIONType">

		<xs:sequence minOccurs="0">

			<xs:any processContents="skip"/>

		</xs:sequence>

		<xs:attribute name="status" use="required">

			<xs:simpleType>

				<xs:restriction base="xs:string">

					<xs:enumeration value="EXECUTABLE" />

					<xs:enumeration value="WAIT_USER_SETTINGS" />

					<xs:enumeration value="WAIT_SYS_SETTINGS" />

					<xs:enumeration value="ASSIGNED" />

					<xs:enumeration value="MISSING" />

				</xs:restriction>

			</xs:simpleType>

		</xs:attribute>

		<xs:attribute name="destroyWorkflow" use="optional" type="xs:string" />

	</xs:complexType>

	<xs:complexType name="SCHEDULINGType">

		<xs:sequence>

			<xs:element name="CRON" type="xs:string" />

			<xs:element name="MININTERVAL" type="xs:int" />

		</xs:sequence>

		<xs:attribute name="enabled" use="required" type="xs:boolean" />

	</xs:complexType>

</xs:schema>

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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

public class SetProviderInfoJobNode extends SimpleJobNode {

	private String providerId;

	private String providerName;

	private String api;

	@Override

	protected String execute(NodeToken token) throws Exception {

		token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_ID, getProviderId());

		token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_NAME, getProviderName());

		token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE, getApi());

		token.getFullEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_ID, getProviderId());

		token.getFullEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_NAME, getProviderName());

		token.getFullEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE, getApi());

		token.getProcess().getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_ID, getProviderId());

		token.getProcess().getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_NAME, getProviderName());

		token.getProcess().getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE, getApi());

		return Arc.DEFAULT_ARC;

	}

	public String getProviderId() {

		return providerId;

	}

	public void setProviderId(String providerId) {

		this.providerId = providerId;

	}

	public String getProviderName() {

		return providerName;

	}

	public void setProviderName(String providerName) {

		this.providerName = providerName;

	}

	public String getApi() {

		return api;

	}

	public void setApi(String api) {

		this.api = api;

	}

}

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