D-Net

{"type_source": "SVN", "goal": "package -U -T 4C source:jar", "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-msro-service/trunk/", "deploy_repository": "dnet4-snapshots", "version": "4", "mail": "sandro.labruzzo@isti.cnr.it,michele.artini@isti.cnr.it, claudio.atzori@isti.cnr.it, alessia.bardi@isti.cnr.it", "deploy_repository_url": "http://maven.research-infrastructures.eu/nexus/content/repositories/dnet4-snapshots", "name": "dnet-msro-service"}

package eu.dnetlib;

import java.io.StringWriter;

import org.junit.Test;

import org.w3c.dom.Node;

import org.w3c.dom.NodeList;

import eu.dnetlib.miscutils.functional.xml.ApplyXslt;

/**

 * Unit test for simple App.

 */

public class AppTest {

	@Test

	public void testXslt() {

		String res = (new ApplyXslt(getXslt())).evaluate(getXml());

		System.out.println("******************");

		System.out.println(res);

		System.out.println("******************");

	}

	public static String javaMethod(NodeList list) {

		System.out.println("******************");

		System.out.println("TYPE   : " + list.toString());

		System.out.println("LENGTH : " + list.getLength());

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

			Node node = list.item(i);

			System.out.println("ELEM " + i + ": " + node.getLocalName());

		}

		System.out.println("******************");

		return "SUCA";

	}

	private String getXml() {

		final StringWriter sw = new StringWriter();

		sw.append("<?xml version='1.0' encoding='UTF-8'?>\n");

		sw.append("<record>\n");

		sw.append("<metadata>\n");

		sw.append("<a>A text value</a>\n");

		sw.append("<b attr='attribute value'/>\n");

		sw.append("</metadata>\n");

		sw.append("</record>\n");

		return sw.toString();

	}

	private String getXslt() {

		final StringWriter sw = new StringWriter();

		sw.append("<?xml version='1.0' encoding='UTF-8'?>\n");

		sw.append("<xsl:stylesheet version='1.0' xmlns:xsl='http://www.w3.org/1999/XSL/Transform' xmlns:dnet='eu.dnetlib.AppTest' exclude-result-prefixes='xsl dnet'>\n");

		sw.append("<xsl:output omit-xml-declaration='yes' indent='yes'/>\n");

		sw.append("<xsl:template match='/*'>\n");

		sw.append("<xsl:variable name='metadata' select=\"//*[local-name()='metadata']/*\" />\n");

		sw.append("<ROWS>\n");

		sw.append("<xsl:value-of select='dnet:javaMethod($metadata)'/>\n");

		sw.append("</ROWS>\n");

		sw.append("</xsl:template>\n");

		sw.append("</xsl:stylesheet>\n");

		return sw.toString();

	}

}

package eu.dnetlib.msro.cron;

import java.util.Date;

import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.quartz.CronExpression;

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

import com.googlecode.sarasvati.GraphProcess;

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.miscutils.datetime.DateUtils;

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

import eu.dnetlib.msro.workflows.sarasvati.registry.GraphProcessRegistry;

public class ScheduledWorkflowLauncher {

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

	private static final DateUtils dateUtils = new DateUtils();

	private WorkflowExecutor workflowExecutor;

	private GraphProcessRegistry graphProcessRegistry;

	private int windowSize = 7200000; // 2 hours

	@Resource

	private UniqueServiceLocator serviceLocator;

