D-Net

package eu.dnetlib.msro.worker.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.worker.WorkflowConstants;

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(WorkflowConstants.SYSTEM_HAS_FAILED, true);

					token.getEnv().setAttribute(WorkflowConstants.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.worker.nodes.unpack;

import java.io.StringReader;

import java.util.Iterator;

import java.util.Queue;

import java.util.concurrent.PriorityBlockingQueue;

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.data.resultSet.ResultSetFactory;

import eu.dnetlib.data.resultSet.client.ResultSetClientFactory;

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

import eu.dnetlib.rmi.objects.resultSet.ResultSet;

public class UnpackJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	@Resource

	private ResultSetClientFactory resultSetClientFactory;

	@Resource

	private ResultSetFactory resultSetFactory;

	private String inputResultSetParam;

	private String outputResultSetParam;

	private String xpath;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		@SuppressWarnings("unchecked")

		final ResultSet<String> rsInput = (ResultSet<String>) token.getEnv().getTransientAttribute(inputResultSetParam);

		final Iterator<String> client = resultSetClientFactory.getClient(rsInput, String.class, 100).iterator();

		final Queue<String> queue = new PriorityBlockingQueue<String>();

		if (client.hasNext()) {

			populateQueue(queue, client.next(), xpath);

		}

		final ResultSet<String> rsOutput = resultSetFactory.createResultSet(new Iterable<String>() {

			@Override

			public Iterator<String> iterator() {

				return new Iterator<String>() {

					@Override

					public boolean hasNext() {

						synchronized (queue) {

							return !queue.isEmpty();

						}

					}

					@Override

					public String next() {

						synchronized (queue) {

							final String res = queue.poll();

							while (queue.isEmpty() &&

									client.hasNext()) {

								populateQueue(queue, client.next(), xpath);

							}

							return res;

						}

					}

					@Override

					public void remove() {}

				};

			}

		});

		token.getEnv().setTransientAttribute(outputResultSetParam, rsOutput);

		return Arc.DEFAULT_ARC;

	}

	private void populateQueue(final Queue<String> queue, final String record, final String xpath) {

		try {

			final SAXReader reader = new SAXReader();

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

			for (Object o : doc.selectNodes(xpath)) {

				queue.add(((Node) o).asXML());

			}

		} catch (Exception e) {

			log.error("Error unpacking record: \n" + record, e);

			throw new RuntimeException(e);

		}

	}

	public String getXpath() {

		return xpath;

	}

	public void setXpath(final String xpath) {

		this.xpath = xpath;

	}

	public String getInputResultSetParam() {

		return inputResultSetParam;

	}

	public void setInputResultSetParam(final String inputResultSetParam) {

		this.inputResultSetParam = inputResultSetParam;

	}

	public String getOutputResultSetParam() {

		return outputResultSetParam;

	}

	public void setOutputResultSetParam(final String outputResultSetParam) {

		this.outputResultSetParam = outputResultSetParam;

	}

}

package eu.dnetlib.msro.worker.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.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

import eu.dnetlib.common.services.locators.DnetServiceLocator;

import eu.dnetlib.enabling.blackboard.BlackboardCallback;

import eu.dnetlib.enabling.blackboard.BlackboardDispatcher;

import eu.dnetlib.msro.worker.WorkflowConstants;

import eu.dnetlib.rmi.blackboard.AbstractBlackboardMessage;

public abstract class BlackboardJobNode<T> extends SarasvatiJobNode implements BlackboardCallback<T> {

	/**

	 * logger.

	 */

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

	@Resource

	private DnetServiceLocator serviceLocator;

	@Resource

	private BlackboardDispatcher blackboardDispatcher;

	private Engine engine;

	private NodeToken token;

	@Override

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

		super.execute(engine, token);

		this.engine = engine;

		this.token = token;

		log.info("executing blackboard node");

