D-Net

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

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

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

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

import com.google.common.collect.Iterables;

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.enabling.resultset.client.utils.EPRUtils;

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

import eu.dnetlib.msro.workflows.nodes.download.UrlExtractor;

// TODO: Auto-generated Javadoc

/**

 * The Class RetrieveURLSJobNode.

 */

public class RetrieveURLSJobNode extends SimpleJobNode {

	/** The epr param. */

	private String inputEprParam;

	/** The output epr param. */

	private String outputEprParam;

	/** The xpath. */

	private String xpath;

	/** The xpath metadata id. */

	private String xpathMetadataId;

	/** The xpath open access. */

	private String xpathOpenAccess;

	/** The xpath embargo date. */

	private String xpathEmbargoDate;

	/** The result set client factory. */

	@Autowired

	private ResultSetClientFactory resultSetClientFactory;

	/** The result set factory. */

	@Resource(name = "iterableResultSetFactory")

	private IterableResultSetFactory resultSetFactory;

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

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

		Iterable<String> extractedUrls = Iterables.transform(input, new UrlExtractor(xpath, xpathMetadataId, xpathOpenAccess, xpathEmbargoDate));

		W3CEndpointReference eprUrls = resultSetFactory.createIterableResultSet(extractedUrls);

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

		return Arc.DEFAULT_ARC;

	}

	/**

	 * Gets the xpath.

	 *

	 * @return the xpath

	 */

	public String getXpath() {

		return xpath;

	}

	/**

	 * Sets the xpath.

	 *

	 * @param xpath

	 *            the xpath to set

	 */

	public void setXpath(final String xpath) {

		this.xpath = xpath;

	}

	/**

	 * Gets the xpath metadata id.

	 *

	 * @return the xpathMetadataId

	 */

	public String getXpathMetadataId() {

		return xpathMetadataId;

	}

	/**

	 * Sets the xpath metadata id.

	 *

	 * @param xpathMetadataId

	 *            the xpathMetadataId to set

	 */

	@Required

	public void setXpathMetadataId(final String xpathMetadataId) {

		this.xpathMetadataId = xpathMetadataId;

	}

	/**

	 * Gets the result set client factory.

	 *

	 * @return the resultSetClientFactory

	 */

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	/**

	 * Sets the result set client factory.

	 *

	 * @param resultSetClientFactory

	 *            the resultSetClientFactory to set

	 */

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

	/**

	 * Gets the result set factory.

	 *

	 * @return the resultSetFactory

	 */

	public IterableResultSetFactory getResultSetFactory() {

		return resultSetFactory;

	}

	/**

	 * Sets the result set factory.

	 *

	 * @param resultSetFactory

	 *            the resultSetFactory to set

	 */

	public void setResultSetFactory(final IterableResultSetFactory resultSetFactory) {

		this.resultSetFactory = resultSetFactory;

	}

	/**

	 * Gets the output epr param.

	 *

	 * @return the outputEprParam

	 */

	public String getOutputEprParam() {

		return outputEprParam;

	}

	/**

	 * Sets the output epr param.

	 *

	 * @param outputEprParam

	 *            the outputEprParam to set

	 */

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	/**

	 * Gets the input epr param.

	 *

	 * @return the inputEprParam

	 */

	public String getInputEprParam() {

		return inputEprParam;

	}

	/**

	 * Sets the input epr param.

	 *

	 * @param inputEprParam

	 *            the inputEprParam to set

	 */

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	/**

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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

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

/**

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

 */

public class SelectiveNode extends SimpleJobNode {

	/** The selection. */

	private String selection = Arc.DEFAULT_ARC;

	/*

	 * (non-Javadoc)

	 * 

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

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "selection", getSelection());

		return getSelection();

	}

	/**

	 * Gets the selection.

	 *

	 * @return the selection

	 */

	public String getSelection() {

		return selection;

	}

	/**

	 * Sets the selection.

	 *

	 * @param selection

	 *            the new selection

	 */

	public void setSelection(final String selection) {

		this.selection = selection;

	}

}

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

import java.util.HashMap;

import java.util.Map;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

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

import org.springframework.core.io.ClassPathResource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.resultset.XSLTMappedResultSetFactory;

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

import eu.dnetlib.msro.rmi.MSROException;

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

public class ApplyXsltJobNode extends SimpleJobNode {

	private String inputEprParam;

	private String outputEprParam;

	private String xsltClasspath;

	private XSLTMappedResultSetFactory xsltMappedResultSetFactory;

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

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

		final Map<String, String> xsltParams = new HashMap<String, String>();

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