	public void verifySheduledWorkflows() {

		log.debug("Verifying scheduled workflows - START");

		final String query = "for $x in collection('/db/DRIVER/MetaWorkflowDSResources/MetaWorkflowDSResourceType') " +

				"where $x//CONFIGURATION/@status='EXECUTABLE' " +

				"and $x//SCHEDULING/@enabled='true' " +

				"return concat($x//RESOURCE_IDENTIFIER/@value, ' @@@ ', $x//SCHEDULING/CRON, ' @@@ ', $x//SCHEDULING/MININTERVAL)";

		try {

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

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

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

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

				final int minInterval = NumberUtils.toInt(arr[2].trim(), 0) * 60000; // MINUTES to MILLIS

				if (CronExpression.isValidExpression(cron)) {

					final Date now = new Date();

					final Date last = calculateLastExecutionDate(id);

					final int elapsed = Math.abs(Math.round(now.getTime() - last.getTime()));

					if (log.isDebugEnabled()) {

						log.debug("**************************************************************");

						log.debug("META WORKFLOW ID   : " + id);

						log.debug("NOW                : " + now);

						log.debug("LAST EXECUTION DATE: " + last);

						log.debug("MIN INTERVAL       : " + minInterval);

						log.debug("WINDOW SIZE        : " + windowSize);

						log.debug("TIME ELAPSED       : " + elapsed);

					}

					if (elapsed > minInterval && isFired(cron, last, now) && !isAlreadyRunning(id)) {

						log.debug("MUST BE EXECUTED   : true");

						try {

							workflowExecutor.startMetaWorkflow(id, false);

						} catch (Exception e) {

							log.error("Error launching scheduled wf: " + id, e);

						}

					} else {

						log.debug("MUST BE EXECUTED   : false");

					}

					log.debug("**************************************************************");

				}

			}

		} catch (ISLookUpException e) {

			log.error("Error executing query " + query);

		}

		log.debug("Verifying scheduled workflows - END");

	}

	private boolean isFired(final String cronExpression, final Date startDate, final Date now) {

		try {

			final CronExpression cron = new CronExpression(cronExpression);

			final Date prev = new Date(now.getTime() - windowSize);

			final Date date = prev.getTime() < startDate.getTime() ? startDate : prev;

			final Date next = cron.getNextValidTimeAfter(date);

			if (log.isDebugEnabled()) {

				log.debug("NEXT EXECUTION DATE: " + next);

				log.debug("FIRED              : " + (next.getTime() < now.getTime()));

			}

			return next.getTime() < now.getTime();

		} catch (Exception e) {

			log.error("Error calculating next cron event: " + cronExpression, e);

			return false;

		}

	}

	private boolean isAlreadyRunning(final String metaWfId) {

		final String query = "document('/db/DRIVER/MetaWorkflowDSResources/MetaWorkflowDSResourceType/" + StringUtils.substringBefore(metaWfId, "_")

				+ "')//WORKFLOW/@id/string()";

		try {

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

				if (profileId.length() > 0) {

					for (GraphProcess p : graphProcessRegistry.findProcessesByResource(profileId)) {

						switch (p.getState()) {

						case Created:

							return true;

						case Executing:

							return true;

						default:

							break;

						}

					}

				}

			}

		} catch (ISLookUpException e) {

			log.error("Error executing query " + query);

		}

		return false;

	}

	private Date calculateLastExecutionDate(final String id) {

		final String query = "for $id in document('/db/DRIVER/MetaWorkflowDSResources/MetaWorkflowDSResourceType/" + StringUtils.substringBefore(id, "_")

				+ "')//WORKFLOW/@id/string() " +

				"for $x in document(concat('/db/DRIVER/WorkflowDSResources/WorkflowDSResourceType/', substring-before($id, '_'))) " +

				"where $x//LAST_EXECUTION_STATUS = 'SUCCESS' " +

				"return $x//LAST_EXECUTION_DATE/text() ";

		long time = 0;

		try {

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

				if (s.length() > 0) {

					final Date d = dateUtils.parse(s);

					if (time < d.getTime()) {

						time = d.getTime();

					}

				}

			}

		} catch (ISLookUpException e) {

			log.error("Error executing query " + query);

		} catch (Exception e) {

			log.error("Error calculating date", e);

		}

		return new Date(time);

	}

	public WorkflowExecutor getWorkflowExecutor() {

		return workflowExecutor;

	}

	@Required

	public void setWorkflowExecutor(final WorkflowExecutor workflowExecutor) {

		this.workflowExecutor = workflowExecutor;

	}

	public GraphProcessRegistry getGraphProcessRegistry() {

		return graphProcessRegistry;

	}

	@Required

	public void setGraphProcessRegistry(final GraphProcessRegistry graphProcessRegistry) {

		this.graphProcessRegistry = graphProcessRegistry;

	}

	public int getWindowSize() {

		return windowSize;

	}

	@Required

	public void setWindowSize(final int windowSize) {

		this.windowSize = windowSize;

	}

}