		try {

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

			final String serviceId = obtainServiceId(token);

			if (StringUtils.isBlank(serviceId)) {

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

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

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

				log.error(msg);

				engine.complete(token, "failed");

				return;

			}

			blackboardDispatcher.createDispatcher(serviceId, getMessage(), this);

		} catch (final Throwable e) {

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

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

			engine.complete(token, "failed");

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

		}

	}

	abstract protected T getMessage();

	abstract protected String obtainServiceId(NodeToken token);

	@Override

	public void onSuccess(final T t) {

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

		saveResponseInEnv(t, token.getEnv());

		engine.complete(token, Arc.DEFAULT_ARC);

		engine.executeQueuedArcTokens(token.getProcess());

	}

	abstract protected void saveResponseInEnv(final T t, final Env env);

	@Override

	public void onFail(final T t) {

		final String error = t instanceof AbstractBlackboardMessage ? ((AbstractBlackboardMessage) t).getError() : "Unknown Error";

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

		saveResponseInEnv(t, token.getEnv());

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

		token.getEnv().setAttribute(WorkflowConstants.SYSTEM_ERROR, error);

		engine.complete(token, "failed");

	}

	@Override

	public void onOngoing(final T t) {

		getToken().getEnv().setAttribute(WorkflowConstants.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.worker.nodes.sel;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

/**

 * The Class SelectiveNode allows to decide which path a workflow must take.

 */

public class SelectiveNode extends SimpleJobNode {

	/** The selection. */

	private String selection = Arc.DEFAULT_ARC;

	/*

	 * (non-Javadoc)

	 * 

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

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		return selection;

	}

	/**

	 * Gets the selection.

	 *

	 * @return the selection

	 */

	public String getSelection() {

		return selection;

	}

	/**

	 * Sets the selection.

	 *

	 * @param selection

	 *            the new selection

	 */

	public void setSelection(final String selection) {

		this.selection = selection;

	}

}

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

import java.util.List;

import java.util.Set;

import javax.annotation.Resource;

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

import com.google.common.collect.Lists;

import com.google.common.collect.Sets;

import com.google.gson.Gson;

import com.google.gson.GsonBuilder;

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

/**

 * The Class RetrieveMdStoreId is a job node used to retrieve the correct MDStore from which extract the url of the file to download.

 * metadata format and interpretation are injected as properties

 */

public class RetrieveMdStoreId extends SimpleJobNode {

	/** The metadata format. */

	private String metadataFormat;

	/** The interpretation. */

	private String interpretation;

	/** The provider id. */

	private String providerId;

	/** The service locator. */

	@Resource

	private DnetServiceLocator serviceLocator;

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		String workflowQuery =

				"for $x in collection('/db/DRIVER/MetaWorkflowDSResources/MetaWorkflowDSResourceType') where($x//DATAPROVIDER/@id='%s') return  distinct-values($x//WORKFLOW/@id/string())";

		List<String> result = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(String.format(workflowQuery, providerId));

		if (result.size() == 0) { throw new RuntimeException("there is no mdStore Associated to the provider " + token.getEnv().getAttribute(getProviderId())); }

		Set<String> workflowIds = Sets.newHashSet(result);

		Set<String> metadataIds = getMdStores(workflowIds);

		Gson g = new GsonBuilder().disableHtmlEscaping().create();

		token.getEnv().setAttribute("mdId", g.toJson(metadataIds));

		token.getEnv().setAttribute("mdFormat", getMetadataFormat());

		return Arc.DEFAULT_ARC;

	}

	private Set<String> getMdStores(final Set<String> workflowsId) {

		try {

			String query = "//RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value='%s']//PARAM[./@category/string()='MDSTORE_ID']/text()";

			Set<String> mdStores = Sets.newHashSet();

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

			for (String workflowId : workflowsId) {

				List<String> result = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(String.format(query, workflowId));

				Set<String> metadataIds = Sets.newHashSet(result);

				mdStores.addAll(getRightMetadataId(Lists.newArrayList(metadataIds)));

			}

			return mdStores;

		} catch (ISLookUpException e) {

			return null;

		}

	}

	/**

	 * Gets the right metadata id whith the format metadataFormat and interpretation interpretation

	 *

	 * @return the right metadata id

	 * @throws ISLookUpException

	 */

	private Set<String> getRightMetadataId(final Iterable<String> ids) throws ISLookUpException {

		String query =

				"let $x:=//RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value='%s'] return concat($x//METADATA_FORMAT/text(), '::<<>>::', $x//METADATA_FORMAT_INTERPRETATION/text())";

		Set<String> result = Sets.newHashSet();

		for (String id : ids) {

			List<String> results = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(String.format(query, id));

			if (results.size() > 0) {

				String[] values = results.get(0).split("::<<>>::");

				if (metadataFormat.equals(values[0]) && interpretation.equals(values[1])) {

					result.add(id);

				}

			}

		}

		return result;

	}

	/**

	 * Gets the interpretation.

	 *

	 * @return the interpretation

	 */

	public String getInterpretation() {

		return interpretation;

	}

	/**

	 * Sets the interpretation.

	 *

	 * @param interpretation

	 *            the interpretation to set

	 */

	@Required

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

	/**

	 * Gets the metadata format.

	 *

	 * @return the metadataFormat

	 */

	public String getMetadataFormat() {

		return metadataFormat;

	}

	/**

	 * Sets the metadata format.

	 *

	 * @param metadataFormat

	 *            the metadataFormat to set

	 */

	@Required

	public void setMetadataFormat(final String metadataFormat) {

		this.metadataFormat = metadataFormat;

	}

	/**

	 * Gets the provider id.

	 *

	 * @return the providerId

	 */

	public String getProviderId() {

		return providerId;

	}

	/**

	 * Sets the provider id.

	 *

	 * @param providerId

	 *            the providerId to set

	 */

	public void setProviderId(final String providerId) {

		this.providerId = providerId;

	}

}

package eu.dnetlib.msro.worker.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.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.worker.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 DnetServiceLocator serviceLocator;

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.worker.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;

	}

}

