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

import java.io.ByteArrayInputStream;

import java.util.ArrayList;

import java.util.List;

import javax.xml.parsers.DocumentBuilder;

import javax.xml.parsers.DocumentBuilderFactory;

import javax.xml.xpath.XPath;

import javax.xml.xpath.XPathConstants;

import javax.xml.xpath.XPathExpression;

import javax.xml.xpath.XPathFactory;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.joda.time.DateTime;

import org.joda.time.format.DateTimeFormat;

import org.joda.time.format.DateTimeFormatter;

import org.w3c.dom.Document;

import org.w3c.dom.NodeList;

import com.google.common.base.Function;

import com.google.gson.Gson;

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

// TODO: Auto-generated Javadoc

/**

 * The Class UrlExtractor.

 */

public class UrlExtractor implements Function<String, String> {

	/** The Constant log. */

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

	/** The xpath url. */

	private String xpathURL;

	/** The xpath. */

	private String xpathMetadataID;

	/** The xpath open access. */

	private String xpathOpenAccess;

	/** The xpath embargo date. */

	private String xpathEmbargoDate;

	/**

	 * Instantiates a new url extractor.

	 *

	 * @param xpath

	 *            the xpath

	 * @param xpathMetadataID

	 *            the xpath metadata id

	 */

	public UrlExtractor(final String xpath, final String xpathMetadataID, final String xpathOpenAccess, final String xpathEmbargoDate) {

		this.xpathURL = xpath;

		this.xpathMetadataID = xpathMetadataID;

		this.xpathOpenAccess = xpathOpenAccess;

		this.xpathEmbargoDate = xpathEmbargoDate;

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see com.google.common.base.Function#apply(java.lang.Object)

	 */

	@Override

	public String apply(final String input) {

		try {

			DownloadItem di = new DownloadItem();

			DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();

			DocumentBuilder builder;

			builder = factory.newDocumentBuilder();

			if (input == null) {

				log.error("Metadata input is null");

				return null;

			}

			Document doc = builder.parse(new ByteArrayInputStream(input.getBytes()));

			XPathFactory xPathFactory = XPathFactory.newInstance();

			XPath myXpath = xPathFactory.newXPath();

			XPathExpression expression = myXpath.compile(xpathURL);

			Object values = expression.evaluate(doc, XPathConstants.NODESET);

			di.setUrl(getNodes((NodeList) values));

			di.setOriginalUrl(getNodes((NodeList) values));

			if (xpathOpenAccess != null) {

				expression = myXpath.compile(xpathOpenAccess);

				String openAccess = expression.evaluate(doc);

				di.setOpenAccess(openAccess);

			}

			if (xpathEmbargoDate != null) {

				expression = myXpath.compile(xpathEmbargoDate);

				String embargoDate = expression.evaluate(doc);

				if (!StringUtils.isEmpty(embargoDate)) {

					try {

						DateTimeFormatter fmt = DateTimeFormat.forPattern("yyyy-MM-dd");

						DateTime dt = fmt.parseDateTime(embargoDate);

						di.setEmbargoDate(dt.toDate());

					} catch (Exception pe) {}

				}

			}

			expression = myXpath.compile(xpathMetadataID);

			String extracted_metadataId = expression.evaluate(doc);

			di.setIdItemMetadata(extracted_metadataId);

			return di.toJSON();

		} catch (Exception e) {

			log.error("OPSSS... Something bad happen on evaluating ", e);

			return null;

		}

	}

	/**

	 * Gets the nodes.

	 *

	 * @param nodes

	 *            the nodes

	 * @return the nodes

	 */

	private String getNodes(final NodeList nodes) {

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

		if (nodes != null) {

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

				extracted_Url.add(nodes.item(i).getNodeValue());

			}

		}

		return new Gson().toJson(extracted_Url);

	}

	/**

	 * Gets the xpath metadata id.

	 *

	 * @return the xpathMetadataID

	 */

	public String getXpathMetadataID() {

		return xpathMetadataID;

	}

	/**

	 * Sets the xpath metadata id.

	 *

	 * @param xpathMetadataID

	 *            the xpathMetadataID to set

	 */

	public void setXpathMetadataID(final String xpathMetadataID) {

		this.xpathMetadataID = xpathMetadataID;

	}

	/**

	 * Gets the xpath url.

	 *

	 * @return the xpath url

	 */

	public String getXpathURL() {

		return xpathURL;

	}

	/**

	 * Sets the xpath url.

	 *

	 * @param xpathURL

	 *            the new xpath url

	 */

	public void setXpathURL(final String xpathURL) {

		this.xpathURL = xpathURL;

	}

	/**

	 * Gets the xpath open access.

	 *

	 * @return the xpath open access

	 */

	public String getXpathOpenAccess() {

		return xpathOpenAccess;

	}