package eu.dnetlib.msro;

import javax.jws.WebService;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import eu.dnetlib.enabling.tools.AbstractBaseService;

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

import eu.dnetlib.msro.rmi.MSROService;

import eu.dnetlib.msro.workflows.sarasvati.registry.GraphProcessRegistry;

@WebService(targetNamespace = "http://services.dnetlib.eu/")

public class MSROServiceImpl extends AbstractBaseService implements MSROService {

	/**

	 * logger.

	 */

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

	/**

	 * notification handler.

	 */

	private NotificationHandler notificationHandler;

	/**

	 * graph process registry.

	 */

	private GraphProcessRegistry processRegistry;

	@Override

	public void notify(final String subscriptionId, final String topic, final String isId, final String message) {

		super.notify(subscriptionId, topic, isId, message);

		log.debug("got notification: " + topic);

		getNotificationHandler().notified(subscriptionId, topic, isId, message);

	}

	@Required

	public void setNotificationHandler(final NotificationHandler notHandler) {

		this.notificationHandler = notHandler;

	}

	public NotificationHandler getNotificationHandler() {

		return notificationHandler;

	}

	@Required

	public void setProcessRegistry(final GraphProcessRegistry processRegistry) {

		this.processRegistry = processRegistry;

	}

	public GraphProcessRegistry getProcessRegistry() {

		return processRegistry;

	}

}

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.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 static final Log log = LogFactory.getLog(EmailDispatcher.class);

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

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

		try {

			final StringTemplate st = new StringTemplate(template);

			st.setAttributes(tmplParams);

			st.setAttribute("baseUrl", baseUrl);

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

			}

			final String subject = success ? "Workflow '" + wfName + "' has been completed successfully" : "Workflow '" + wfName + "' is failed";

			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;

	}

}

package eu.dnetlib.msro.workflows.metawf;

import org.springframework.context.ApplicationContext;

import org.springframework.context.ApplicationContextAware;

public class DatasourceMetaWorkflowFactory implements ApplicationContextAware {

	private transient ApplicationContext applicationContext;

	public DatasourceMetaWorkflow newMetaWorkflow(final String beanName) {

		final DatasourceMetaWorkflow prototypeMetaWf = (DatasourceMetaWorkflow) applicationContext.getBean(beanName, DatasourceMetaWorkflow.class);

		if (prototypeMetaWf != null) {

			return prototypeMetaWf;

		} else {

			throw new IllegalArgumentException("cannot find bean " + beanName);

		}

	}

	@Override

	public void setApplicationContext(final ApplicationContext context) {

		this.applicationContext = context;

	}

}

package eu.dnetlib.msro.workflows.metawf;

import java.io.IOException;

import java.io.StringWriter;

import java.util.List;

import java.util.Map;

import javax.annotation.Resource;

import org.apache.commons.lang.StringEscapeUtils;

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

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

public class WorkflowTree {

	private String id;

	private String name;

	private String start = "auto";

	private List<WorkflowTree> children;

	private org.springframework.core.io.Resource template;

	@Resource

	private UniqueServiceLocator serviceLocator;

