D-Net

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

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 query = getXqueryForServiceId(token);

			final String serviceId = serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(query);

			if (StringUtils.isBlank(serviceId)) {

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

				final String msg = "cannot locate target service profile, using query: " + query;

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

	}

}

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

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

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

public class DeleteObjectStoreJobNode extends BlackboardJobNode {

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

	private String objectstoreId;

	@Override

	protected String getXqueryForServiceId(final NodeToken token) {

		return "//RESOURCE_IDENTIFIER[../RESOURCE_TYPE/@value='ObjectStoreServiceResourceType']/@value/string()";

	}

	@Override

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

		log.info("preparing blackboard job: delete of mdstore " + getObjectstoreId());

		job.setAction("DELETE");

		job.getParameters().put("obsID", getObjectstoreId());

	}

	/**

	 * @return the objectstoreId

	 */

	public String getObjectstoreId() {

		return objectstoreId;

	}

	/**

	 * @param objectstoreId

	 *            the objectstoreId to set

	 */

	public void setObjectstoreId(final String objectstoreId) {

		this.objectstoreId = objectstoreId;

	}

}

package eu.dnetlib.msro.workflows.nodes;

import java.util.Map;

import javax.annotation.Resource;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.StringEscapeUtils;

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 com.google.common.collect.Maps;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.common.logging.DnetLogger;

import eu.dnetlib.common.logging.LogMessage;

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.util.WorkflowsConstants;

/**

 * The success node sets the "isCompletedSuccessfully" env var.

 * 

 */

public class SuccessFailureNode extends SimpleJobNode {

	/**

	 * is completed successfully.

	 */

	private boolean success;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Resource(name = "msroWorkflowLogger")

	private DnetLogger dnetLogger;

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

	@Override

	protected String execute(final NodeToken token) {

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

		final long now = DateUtils.now();

		final String date = DateUtils.calculate_ISO8601(now);

		token.getProcess().getEnv().setAttribute(WorkflowsConstants.SYSTEM_END_DATE, now);

		token.getProcess().getEnv().setAttribute(WorkflowsConstants.SYSTEM_END_HUMAN_DATE, date);

		final Map<String, String> params = mergeEnvAttributes(token);

		final LogMessage logMessage = dnetLogger.newLogMessage().addDetails(params);

		try {

			final String template = IOUtils.toString(getClass().getResourceAsStream("/eu/dnetlib/msro/workflows/templates/workflow_status.xml.st"));

			final StringTemplate st = new StringTemplate(template);

			st.setAttribute("procId", StringEscapeUtils.escapeXml(params.get(WorkflowsConstants.SYSTEM_WF_PROCESS_ID)));

			st.setAttribute("date", StringEscapeUtils.escapeXml(date));

			st.setAttribute("params", filterOutputParams(params));

			if (!isSuccess()) {

				st.setAttribute("error", StringEscapeUtils.escapeXml(params.get(WorkflowsConstants.SYSTEM_ERROR)));

			}

			serviceLocator.getService(ISRegistryService.class).updateProfileNode(profileId, "//STATUS", st.toString());

			token.getProcess().getEnv().setAttribute(WorkflowsConstants.SYSTEM_COMPLETED_SUCCESSFULLY, isSuccess());

			logMessage.addDetail(WorkflowsConstants.SYSTEM_COMPLETED_SUCCESSFULLY, Boolean.toString(isSuccess()));

		} catch (Exception e) {

			log.error("Error updating workflow profile: " + profileId, e);

			token.getProcess().getEnv().setAttribute(WorkflowsConstants.SYSTEM_COMPLETED_SUCCESSFULLY, false);

			logMessage.addDetail(WorkflowsConstants.SYSTEM_COMPLETED_SUCCESSFULLY, Boolean.toString(false));

		}

		logMessage.flush();

		return Arc.DEFAULT_ARC;

	}

	private Map<String, String> filterOutputParams(final Map<String, String> map) {

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

		if (map != null) {

			for (String k : map.keySet()) {

				if (!StringUtils.isBlank(k) && (k.startsWith(WorkflowsConstants.DATAPROVIDER_PREFIX) || k.startsWith(WorkflowsConstants.MAIN_LOG_PREFIX))) {

					final String key = StringEscapeUtils.escapeXml(k);

					final String v = map.get(k);

					res.put(key, v != null ? StringEscapeUtils.escapeXml(v) : "null");

				}

			}

		}

		return res;

	}

	private Map<String, String> mergeEnvAttributes(final NodeToken token) {

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

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

			map.put(s, token.getEnv().getAttribute(s));

		}

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

			map.put(s, token.getFullEnv().getAttribute(s));

		}

		return map;

	}

	public boolean isSuccess() {

		return success;

	}

	@Required

	public void setSuccess(final boolean success) {

		this.success = success;

	}

}

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

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

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

