D-Net

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

import java.io.StringReader;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

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.collector.rmi.CollectorService;

import eu.dnetlib.data.collector.rmi.InterfaceDescriptor;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

public class CollectRecordsJobNode extends SimpleJobNode {

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String datasourceId;

	private String datasourceInterface;

	private String eprParam;

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

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

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

		final InterfaceDescriptor interfaceDescriptor = InterfaceDescriptor.newInstance(ifcNode);

		final W3CEndpointReference epr = serviceLocator.getService(CollectorService.class).collect(interfaceDescriptor);

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

		return Arc.DEFAULT_ARC;

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

}

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

import java.io.StringReader;

import java.util.Map;

import java.util.Set;

import javax.annotation.Resource;

import org.apache.cxf.common.util.StringUtils;

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

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

import com.google.common.base.Splitter;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.datasource.common.utils.DatasourceUpdater;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

public class RemoveApiExtraFieldsJobNode extends SimpleJobNode {

	private String datasourceId;

	private String datasourceInterface;

	private String fields;

	@Resource

	private UniqueServiceLocator serviceLocator;

	private DatasourceUpdater datasourceUpdater;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		if (!StringUtils.isEmpty(fields)) {

			final Set<String> invalidFields = Sets.newHashSet(Splitter.on(",").omitEmptyStrings().trimResults().split(fields.toLowerCase()));

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

			datasourceUpdater.updateApiExtraFields(datasourceId, datasourceInterface, map);

		}

		return Arc.DEFAULT_ARC;

	}

	private Map<String, String> calculateValidExtraFields(final String repoId, final String ifaceId, final Set<String> invalidFields) throws Exception {

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

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

		final SAXReader reader = new SAXReader();

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

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

		if (ifcNode != null) {

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

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

				if (!invalidFields.contains(name.toLowerCase())) {

					res.put(name, ((Node) o).getText());

				}

			}

		}

		return res;

	}

	public String getFields() {

		return fields;

	}

	public void setFields(final String fields) {

		this.fields = fields;

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

	public DatasourceUpdater getDatasourceUpdater() {

		return datasourceUpdater;

	}

	@Required

	public void setDatasourceUpdater(final DatasourceUpdater datasourceUpdater) {

		this.datasourceUpdater = datasourceUpdater;

	}

}

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

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

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet-parent</artifactId>

		<version>1.0.0</version>

		<relativePath />

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

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

	<packaging>jar</packaging>

	<version>2.1.1</version>

	<scm>

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

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

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

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

import org.springframework.core.io.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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.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 MdBuilderJobNode extends SimpleJobNode {

	private Resource mdBuilderTemplateXslt;

	private String inputEprParam;

	private String outputEprParam;

	private String datasourceId;

	private String datasourceInterface;

	private XSLTMappedResultSetFactory xsltMappedResultSetFactory;

	@javax.annotation.Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws MSROException {

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

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

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

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

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

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

		try {

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

			st.setAttribute("xpath", lookupService.getResourceProfileByQuery(xQuery));

			st.setAttribute("datasourceId", datasourceId);

			/*

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

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

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

			 */

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

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

			} else {

				List<String> namespacePrefix;

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

				namespacePrefix = lookupService.quickSearchProfile(xQuery);

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

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

				} else {

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

					namespacePrefix = lookupService.quickSearchProfile(xQuery);

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

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

					} else {

						st.setAttribute("namespacePrefix", datasourceId);

					}

				}

			}

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

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

			return Arc.DEFAULT_ARC;

		} catch (ISLookUpDocumentNotFoundException e) {

			throw new MSROException("Missing metadata_identifier_path extra field for ds: " + datasourceId, e);

		} catch (ISLookUpException e) {

			throw new MSROException("Error searching metadata_identifier_path extra field for ds: " + datasourceId, e);

		} catch (IOException e) {

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

		}

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public XSLTMappedResultSetFactory getXsltMappedResultSetFactory() {

		return xsltMappedResultSetFactory;

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

	@Required

	public void setXsltMappedResultSetFactory(final XSLTMappedResultSetFactory xsltMappedResultSetFactory) {

		this.xsltMappedResultSetFactory = xsltMappedResultSetFactory;

	}

	public Resource getMdBuilderTemplateXslt() {

		return mdBuilderTemplateXslt;

	}

	@Required

	public void setMdBuilderTemplateXslt(final Resource mdBuilderTemplateXslt) {

		this.mdBuilderTemplateXslt = mdBuilderTemplateXslt;

	}

}

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