			xsltParams.put(name.replaceAll(":", "_"), token.getFullEnv().getAttribute(name));

		}

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

			xsltParams.put(name.replaceAll(":", "_"), token.getEnv().getAttribute(name));

		}

		xsltParams.putAll(parseJsonParameters(token));

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

				(new ClassPathResource(xsltClasspath)), xsltParams);

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

		return Arc.DEFAULT_ARC;

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public String getXsltClasspath() {

		return xsltClasspath;

	}

	public void setXsltClasspath(final String xsltClasspath) {

		this.xsltClasspath = xsltClasspath;

	}

	public XSLTMappedResultSetFactory getXsltMappedResultSetFactory() {

		return xsltMappedResultSetFactory;

	}

	@Required

	public void setXsltMappedResultSetFactory(final XSLTMappedResultSetFactory xsltMappedResultSetFactory) {

		this.xsltMappedResultSetFactory = xsltMappedResultSetFactory;

	}

}

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

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

public class CleanJobNode extends SimpleJobNode {

	private String inputEprParam;

	private String outputEprParam;

	private String ruleId;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

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

				ruleId);

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

		return Arc.DEFAULT_ARC;

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public String getRuleId() {

		return ruleId;

	}

	public void setRuleId(final String ruleId) {

		this.ruleId = ruleId;

	}

}

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

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

public class TransformJobNode extends SimpleJobNode {

	private String inputEprParam;

	private String outputEprParam;

	private String ruleId;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

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

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

		return Arc.DEFAULT_ARC;

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public String getRuleId() {

		return ruleId;

	}

	public void setRuleId(final String ruleId) {

		this.ruleId = ruleId;

	}

}

package eu.dnetlib.msro.workflows.nodes.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 javax.annotation.Resource;

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;

import org.apache.commons.lang.StringUtils;

public class ExecuteSqlFromEnvJobNode extends AsyncJobNode {

	private String db;

	private String dbParam;

	private String dbProperty;

	private String sqlParamName;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		String sql = token.getEnv().getAttribute(sqlParamName);

		if (StringUtils.isBlank(sql)) throw new IllegalArgumentException("Missing value in env attribute named: " + sqlParamName);

		serviceLocator.getService(DatabaseService.class).updateSQL(findDb(token), sql);

		return Arc.DEFAULT_ARC;

	}

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

		return dbProperty;

	}

	public void setDbProperty(final String dbProperty) {

		this.dbProperty = dbProperty;

	}

	public String getSqlParamName() {

		return sqlParamName;

	}

	public void setSqlParamName(final String sqlParamName) {

		this.sqlParamName = sqlParamName;

	}

}

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

		while (queue.isEmpty() && 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.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;

	}

}

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

import java.io.IOException;

import java.io.StringReader;

import java.io.StringWriter;

import javax.annotation.Resource;

import javax.xml.transform.Transformer;

import javax.xml.transform.TransformerConfigurationException;

import javax.xml.transform.TransformerException;

import javax.xml.transform.TransformerFactory;

import javax.xml.transform.dom.DOMResult;

import javax.xml.transform.dom.DOMSource;

import javax.xml.transform.stream.StreamResult;

import javax.xml.transform.stream.StreamSource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.codec.binary.Base64;

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.data.provision.index.rmi.IndexService;

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

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

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

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

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

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

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

public class UpdateIndexJobNode extends BlackboardJobNode implements ProgressJobNode {

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

	private String eprParam;

	private String indexId;

	private String format;

	private String layout;

	private String feedingType;

	private String defaultIndexId;

	/**

	 * xslt mapped resultset factory.

	 */

	private XSLTMappedResultSetFactory xsltRSFactory;

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	private ResultsetProgressProvider progressProvider;

	/**

	 * Stylesheet which transforms a layout to another stylesheet which converts a input record to a index record.

	 */

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

	/**

	 * service locator.

	 */

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		log.info("preparing blackboard job update index: " + getIndexId());

		final W3CEndpointReference epr = new EPRUtils().getEpr(token.getEnv().getAttribute(getEprParam()));

		final W3CEndpointReference mappedEpr = prepareForIndexing(epr, getFormat(), getLayout());

		progressProvider = processCountingResultSetFactory.createProgressProvider(token.getProcess(), mappedEpr);

		job.setAction("FEED");

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

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