public class DeleteIndexJobNode extends BlackboardJobNode {

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

	private String indexId;

	private String defaultIndexId;

	@Override

	protected String getXqueryForServiceId(final NodeToken token) {

		return "collection('/db/DRIVER/ServiceResources/IndexServiceResourceType')//RESOURCE_IDENTIFIER/@value/string()";

	}

	@Override

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

		log.info("deleting index id = " + getIndexId());

		job.setAction("DELETE");

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

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

	}

	public String getIndexId() {

		return indexId;

	}

	public void setIndexId(final String indexId) {

		this.indexId = indexId;

	}

	public String getDefaultIndexId() {

		return defaultIndexId;

	}

	public void setDefaultIndexId(final String defaultIndexId) {

		this.defaultIndexId = defaultIndexId;

	}

}

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

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.utility.cleaner.rmi.CleanerService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

public class CleanJobNode extends SimpleJobNode {

	private String inputEprParam;

	private String outputEprParam;

	private String ruleId;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

		final W3CEndpointReference outputEpr = ruleId == null || ruleId.isEmpty() ? inputEpr : serviceLocator.getService(CleanerService.class).clean(inputEpr,

				ruleId);

		token.getEnv().setAttribute(outputEprParam, outputEpr.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 getRuleId() {

		return ruleId;

	}

	public void setRuleId(final String ruleId) {

		this.ruleId = ruleId;

	}

}

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.nodes.transform;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.transformation.service.rmi.TransformationService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

public class TransformJobNode extends SimpleJobNode {

	private String inputEprParam;

	private String outputEprParam;

	private String ruleId;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

		final W3CEndpointReference outputEpr = serviceLocator.getService(TransformationService.class).transform(ruleId, inputEpr);

		token.getEnv().setAttribute(outputEprParam, outputEpr.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 getRuleId() {

		return ruleId;

	}

	public void setRuleId(final String ruleId) {

		this.ruleId = ruleId;

	}

}

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

import java.util.Iterator;

import java.util.List;

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. */

	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).deliverMDRecords(id, "", "", "");

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

			return input.iterator();

		} catch (MDStoreServiceException e) {

			return null;

		}

	}

}

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

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

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

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

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

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

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

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

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

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

public class StoreMDStoreRecordsJobNode extends BlackboardJobNode implements ProgressJobNode {

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

	private String eprParam;

	private String mdId;

	private String storingType;

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	private ResultsetProgressProvider progressProvider;

	@Override

	protected String getXqueryForServiceId(final NodeToken token) {

		return "//RESOURCE_IDENTIFIER[../RESOURCE_TYPE/@value='MDStoreServiceResourceType']/@value/string()";

	}

	@Override

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

		job.setAction("FEED");

		final String eprS = token.getEnv().getAttribute(getEprParam());

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

		job.getParameters().put("epr", progressProvider.getEpr().toString());

		job.getParameters().put("storingType", getStoringType());

		job.getParameters().put("mdId", getMdId());

	}

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

				log.info("Number of stored records: " + responseParams.get("total"));

				env.setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "total", responseParams.get("total"));

			}

		};

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

	public String getMdId() {

		return mdId;

	}

	public void setMdId(final String mdId) {

		this.mdId = mdId;

	}

	public String getStoringType() {

		return storingType;

	}

	public void setStoringType(final String storingType) {

		this.storingType = storingType;

	}

	public ProcessCountingResultSetFactory getProcessCountingResultSetFactory() {

		return processCountingResultSetFactory;

	}

	@Required

	public void setProcessCountingResultSetFactory(final ProcessCountingResultSetFactory processCountingResultSetFactory) {

		this.processCountingResultSetFactory = processCountingResultSetFactory;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return progressProvider;

	}

}

<?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.0.1</version>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-msro-service/tags/dnet-msro-service-2.0.1</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-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.blackboard;

import javax.annotation.Resource;

import javax.xml.ws.Endpoint;

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 eu.dnetlib.enabling.is.sn.rmi.ISSNException;

import eu.dnetlib.enabling.is.sn.rmi.ISSNService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.soap.EndpointReferenceBuilder;

public class BlackboardSubscriber {

	/**

	 * Logger.

	 */

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

	/**

	 * service locator.

	 */

	@Resource

	private UniqueServiceLocator serviceLocator;

	/**

	 * injected EPR builder.

	 */

	@Resource(name = "jaxwsEndpointReferenceBuilder")

	private EndpointReferenceBuilder<Endpoint> eprBuilder;

	/**

	 * notification endpoint (normally the msro service).

	 */

	private Endpoint endpoint;

	/**

	 * performs the subscription.

	 * 

	 * @throws ISSNException

	 *             could happen

	 */