package eu.dnetlib.msro.worker.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.msro.worker.nodes.SimpleJobNode;

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

public class DeleteMetaWfJobNode extends SimpleJobNode {

	private String metaWfId;

	@Resource

	private DnetServiceLocator 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(WorkflowConstants.DATAPROVIDER_ID, dsId);

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

		token.getEnv().setAttribute(WorkflowConstants.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.worker.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.services.locators.DnetServiceLocator;

import eu.dnetlib.miscutils.DateUtils;

import eu.dnetlib.msro.worker.WorkflowConstants;

/**

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

 * 

 */

public class SuccessFailureNode extends SimpleJobNode {

	/**

	 * is completed successfully.

	 */

	private boolean success;

	@Resource

	private DnetServiceLocator serviceLocator;

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

	@Override

	protected String execute(final NodeToken token) {

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

		final long now = DateUtils.now();

		final String date = DateUtils.getDate_ISO8601(now);

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

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

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

		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(WorkflowConstants.SYSTEM_WF_PROCESS_ID)));

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

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

			if (!isSuccess()) {

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

			}

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

		} catch (Exception e) {

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

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

		}

		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(WorkflowConstants.DATAPROVIDER_PREFIX) || k.startsWith(WorkflowConstants.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.worker.nodes.index;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.msro.worker.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.worker.nodes.download;

import java.util.Map;

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

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

import eu.dnetlib.data.download.rmi.DownloadService;

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

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

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

import eu.dnetlib.msro.worker.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.WorkflowConstants;

// TODO: Auto-generated Javadoc

/**

 * The Class DownloadFromMetadata is a job node that send a blackboard message to the Download service to start to download file from url

 * retrieved by Metadata .

 */

public class DownloadFromMetadataJobNode extends BlackboardJobNode implements ProgressJobNode {

	/** The inputepr param. */

	private String inputeprParam;

	/** The obejct store id. */

	private String objectStoreID;

	/** The plugin. */

	private String plugin;

	/** The protocol. */

	private String protocol;

	/** The mime type. */

	private String mimeType;

	/** The process counting result set factory. */

	@Autowired

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	/** The progress provider. */

	private ResultsetProgressProvider progressProvider;

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.worker.nodes.BlackboardJobNode#obtainServiceId(com.googlecode.sarasvati.NodeToken)

	 */

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	/*

	 * (non-Javadoc)

	 * 

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

	 * com.googlecode.sarasvati.NodeToken)

	 */

	@Override

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

		job.setAction("DOWNLOAD");

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

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

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

		job.getParameters().put("protocol", getProtocol());

		job.getParameters().put("plugin", getPlugin());

		job.getParameters().put("mimeType", getMimeType());

		job.getParameters().put("objectStoreID", getObjectStoreID());

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.worker.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(WorkflowConstants.MAIN_LOG_PREFIX + "total", responseParams.get("total"));

			}

		};

	}

	/**

	 * Gets the inputepr param.

	 *

	 * @return the inputeprParam

	 */

	public String getInputeprParam() {

		return inputeprParam;

	}

	/**

	 * Gets the object store id.

	 *

	 * @return the objectStoreID

	 */

	public String getObjectStoreID() {

		return objectStoreID;

	}

	/**

	 * Sets the object store id.

	 *

	 * @param objectStoreID

	 *            the objectStoreID to set

	 */

	public void setObjectStoreID(final String objectStoreID) {

		this.objectStoreID = objectStoreID;

	}

	/**

	 * Gets the plugin.

	 *

	 * @return the plugin

	 */

	public String getPlugin() {

		return plugin;

	}

	/**

	 * Sets the plugin.

	 *

	 * @param plugin

	 *            the plugin to set

	 */

	public void setPlugin(final String plugin) {

		this.plugin = plugin;

	}

	/**

	 * Gets the protocol.

	 *

	 * @return the protol

	 */

	public String getProtocol() {

		return protocol;

	}

	/**

	 * Sets the protocol.

	 *

	 * @param protol

	 *            the protol to set

	 */

	public void setProtocol(final String protol) {

		this.protocol = protol;

	}

	/**

	 * Gets the mime type.

	 *

	 * @return the mimeType

	 */

	public String getMimeType() {

		return mimeType;

	}

	/**

	 * Sets the mime type.

	 *

	 * @param mimeType

	 *            the mimeType to set

	 */

	public void setMimeType(final String mimeType) {

		this.mimeType = mimeType;

	}

	/**

	 * Sets the inputepr param.

	 *

	 * @param inputeprParam

	 *            the inputeprParam to set

	 */

	public void setInputeprParam(final String inputeprParam) {

		this.inputeprParam = inputeprParam;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.worker.nodes.ProgressJobNode#getProgressProvider()

	 */

	@Override

	public ProgressProvider getProgressProvider() {

		return progressProvider;

	}

}