import java.util.Map;

import eu.dnetlib.miscutils.functional.UnaryFunction;

public abstract class GroovyUnaryFunction implements UnaryFunction<String, String> {

	private Map<String, String> params;

	@Override

	abstract public String evaluate(String input);

	public Map<String, String> getParams() {

		return params;

	}

	public void setParams(final Map<String, String> params) {

		this.params = params;

	}

}

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 ? "Workflow '" + wfName + "' has been completed successfully" : "Workflow '" + wfName + "' is failed";

			subject += " 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;

	}

	public String getCc() {

		return cc;

	}

	@Required

	public void setCc(final String cc) {

		this.cc = cc;

	}

	public String getSmtpHost() {

		return smtpHost;

	}

	@Required

	public void setSmtpHost(final String smtpHost) {

		this.smtpHost = smtpHost;

	}

	public int getSmtpPort() {

		return smtpPort;

	}

	public void setSmtpPort(final int smtpPort) {

		this.smtpPort = smtpPort;

	}

	public String getSmtpUser() {

		return smtpUser;

	}

	public void setSmtpUser(final String smtpUser) {

		this.smtpUser = smtpUser;

	}

	public String getSmtpPassword() {

		return smtpPassword;

	}

	public void setSmtpPassword(final String smtpPassword) {

		this.smtpPassword = smtpPassword;

	}

	public String getBaseUrl() {

		return baseUrl;

	}

	@Required

	public void setBaseUrl(final String baseUrl) {

		this.baseUrl = baseUrl;

	}

	public String getInfrastructure() {

		return infrastructure;

	}

	@Required

	public void setInfrastructure(final String infrastructure) {

		this.infrastructure = infrastructure;

	}

}

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

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

			}

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

import java.io.StringReader;

import java.util.Map;

import java.util.Set;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

import com.google.common.base.Splitter;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import eu.dnetlib.enabling.actions.AbstractSubscriptionAction;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.workflows.sarasvati.loader.WorkflowExecutor;

public class WfDependencyLauncherNotificationHandler extends AbstractSubscriptionAction {

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Resource

	private WorkflowExecutor workflowExecutor;

	@Resource

	private EmailDispatcher emailDispatcher;

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

	@Override

	public void notified(final String subscrId, final String topic, final String rsId, final String profile) {

		final SAXReader reader = new SAXReader();

		try {

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

			final Set<String> emails = calculateEmails(rsId);

			final String procId = doc.valueOf("//LAST_EXECUTION_ID");

			final String wfName = doc.valueOf("//WORKFLOW_NAME");

			final boolean success = doc.valueOf("//LAST_EXECUTION_STATUS").equals("SUCCESS");

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

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

				Node n = (Node) o;

				responses.put(n.valueOf("@name"), n.getText());

			}

			if (!success) {

				log.info("Last execution of " + rsId + " failed, dependencies NOT STARTED");

			}

			final String query = "for $x in collection('/db/DRIVER/MetaWorkflowDSResources/MetaWorkflowDSResourceType')//WORKFLOW[@id='" + rsId

					+ "']/WORKFLOW let $y := /RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = $x/@id] "

					+ "where $y//CONFIGURATION/@start != 'disabled' return concat ($x/@id , ' @@@ ', $x/@name , ' @@@ ', $y//CONFIGURATION/@start)";

			try {

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

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

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

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

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

					final boolean manual = arr[2].trim().toLowerCase().equals("manual");

					if (success && !manual) {

						try {

							String pid = workflowExecutor.startProcess(id);

							log.info("PROC " + pid + " of WF " + id + " STARTED AS CHILD OF " + rsId);

						} catch (Exception e) {

							log.error("Error starting wf: " + id);

						}

					} else {

						pendingWfs.put(id, name);

					}

				}

				if (!emails.isEmpty()) {

					if (success) {

						emailDispatcher.sendSuccessMail(emails, rsId, procId, wfName, pendingWfs, responses);

					} else {

						final String error = doc.valueOf("//LAST_EXECUTION_ERROR");

						emailDispatcher.sendFailedMail(emails, rsId, procId, wfName, pendingWfs, responses, error);

					}

				}

			} catch (ISLookUpException e) {

				log.error("Error executing xquery: " + query, e);

			}

		} catch (DocumentException e) {

			log.error("Error parsing profile with id " + rsId + ": " + profile);

		}

	}

	private Set<String> calculateEmails(final String id) {

		final Set<String> list = Sets.newHashSet();

		try {

			for (String val : serviceLocator.getService(ISLookUpService.class).quickSearchProfile("//ADMIN_EMAIL[..//WORKFLOW/@id='" + id + "']/text()")) {

				for (String s : Splitter.on(",").trimResults().omitEmptyStrings().split(val)) {

					list.add(s);

				}

			}

		} catch (Exception e) {

			log.error("Error searching email adresses", e);

		}

		return list;

	}

}

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