		job.getParameters().put("feeding_type", getFeedingType());

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

	}

	// helpers

	/**

	 * Transforms each mdstore record into a index record.

	 * 

	 * @param mdStoreRsetEpr

	 *            mdstore resulsetset

	 * @param layout

	 *            layout

	 * @param format

	 *            format

	 * @return resultset with transformed records

	 * @throws ISLookUpException

	 *             could happen

	 * @throws IOException

	 *             could happen

	 * @throws TransformerException

	 *             could happen

	 */

	protected W3CEndpointReference prepareForIndexing(final W3CEndpointReference mdStoreRsetEpr, final String format, final String layout)

			throws ISLookUpException, IOException, TransformerException {

		final TransformerFactory factory = TransformerFactory.newInstance();

		final Transformer layoutTransformer = factory.newTransformer(new StreamSource(getLayoutToRecordStylesheet().getInputStream()));

		final DOMResult layoutToXsltXslt = new DOMResult();

		layoutTransformer.setParameter("format", format);

		layoutTransformer.transform(new StreamSource(new StringReader(getLayoutSource(format, layout))), layoutToXsltXslt);

		dumpXslt(factory, layoutToXsltXslt);

		return getXsltRSFactory().createMappedResultSet(mdStoreRsetEpr, new DOMSource(layoutToXsltXslt.getNode()),

				"dynamic layout xslt for " + format + ", " + layout);

	}

	private String getLayoutSource(final String format, final String layout) throws ISLookUpDocumentNotFoundException, ISLookUpException {

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

				"collection('')//RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'MDFormatDSResourceType' and .//NAME='" + format + "']//LAYOUT[@name='" + layout

						+ "']");

	}

	private void dumpXslt(final TransformerFactory factory, final DOMResult layoutToXsltXslt) throws TransformerConfigurationException, TransformerException {

		if (log.isDebugEnabled()) {

			final StringWriter buffer = new StringWriter();

			factory.newTransformer().transform(new DOMSource(layoutToXsltXslt.getNode()), new StreamResult(buffer));

			log.debug(buffer.toString());

		}

	}

	private String encode(final String epr) {

		return new String(Base64.encodeBase64(epr.getBytes()));

	}

	// setters and getters

	public String getIndexId() {

		return indexId;

	}

	public void setIndexId(final String indexId) {

		this.indexId = indexId;

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

	public String getFeedingType() {

		return feedingType;

	}

	public void setFeedingType(final String feedingType) {

		this.feedingType = feedingType;

	}

	public ProcessCountingResultSetFactory getProcessCountingResultSetFactory() {

		return processCountingResultSetFactory;

	}

	@Required

	public void setProcessCountingResultSetFactory(final ProcessCountingResultSetFactory processCountingResultSetFactory) {

		this.processCountingResultSetFactory = processCountingResultSetFactory;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return progressProvider;

	}

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

		return layoutToRecordStylesheet;

	}

	@Required

	public void setLayoutToRecordStylesheet(final org.springframework.core.io.Resource layoutToRecordStylesheet) {

		this.layoutToRecordStylesheet = layoutToRecordStylesheet;

	}

	public String getFormat() {

		return format;

	}

	public void setFormat(final String format) {

		this.format = format;

	}

	public String getLayout() {

		return layout;

	}

	public void setLayout(final String layout) {

		this.layout = layout;

	}

	public XSLTMappedResultSetFactory getXsltRSFactory() {

		return xsltRSFactory;

	}

	@Required

	public void setXsltRSFactory(final XSLTMappedResultSetFactory xsltRSFactory) {

		this.xsltRSFactory = xsltRSFactory;

	}

	/**

	 * @return the defaultIndexId

	 */

	public String getDefaultIndexId() {

		return defaultIndexId;

	}

	/**

	 * @param defaultIndexId

	 *            the defaultIndexId to set

	 */

	@Required

	public void setDefaultIndexId(final String defaultIndexId) {

		this.defaultIndexId = defaultIndexId;

	}

}

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

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

import eu.dnetlib.data.provision.index.rmi.IndexService;

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

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

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

// TODO: Auto-generated Javadoc

/**

 * The Class CreateIndexJobNode.

 */

public class CreateIndexJobNode extends BlackboardJobNode {

	/** The Constant log. */

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

	/** The output prefix. */

	private String outputPrefix = "index_";

	/** The default backend id. */

	private String defaultBackendId;

	/** The format. */

	private String format;

	/** The layout. */

	private String layout;

	/** The interpretation. */

	private String interpretation;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	/*

	 * (non-Javadoc)

	 * 

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

	 * com.googlecode.sarasvati.NodeToken)

	 */

	@Override

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

		final String env_format = token.getFullEnv().getAttribute("format");

		final String env_layout = token.getFullEnv().getAttribute("layout");

		final String env_interp = token.getFullEnv().getAttribute("interpretation");

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

			this.format = env_format;

		}

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

			this.layout = env_layout;

		}

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

			this.interpretation = env_interp;

		}

		log.info("preparing CREATE blackboard job: " + format + "-" + layout + "-" + interpretation);

		job.setAction("CREATE");

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

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

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

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

	}

	/**

	 * Gets the output prefix.

	 * 

	 * @return the output prefix

	 */

	public String getOutputPrefix() {

		return outputPrefix;

	}

	/**

	 * Sets the output prefix.

	 * 

	 * @param outputPrefix

	 *            the output prefix

	 */

	public void setOutputPrefix(final String outputPrefix) {

		this.outputPrefix = outputPrefix;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.workflows.nodes.BlackboardJobNode#generateBlackboardListener(com.googlecode.sarasvati.Engine,

	 * com.googlecode.sarasvati.NodeToken)

	 */

	@Override