package eu.dnetlib.msro.worker.nodes;

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.worker.WorkflowConstants;

public abstract class SimpleJobNode extends SarasvatiJobNode {

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

	@Override

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

		super.execute(engine, token);

		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(WorkflowConstants.SYSTEM_HAS_FAILED, true);

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

			engine.complete(token, "failed");

		}

	}

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

	protected void beforeStart(final NodeToken token) {

		// For optional overwrites

	}

	protected void beforeCompleted(final NodeToken token) {

		// For optional overwrites

	}

	protected void beforeFailed(final NodeToken token) {

		// For optional overwrites

	}

}

package eu.dnetlib.msro.worker.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.resultset.MappedResultSetFactory;

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

import eu.dnetlib.msro.rmi.ManagerServiceException;

import eu.dnetlib.msro.worker.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 DnetServiceLocator 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 ManagerServiceException("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.worker.nodes.transform;

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

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.resultSet.ResultSetFactory;

import eu.dnetlib.data.resultSet.transform.TransformerFactory;

import eu.dnetlib.enabling.datastructures.TransformationRule;

import eu.dnetlib.enabling.is.client.InformationServiceClient;

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

import eu.dnetlib.rmi.objects.resultSet.ResultSet;

import eu.dnetlib.rmi.soap.exceptions.ManagerServiceException;

public class TransformJobNode extends SimpleJobNode {

	private String inputResultSetParam;

	private String outputResultSetParam;

	private String ruleId;

	@Autowired

	private InformationServiceClient isClient;

	@Autowired

	private ResultSetFactory resultSetFactory;

	private TransformerFactory transformerFactory;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		@SuppressWarnings("unchecked")

		final ResultSet<String> inputRs = (ResultSet<String>) token.getEnv().getTransientAttribute(inputResultSetParam);

		if (inputRs == null) { throw new ManagerServiceException("InputEprParam (" + inputResultSetParam + ") not found in ENV"); }

		final TransformationRule rule = isClient.getResourceByCode(ruleId, TransformationRule.class);

		final ResultSet<String> outputRs = resultSetFactory.transform(inputRs, transformerFactory.createTransformer(rule));

		token.getEnv().setAttribute(outputResultSetParam, outputRs.toString());

		return Arc.DEFAULT_ARC;

	}

	public String getRuleId() {

		return ruleId;

	}

	public void setRuleId(final String ruleId) {

		this.ruleId = ruleId;

	}

	public String getInputResultSetParam() {

		return inputResultSetParam;

	}

	public void setInputResultSetParam(final String inputResultSetParam) {

		this.inputResultSetParam = inputResultSetParam;

	}

	public String getOutputResultSetParam() {

		return outputResultSetParam;

	}

	public void setOutputResultSetParam(final String outputResultSetParam) {

		this.outputResultSetParam = outputResultSetParam;

	}

	public TransformerFactory getTransformerFactory() {

		return transformerFactory;

	}

	@Required

	public void setTransformerFactory(final TransformerFactory transformerFactory) {

		this.transformerFactory = transformerFactory;

	}

}

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

import java.util.Iterator;

import java.util.List;

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

import eu.dnetlib.data.mdstore.MDStoreService;

import eu.dnetlib.data.mdstore.MDStoreServiceException;

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

// TODO: Auto-generated Javadoc

/**

 * The Class MultipleMdStoreIterator.

 */

public class MultipleMdStoreIterator implements Iterator<String> {

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

	/** The service locator. */

	@Resource

	private DnetServiceLocator 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 DnetServiceLocator serviceLocator, final List<String> mdIds, final ResultSetClientFactory resultSetClientFactory) {

		this.serviceLocator = serviceLocator;

		this.mdIDs = mdIds;

		this.resultSetClientFactory = resultSetClientFactory;

		getNextMDStoreRecords();

	}

	/*

	 * (non-Javadoc)

	 * 

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

	 */

	@Override

	public boolean hasNext() {

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

		return currentIterator.hasNext();

	}

	/*

	 * (non-Javadoc)

	 * 

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

	 */

	@Override

	public String next() {

		String nextElement = currentIterator.next();

		if (!currentIterator.hasNext()) {

			getNextMDStoreRecords();

		}

		return nextElement;

	}

	/*

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