	public void populateMetaWfXml(final StringWriter sw) {

		sw.append("<WORKFLOW id='" + StringEscapeUtils.escapeXml(id) + "' name='" + StringEscapeUtils.escapeXml(name) + "'");

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

			sw.append(" />");

		} else {

			sw.append(">");

			for (WorkflowTree child : children) {

				child.populateMetaWfXml(sw);

			}

			sw.append("</WORKFLOW>");

		}

	}

	public int registerAllWorkflows(final Map<String, String> params) throws ISRegistryException, IOException {

		int count = 0;

		if (this.id == null || this.id.isEmpty()) {

			final String profile = WorkflowProfileCreator.generateProfile(name, "aggregator", params, template);

			this.id = serviceLocator.getService(ISRegistryService.class).registerProfile(profile);

			count++;

		}

		if (children != null) {

			for (WorkflowTree child : children) {

				count += child.registerAllWorkflows(params);

			}

		}

		return count;

	}

	public String getId() {

		return id;

	}

	public List<WorkflowTree> getChildren() {

		return children;

	}

	public void setChildren(final List<WorkflowTree> children) {

		this.children = children;

	}

	public String getName() {

		return name;

	}

	@Required

	public void setName(final String name) {

		this.name = name;

	}

	public org.springframework.core.io.Resource getTemplate() {

		return template;

	}

	@Required

	public void setTemplate(final org.springframework.core.io.Resource template) {

		this.template = template;

	}

	public String getStart() {

		return start;

	}

	public void setStart(final String start) {

		this.start = start;

	}

}

package eu.dnetlib.msro.workflows.metawf;

import java.io.IOException;

import java.util.Map;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import org.springframework.core.io.ClassPathResource;

import org.springframework.core.io.Resource;

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

public class WorkflowProfileCreator {

	private static final Resource wfTemplate = new ClassPathResource("/eu/dnetlib/msro/workflows/templates/workflow.xml.st");

	public static String generateProfile(final String name, final String type, final Map<String, String> params, final Resource confTemplate)

			throws IOException {

		final StringTemplate conf = new StringTemplate(IOUtils.toString(confTemplate.getInputStream()));

		conf.setAttribute("params", params);

		final StringTemplate profile = new StringTemplate(IOUtils.toString(wfTemplate.getInputStream()));

		profile.setAttribute("name", name);

		profile.setAttribute("type", type);

		profile.setAttribute("priority", WorkflowsConstants.DEFAULT_WF_PRIORITY);

		profile.setAttribute("conf", conf.toString());

		return profile.toString();

	}

}

package eu.dnetlib.msro.workflows.metawf;

import java.io.IOException;

import java.io.StringWriter;

import java.util.Map;

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

import org.springframework.core.io.Resource;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

public class DatasourceMetaWorkflow {

	private WorkflowTree tree;

	private Resource destroyWorkflowTemplate;

	@javax.annotation.Resource

	private UniqueServiceLocator serviceLocator;

	public int registerAllWorkflows(final Map<String, String> params) throws ISRegistryException, IOException {

		return tree.registerAllWorkflows(params);

	}

	public String registerDestroyWorkflow(final Map<String, String> params) throws ISRegistryException, IOException {

		final String profile = WorkflowProfileCreator.generateProfile("Repo BYE", "REPO_BYE", params, destroyWorkflowTemplate);

		return serviceLocator.getService(ISRegistryService.class).registerProfile(profile);

	}

	public String asXML() {

		final StringWriter sw = new StringWriter();

		tree.populateMetaWfXml(sw);

		return sw.toString();

	}

	public WorkflowTree getTree() {

		return tree;

	}

	@Required

	public void setTree(final WorkflowTree tree) {

		this.tree = tree;

	}

	public Resource getDestroyWorkflowTemplate() {

		return destroyWorkflowTemplate;

	}

	@Required

	public void setDestroyWorkflowTemplate(final Resource destroyWorkflowTemplate) {

		this.destroyWorkflowTemplate = destroyWorkflowTemplate;

	}

}

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

import java.io.StringReader;

import java.util.Iterator;

import java.util.Queue;

import java.util.concurrent.PriorityBlockingQueue;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

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 org.springframework.beans.factory.annotation.Required;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.resultset.IterableResultSetFactory;

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

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