	/**

	 * Sets the xpath open access.

	 *

	 * @param xpathOpenAccess

	 *            the new xpath open access

	 */

	public void setXpathOpenAccess(final String xpathOpenAccess) {

		this.xpathOpenAccess = xpathOpenAccess;

	}

	/**

	 * Gets the xpath embargo date.

	 *

	 * @return the xpath embargo date

	 */

	public String getXpathEmbargoDate() {

		return xpathEmbargoDate;

	}

	/**

	 * Sets the xpath embargo date.

	 *

	 * @param xpathEmbargoDate

	 *            the new xpath embargo date

	 */

	public void setXpathEmbargoDate(final String xpathEmbargoDate) {

		this.xpathEmbargoDate = xpathEmbargoDate;

	}

}

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 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.locators.UniqueServiceLocator;

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

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

			return input.iterator();

		} catch (MDStoreServiceException e) {

			log.error(e);

			return null;

		}

	}

}

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;

	}

}

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.NodeToken;

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

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.resultset.ProcessCountingResultSetFactory;

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

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

public class UpdateDbJobNode extends BlackboardJobNode implements ProgressJobNode {

	private String db;

	private String dbParam;

	private String eprParam;

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	private ResultsetProgressProvider progressProvider;

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

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

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

			db = token.getEnv().getAttribute(getDbParam());

		}

		log.info("preparing blackboard job to update DB: " + db);

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

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

		job.setAction("IMPORT");

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

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

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return progressProvider;

	}

	public String getDb() {

		return db;

	}

	public void setDb(final String db) {

		this.db = db;

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

	public ProcessCountingResultSetFactory getProcessCountingResultSetFactory() {

		return processCountingResultSetFactory;

	}

	@Required

	public void setProcessCountingResultSetFactory(final ProcessCountingResultSetFactory processCountingResultSetFactory) {

		this.processCountingResultSetFactory = processCountingResultSetFactory;

	}

	public String getDbParam() {

		return dbParam;

	}

	public void setDbParam(final String dbParam) {

		this.dbParam = dbParam;

	}

}

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

import java.io.IOException;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.StringUtils;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

public class QueryDbJobNode extends AsyncJobNode {

	private String db;

	private String dbParam;

	private String dbProperty;

	private String sql;

	private String sqlForSize;

	private String xslt;

	private String outputEprParam;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String sqlText = fetchSqlAsText(sql);

		W3CEndpointReference epr = null;

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

		if (StringUtils.isNotBlank(xslt)) {

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

			if (StringUtils.isBlank(sqlForSize)) {

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

			} else {

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

			}

		} else {

			if (StringUtils.isBlank(sqlForSize)) {

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

			} else {

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

			}

		}

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