	public void subscribeAll() throws ISSNException {

		log.info("Subscribing msro service");

		final W3CEndpointReference endpointReference = eprBuilder.getEndpointReference(getEndpoint());

		serviceLocator.getService(ISSNService.class).subscribe(endpointReference, "UPDATE/*/*/RESOURCE_PROFILE/BODY/BLACKBOARD/LAST_RESPONSE", 0);

	}

	public Endpoint getEndpoint() {

		return endpoint;

	}

	@Required

	public void setEndpoint(final Endpoint endpoint) {

		this.endpoint = endpoint;

	}

}

package eu.dnetlib.msro.workflows.nodes;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

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.msro.workflows.util.WorkflowsConstants;

public abstract class AsyncJobNode extends SarasvatiJobNode {

	/**

	 * logger.

	 */

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

	private final ExecutorService executor = Executors.newCachedThreadPool();

	@Override

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

		super.execute(engine, token);

		log.info("executing async node");

		executor.execute(new Runnable() {

			@Override

			public void run() {

				try {

					log.debug("START NODE: " + getBeanName());

					beforeStart(token);

					String arc = execute(token);

					beforeCompleted(token);

					log.debug("END NODE (SUCCESS): " + getBeanName());

					engine.complete(token, arc);

				} catch (Throwable e) {

					log.error("got exception while executing workflow node", e);

					log.debug("END NODE (FAILED): " + getBeanName());

					beforeFailed(token);

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

					token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_ERROR, e.getMessage());

					engine.complete(token, "failed");

				}

			}

		});

	}

	abstract protected String execute(final NodeToken token) throws Exception;

	protected void beforeStart(final NodeToken token) {}

	protected void beforeCompleted(final NodeToken token) {}

	protected void beforeFailed(final NodeToken token) {}

}

package eu.dnetlib.msro.workflows.nodes.harvest.fs;

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

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

import eu.dnetlib.msro.rmi.MSROException;

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

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

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

public class StartHarvestFSJobNode extends BlackboardJobNode {

	/**

	 * Logger.

	 */

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

	/**

	 * service locator.

	 */

	private UniqueServiceLocator serviceLocator;

	private static final String protocol = "filesystem";

	private String datasourceId;

	private String username;

	private String password;

	private String filter;

	private String eprParam;

	private String datasourceInterface;

	private String format;

	@Override

	protected String getXqueryForServiceId(final NodeToken token) {

		return "collection('/db/DRIVER/ServiceResources/HarvestingServiceResourceType')/RESOURCE_PROFILE"

				+ "[.//SERVICE_PROPERTIES/PROPERTY[@key='protocol']/@value = '" + protocol + "']//RESOURCE_IDENTIFIER/@value/string()";

	}

	@Override

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

		job.setAction("LISTRECORDS");

		if (datasourceId == null || datasourceId.isEmpty()) {

			throw new MSROException("Missing field datasourceId");

		} else {

			job.getParameters().put("repository_id", getDatasourceId());

		}

		if (datasourceInterface == null || datasourceInterface.isEmpty()) {

			throw new MSROException("Missing field datasourceInterface");

		} else {

			job.getParameters().put("base_url", obtainBaseUrl());

		}

		if (format == null || format.isEmpty()) {

			throw new MSROException("Missing field format");

		} else {

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

		}

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

			job.getParameters().put("username", getUsername());

		}

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

			job.getParameters().put("password", getPassword());

		}

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

			job.getParameters().put("filter", getFilter());

		}

	}

	private String obtainBaseUrl() throws ISLookUpDocumentNotFoundException, ISLookUpException {

		String query = "/RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '" + getDatasourceId() + "']" +

				"//INTERFACE[@label='" + getDatasourceInterface() +

				"' and ./ACCESS_PROTOCOL = '" + protocol +

				"' and .//FORMAT = '" + getFormat() + "']/BASE_URL/text()";

		log.info("Executing query: " + query);

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

	}

	@Override

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

		return new OngoingBlackboardWorkflowJobListener(engine, token) {

			@Override

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

				log.info("Obtained epr: " + responseParams.get("rs_epr"));

				env.setAttribute(getEprParam(), responseParams.get("rs_epr"));

			}

		};

	}

	public String getUsername() {

		return username;

	}

	public void setUsername(final String username) {

		this.username = username;

	}

	public String getPassword() {

		return password;

	}

	public void setPassword(final String password) {

		this.password = password;

	}

	public String getFilter() {

		return filter;

	}

	public void setFilter(final String filter) {

		this.filter = filter;

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

	public String getFormat() {

		return format;

	}

	public void setFormat(final String format) {

		this.format = format;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

}

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

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

public class OngoingBlackboardWorkflowJobListener extends BlackboardWorkflowJobListener {

	public OngoingBlackboardWorkflowJobListener(Engine engine, NodeToken token) {

		super(engine, token);

	}

	@Override

	protected void onOngoing(BlackboardJob job) {

		super.onOngoing(job);

		complete(job, Arc.DEFAULT_ARC);

	}

}

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

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