public class UnpackJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	private String inputEprParam;

	private String outputEprParam;

	private String xpath;

	private IterableResultSetFactory iterableResultSetFactory;

	private ResultSetClientFactory resultSetClientFactory;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final Iterator<String> client = resultSetClientFactory.getClient(token.getEnv().getAttribute(inputEprParam)).iterator();

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

		if (client.hasNext()) {

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

		}

		final W3CEndpointReference epr = iterableResultSetFactory.createIterableResultSet(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().setAttribute(outputEprParam, epr.toString());

		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 IterableResultSetFactory getIterableResultSetFactory() {

		return iterableResultSetFactory;

	}

	@Required

	public void setIterableResultSetFactory(final IterableResultSetFactory iterableResultSetFactory) {

		this.iterableResultSetFactory = iterableResultSetFactory;

	}

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	@Required

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

	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 getXpath() {

		return xpath;

	}

	public void setXpath(final String xpath) {

		this.xpath = xpath;

	}

}

package eu.dnetlib.msro.workflows.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.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.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.workflows.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 UniqueServiceLocator serviceLocator;

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.workflows.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 Gson();

		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 = "concat(//RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value=\"%s\"]//METADATA_FORMAT/text(), \"::<<>>::\""

				+ ",//RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value=\"%s\"]//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, 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.workflows.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.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

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

public class UpdateObjectStoreSizeJobNode extends SimpleJobNode {

	/** The obejct store id. */

	private String obejctStoreIdParam;

	/** The service locator. */

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

		int size = serviceLocator.getService(ObjectStoreService.class, obejctStoreIdParam).getSize(obejctStoreIdParam);

		String now = DateUtils.now_ISO8601();

		String mdstoreXUpdate = "for $x in //RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '" + obejctStoreIdParam + "']"

				+ "return update value $x//LAST_STORAGE_DATE with '" + now + "'";

		registry.executeXUpdate(mdstoreXUpdate);

		String mdstoreNumberXUpdate = "for $x in //RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '" + obejctStoreIdParam + "']"

				+ "return update value $x//COUNT_STORE with '" + size + "'";

		registry.executeXUpdate(mdstoreNumberXUpdate);

		mdstoreNumberXUpdate = "for $x in //RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '" + obejctStoreIdParam + "']"

				+ "return update value $x//STORE_SIZE with '" + size + "'";

		registry.executeXUpdate(mdstoreNumberXUpdate);

		return Arc.DEFAULT_ARC;

	}

	/**

	 * @return the obejctStoreIdParam

	 */

	public String getObejctStoreIdParam() {

		return obejctStoreIdParam;

	}

	/**

	 * @param obejctStoreIdParam

	 *            the obejctStoreIdParam to set

	 */

	public void setObejctStoreIdParam(final String obejctStoreIdParam) {

		this.obejctStoreIdParam = obejctStoreIdParam;

	}

}

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

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

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

	}

	@Override

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

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

			@Override

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

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

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

			}

		};

	}

	public String getObjectIsInsideEpr() {

		return objectIsInsideEpr;

	}

	public void setObjectIsInsideEpr(final String objectIsInsideEpr) {

		this.objectIsInsideEpr = objectIsInsideEpr;

	}

	@Override

	public ResultsetProgressProvider getProgressProvider() {

		return progressProvider;

	}

	public void setProgressProvider(final ResultsetProgressProvider progressProvider) {

		this.progressProvider = progressProvider;

	}

	public ProcessCountingResultSetFactory getProcessCountingResultSetFactory() {

		return processCountingResultSetFactory;

	}

	public void setProcessCountingResultSetFactory(final ProcessCountingResultSetFactory processCountingResultSetFactory) {

		this.processCountingResultSetFactory = processCountingResultSetFactory;

	}

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

	public IterableResultSetFactory getIterableResultSetFactory() {

		return iterableResultSetFactory;

	}