		return Arc.DEFAULT_ARC;

	}

	private String fetchSqlAsText(final String path) throws IOException {

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

	}

	private String findDb(final NodeToken token) {

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

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

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

			return getPropertyFetcher().getProperty(dbProperty);

		} else {

			return db;

		}

	}

	public String getDb() {

		return db;

	}

	public void setDb(final String db) {

		this.db = db;

	}

	public String getDbParam() {

		return dbParam;

	}

	public void setDbParam(final String dbParam) {

		this.dbParam = dbParam;

	}

	public String getSql() {

		return sql;

	}

	public void setSql(final String sql) {

		this.sql = sql;

	}

	public String getXslt() {

		return xslt;

	}

	public void setXslt(final String xslt) {

		this.xslt = xslt;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public String getDbProperty() {

		return dbProperty;

	}

	public void setDbProperty(final String dbProperty) {

		this.dbProperty = dbProperty;

	}

	public String getSqlForSize() {

		return sqlForSize;

	}

	public void setSqlForSize(final String sqlForSize) {

		this.sqlForSize = sqlForSize;

	}

}

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

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet-parent</artifactId>

		<version>1.0.0</version>

		<relativePath />

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

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

	<packaging>jar</packaging>

	<version>3.0.1</version>

	<scm>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

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

		</dependency>

		<dependency>

			<groupId>com.googlecode</groupId>

			<artifactId>sarasvati</artifactId>

			<version>1.0.3</version>

		</dependency>

		<dependency>

			<groupId>com.googlecode</groupId>

			<artifactId>sarasvati-visual</artifactId>

			<version>1.0.3</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

			<version>[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.repobye;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

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

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 obtainServiceId(final NodeToken token) {

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

	}

	@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.notification;

import java.util.Arrays;

import java.util.Date;

import java.util.Map;

import java.util.Properties;

import java.util.Set;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.LinkedBlockingQueue;

import javax.mail.Authenticator;

import javax.mail.Message;

import javax.mail.MessagingException;

import javax.mail.PasswordAuthentication;

import javax.mail.Session;

import javax.mail.Transport;

import javax.mail.internet.InternetAddress;

import javax.mail.internet.MimeMessage;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.google.common.base.Splitter;

import com.google.common.collect.Maps;

public class EmailDispatcher {

	private String from;

	private String fromName;

	private String cc;

	private String smtpHost;

	private int smtpPort = 587;

	private String smtpUser;

	private String smtpPassword;

	private String baseUrl;

	private String infrastructure;

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

	private final BlockingQueue<Message> queue = new LinkedBlockingQueue<Message>();

	public StringTemplate prepareMessage(final String template, final Map<String, Object> tmplParams) {

		final StringTemplate st = new StringTemplate(template);

		st.setAttributes(tmplParams);

		st.setAttribute("baseUrl", baseUrl);

		st.setAttribute("infrastructure", infrastructure);

		return st;

	}

	public void sendMail(final Set<String> to, final String subject, final String template, final Map<String, Object> tmplParams) {

		try {

			final StringTemplate st = prepareMessage(template, tmplParams);

			final Session session = Session.getInstance(obtainProperties(), obtainAuthenticator());

			final MimeMessage message = new MimeMessage(session);

			message.setFrom(new InternetAddress(from, fromName));

			message.setSubject(subject);

			message.setContent(st.toString(), "text/html; charset=utf-8");

			message.setSentDate(new Date());

			for (String s : to) {

				message.addRecipient(Message.RecipientType.TO, new InternetAddress(s));

			}

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

				for (String aCC : Splitter.on(",").omitEmptyStrings().trimResults().split(getCc())) {

					message.addRecipient(Message.RecipientType.CC, new InternetAddress(aCC));

				}

			}

			queue.add(message);

			log.info("Mail to " + Arrays.toString(to.toArray()) + " in queue");

		} catch (Exception e) {

			log.error("Error sending mail", e);

		}

	}

	public void processMailQueue() {

		while (true) {

			final Message message = queue.poll();

			if (message == null) return;

			else {

				try {

					log.info("Sending mail...");

					Transport.send(message);

					log.info("...sent");

				} catch (MessagingException e) {

					log.error("Error sending email", e);

					queue.add(message);

					return;

				}

			}

		}

	}

	private void sendWfStatusMail(final boolean success,

			final Set<String> to,

			final String wfId,

			final String procId,

			final String wfName,

			final Map<String, String> pendingWfs,

			final Map<String, String> responses,

			final String error) {

		try {

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

			map.put("wfId", wfId);

			map.put("wfName", wfName);

			map.put("procId", procId);

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

				map.put("pendingWfs", pendingWfs);

			}

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

				map.put("responses", responses);

			}

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

				map.put("error", error);

			}

			String subject = success ? "[D-NET WORKFLOW] -- SUCCESS: " : "WF FAILURE: ";

			subject += wfName + " on " + infrastructure;

			final String tmplName = success ? "wf_success.mail.st" : "wf_failed.mail.st";

			final String template = IOUtils.toString(getClass().getResourceAsStream("/eu/dnetlib/msro/mail/" + tmplName));

			sendMail(to, subject, template, map);

		} catch (Exception e) {

			log.error("Error generating success-mail", e);

		}

	}

	public void sendSuccessMail(final Set<String> to,

			final String wfId,

			final String procId,

			final String wfName,

			final Map<String, String> pendingWfs,

			final Map<String, String> responses) {

		sendWfStatusMail(true, to, wfId, procId, wfName, pendingWfs, responses, "");

	}

	public void sendFailedMail(final Set<String> to,

			final String wfId,

			final String procId,

			final String wfName,

			final Map<String, String> pendingWfs,

			final Map<String, String> responses,

			final String error) {

		sendWfStatusMail(false, to, wfId, procId, wfName, pendingWfs, responses, error);

	}

	private Properties obtainProperties() {

		final Properties props = new Properties();

		props.put("mail.transport.protocol", "smtp");

		props.put("mail.smtp.host", smtpHost);

		props.put("mail.smtp.port", smtpPort);

		props.put("mail.smtp.auth", Boolean.toString((smtpUser != null) && !smtpUser.isEmpty()));

		return props;

	}

	private Authenticator obtainAuthenticator() {

		if ((smtpUser == null) || smtpUser.isEmpty()) return null;

		return new Authenticator() {

			private final PasswordAuthentication authentication = new PasswordAuthentication(smtpUser, smtpPassword);

			@Override

			protected PasswordAuthentication getPasswordAuthentication() {

				return authentication;

			}

		};

	}

	public String getFrom() {

		return from;

	}

	@Required

	public void setFrom(final String from) {

		this.from = from;

	}

	public String getFromName() {

		return fromName;

	}

	@Required

	public void setFromName(final String fromName) {

		this.fromName = fromName;

	}