import java.util.List;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

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

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.enabling.resultset.IterableResultSetFactory;

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

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

public class FetchMultipleMDStores extends SimpleJobNode {

	@Resource

	private UniqueServiceLocator serviceLocator;

	private List<String> mdId;

	private String mdFormat;

	private String eprParam;

	/** The result set factory. */

	@Resource(name = "iterableResultSetFactory")

	private IterableResultSetFactory resultSetFactory;

	/** The result set client factory. */

	@Autowired

	private ResultSetClientFactory resultSetClientFactory;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		if (getMdId() == null) {

			@SuppressWarnings("unchecked")

			List<String> mdIds = new Gson().fromJson(token.getEnv().getAttribute("mdId"), List.class);

			setMdId(mdIds);

		}

		if (getMdFormat() == null) {

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

		}

		MultipleMdStoreIterator iterators = new MultipleMdStoreIterator(serviceLocator, getMdId(), resultSetClientFactory);

		W3CEndpointReference eprUrls = resultSetFactory.createIterableResultSet(iterators);

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

		return Arc.DEFAULT_ARC;

	}

	public List<String> getMdId() {

		return mdId;

	}

	public void setMdId(final List<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.objectStore;

import java.io.StringReader;

import java.util.Iterator;

import java.util.Map;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import javax.xml.xpath.XPath;

import javax.xml.xpath.XPathFactory;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.xml.sax.InputSource;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

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

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

import eu.dnetlib.enabling.resultset.IterableResultSetFactory;

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

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 DownloadIntoObjectStoreJobNode extends BlackboardJobNode implements ProgressJobNode {

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

	class MetadataObjectIterator implements Iterator<String> {

		private Iterator<String> inputIterator;

		private String mime;

		public MetadataObjectIterator(final Iterator<String> inputIterator, final String xpath, final String mime) {

			this.inputIterator = inputIterator;

		}

		@Override

		public boolean hasNext() {

			return inputIterator.hasNext();

		}

		@Override

		public String next() {

			try {

				String record = inputIterator.next();

				XPath xpath = XPathFactory.newInstance().newXPath();

				InputSource doc = new InputSource(new StringReader(record));

				String identifier = xpath.evaluate(getIdXpath(), doc);

				MetadataObjectRecord objectrecord = new MetadataObjectRecord(identifier, record, mime);

				return objectrecord.toJSON();

			} catch (Exception e) {

				return null;

			}

		}

		@Override

		public void remove() {

		}

	}

	private String eprParam;

	private String objectStoreId;

	private String idXpath; // "//*[local-name()='objIdentifier']

	private String contentDescription;

	private String objectIsInsideEpr;

	private IterableResultSetFactory iterableResultSetFactory;

	private ResultSetClientFactory resultSetClientFactory;

	private ResultsetProgressProvider progressProvider;

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

	public String getObjectStoreId() {

		return objectStoreId;

	}

	public void setObjectStoreId(final String objectStoreId) {

		this.objectStoreId = objectStoreId;

	}

	public ProgressProvider getProgressProvider(final NodeToken token) {

		return progressProvider;

	}

	@Override

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

		job.setAction("FEEDOBJECT");

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

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

		job.getParameters().put("mime", getContentDescription());

		final Iterator<String> client = resultSetClientFactory.getClient(eprS).iterator();

		final W3CEndpointReference epr = iterableResultSetFactory.createIterableResultSet(new Iterable<String>() {

			@Override

			public Iterator<String> iterator() {

				return new MetadataObjectIterator(client, "//*[local-name()='objIdentifier']", "xml");

			}

		});

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

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

	}

	@Override

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

		return new BlackboardWorkflowJobListener(engine, token) {