D-Net

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

import com.googlecode.sarasvati.NodeToken;

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

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class DeleteObjectStoreJobNode extends BlackboardJobNode {

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

	private String objectstoreId;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		log.info("preparing blackboard job: delete of objectstore " + getObjectstoreId());

		job.setAction("DELETE");

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

	}

	/**

	 * @return the objectstoreId

	 */

	public String getObjectstoreId() {

		return objectstoreId;

	}

	/**

	 * @param objectstoreId

	 *            the objectstoreId to set

	 */

	public void setObjectstoreId(final String objectstoreId) {

		this.objectstoreId = objectstoreId;

	}

}

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

import javax.annotation.Resource;

import com.google.common.base.Splitter;

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.msro.rmi.MSROException;

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

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

public class VerifyDatasourceJobNode extends SimpleJobNode {

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String expectedInterfaceTypologyPrefixes;

	private String expectedCompliancePrefixes;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String dsId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ID);

		final String ifaceId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);

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

		String compliance;

		try {

			compliance = lookupService.getResourceProfileByQuery("/*[.//RESOURCE_IDENTIFIER/@value='" + dsId + "']//INTERFACE[@id = '" + ifaceId

					+ "']/INTERFACE_EXTRA_FIELD[@name='overriding_compliance']/text()");

		} catch (ISLookUpDocumentNotFoundException e) {

			compliance = lookupService.getResourceProfileByQuery("/*[.//RESOURCE_IDENTIFIER/@value='" + dsId + "']//INTERFACE[@id = '" + ifaceId

					+ "']/@compliance/string()");

		}

		final String typology = lookupService.getResourceProfileByQuery("/*[.//RESOURCE_IDENTIFIER/@value='" + dsId + "']//INTERFACE[@id = '" + ifaceId

				+ "']/@typology/string()");

		verifyValue(compliance, expectedCompliancePrefixes);

		verifyValue(typology, expectedInterfaceTypologyPrefixes);

		token.getFullEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE_COMPLIANCE, compliance);

		if (isPending(dsId)) {

			return "validateDs";

		} else {

			return Arc.DEFAULT_ARC;

		}

	}

	private void verifyValue(final String value, final String expected) throws Exception {

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

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

				if (value.toLowerCase().startsWith(s.toLowerCase())) { return; }

			}

			throw new MSROException("Invalid value: " + value + ", Valid term prefixes are: [" + expected + "]");

		}

	}

	private boolean isPending(final String id) throws ISLookUpDocumentNotFoundException, ISLookUpException {

		final String query = "/*[.//RESOURCE_IDENTIFIER/@value='" + id + "']//RESOURCE_KIND/@value/string()";

		final String res = serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(query);

		return res.trim().equals("PendingRepositoryResources");

	}

	public String getExpectedInterfaceTypologyPrefixes() {

		return expectedInterfaceTypologyPrefixes;

	}

	public void setExpectedInterfaceTypologyPrefixes(final String expectedInterfaceTypologyPrefixes) {

		this.expectedInterfaceTypologyPrefixes = expectedInterfaceTypologyPrefixes;

	}

	public String getExpectedCompliancePrefixes() {

		return expectedCompliancePrefixes;

	}

	public void setExpectedCompliancePrefixes(final String expectedCompliancePrefixes) {

		this.expectedCompliancePrefixes = expectedCompliancePrefixes;

	}

}

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

import java.io.StringReader;

import java.io.StringWriter;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

public class DeleteMetaWfJobNode extends SimpleJobNode {

	private String metaWfId;

	@Resource

	private UniqueServiceLocator serviceLocator;

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

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

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

		final String dsId = doc.valueOf("//DATAPROVIDER/@id");

		final String dsName = doc.valueOf("//DATAPROVIDER/text()");

		final String ifaceId = doc.valueOf("//DATAPROVIDER/@interface");

		final String destroyWfId = doc.valueOf("//CONFIGURATION/@destroyWorkflow");

		log.info("Removing a MetaWf of dataprovider: " + dsId);

		token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_ID, dsId);

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

		token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE, ifaceId);

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

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

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

			try {

				registry.deleteProfile(wfId);

				log.info(" - Deleted Workflow: " + wfId);

			} catch (Exception e) {

				log.error(" - (ERR) Error deleting profile " + wfId);

			}

		}

		registry.deleteProfile(metaWfId);

		log.info(" - Deleted MetaWorkflow: " + metaWfId);

		registry.deleteProfile(destroyWfId);

		log.info(" - Deleted destroy workflow: " + destroyWfId);

		verifyDatasource(dsId, ifaceId);

		return Arc.DEFAULT_ARC;

	}

	private void verifyDatasource(final String dsId, final String ifaceId) throws Exception {

		final StringWriter sw = new StringWriter();

		sw.append("for $x in collection('/db/DRIVER/MetaWorkflowDSResources/MetaWorkflowDSResourceType') where");

		sw.append("  $x//DATAPROVIDER/@id = '" + dsId + "' and ");

		sw.append("  $x//DATAPROVIDER/@interface = '" + ifaceId + "' and ");

		sw.append("  $x//RESOURCE_IDENTIFIER/@value != '" + metaWfId + "' ");

		sw.append("return $x//RESOURCE_IDENTIFIER/@value/string()");

		final boolean active = !serviceLocator.getService(ISLookUpService.class).quickSearchProfile(sw.toString()).isEmpty();

		log.info(" - Updating iface, active status: " + active);

		updateIfaceActivationStatus(dsId, ifaceId, active);

	}

	protected void updateIfaceActivationStatus(final String dsId, final String ifaceId, final boolean active) throws Exception {

		serviceLocator.getService(ISRegistryService.class).updateProfileNode(dsId, "//INTERFACE[@id = '" + ifaceId + "']/@active", "'" + active + "'");

	}

	public String getMetaWfId() {

		return metaWfId;

	}

	public void setMetaWfId(final String metaWfId) {

		this.metaWfId = metaWfId;

	}

}

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

import java.io.IOException;

import java.io.StringWriter;

import javax.annotation.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

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

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.StringEscapeUtils;

public class RegisterMetaWfJobNode extends SimpleJobNode {

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String adminEmail;

	private String wfName;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String dsId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ID);

		final String dsName = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_NAME);

		final String ifaceId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);

		final String metaWfId = registerDatasourceWorkflow(dsId, dsName, ifaceId);

		token.getFullEnv().setAttribute("META_WORKFLOW_ID", metaWfId);

		token.getFullEnv().setAttribute("META_WORKFLOW_STATUS", WorkflowStatus.ASSIGNED.toString());

		return Arc.DEFAULT_ARC;

	}

	public String registerDatasourceWorkflow(final String dsId, final String dsName, final String ifaceId) throws ISRegistryException, IOException {

		final StringWriter sw = new StringWriter();

		IOUtils.copy(getClass().getResourceAsStream("/eu/dnetlib/msro/workflows/templates/meta-workflow.xml.st"), sw);

		final StringTemplate st = new StringTemplate(sw.toString());

		st.setAttribute("dsId", dsId);

		st.setAttribute("ifaceId", ifaceId);

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

		st.setAttribute("section", "dataproviders");

		st.setAttribute("wfName", StringEscapeUtils.escapeXml(getWfName()));

		st.setAttribute("wfFamily", StringEscapeUtils.escapeXml(getWfName()));

		st.setAttribute("adminEmail", adminEmail);

		st.setAttribute("status", WorkflowStatus.ASSIGNED.toString());

		return serviceLocator.getService(ISRegistryService.class).registerProfile(st.toString());

	}

	public String getWfName() {

		return wfName;

	}

	public void setWfName(final String wfName) {

		this.wfName = wfName;

	}

	public String getAdminEmail() {

		return adminEmail;

	}

	public void setAdminEmail(final String adminEmail) {

		this.adminEmail = adminEmail;

	}

}

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

import java.util.Map;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

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

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

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

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

public class CreateObjectStoreJobNode extends BlackboardJobNode {

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

	private String interpretation;

	private String outputPrefix = "objectStore_";

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		log.info("preparing blackboard job for the creation of the objectStore ");

		String basePath  = token.getEnv().getAttribute("objectStoreBasePath");

		job.setAction("CREATE");

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

		if (!StringUtils.isEmpty(basePath)) {

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

		}

	}

	public String getInterpretation() {

		return interpretation;

	}

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

	public String getOutputPrefix() {

		return outputPrefix;

	}

	public void setOutputPrefix(final String outputPrefix) {

		this.outputPrefix = outputPrefix;

	}

	@Override

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

		return new BlackboardWorkflowJobListener(engine, token) {

			@Override

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

				env.setAttribute(getOutputPrefix() + "interpretation", interpretation);

				env.setAttribute(getOutputPrefix() + "id", responseParams.get("objectStoreId"));

			}

		};

	}

}

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

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

public class PrepareCreateIndexJobNode extends SimpleJobNode {

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

	private String layout;

	private String format;

	private String interpretation;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		log.info("Preparing env for CreateIndexJobNode");

		token.getEnv().setAttribute("layout", layout);

		token.getEnv().setAttribute("format", format);

		token.getEnv().setAttribute("interpretation", interpretation);

		return null;

	}

	public String getLayout() {

		return layout;

	}

	public void setLayout(final String layout) {

		this.layout = layout;

	}

	public String getFormat() {

		return format;

	}

	public void setFormat(final String format) {

		this.format = format;

	}

	public String getInterpretation() {

		return interpretation;

	}

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

}

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

import java.io.StringReader;

import java.util.List;

import javax.annotation.Resource;

import javax.xml.XMLConstants;

import javax.xml.transform.stream.StreamSource;

import javax.xml.validation.Schema;

import javax.xml.validation.SchemaFactory;

import javax.xml.validation.Validator;

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.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.rmi.MSROException;

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

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

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

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

public class ValidateProfilesJobNode extends SimpleJobNode implements ProgressJobNode {

	@Resource

	private UniqueServiceLocator serviceLocator;

	private int total = 0;

	private int current = 0;

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

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

		final SchemaFactory schemaFactory = SchemaFactory.newInstance(XMLConstants.W3C_XML_SCHEMA_NS_URI);

		int invalidTotals = 0;

		final List<String> list = lookup.listResourceTypes();

		this.total = list.size();

		this.current = 0;

		for (String resourceType : list) {

			int valid = 0;

			int invalid = 0;

			final String schemaSource = lookup.getResourceTypeSchema(resourceType);

			final Schema schema = schemaFactory.newSchema(new StreamSource(new StringReader(schemaSource)));

			final Validator validator = schema.newValidator();

			for (String profile : lookup.quickSearchProfile("/RESOURCE_PROFILE[./HEADER/RESOURCE_TYPE/@value='" + resourceType + "']")) {

				try {

					validator.validate(new StreamSource(new StringReader(profile)));

					valid++;

				} catch (Exception e) {

					invalid++;

				}

			}

			invalidTotals += invalid;

			final String message = String.format("Valid: %s, Invalid: %s, Total: %s", valid, invalid, valid + invalid);

			token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + resourceType, message);

			log.info("Validation of " + resourceType + " profiles: " + message);

		}

		if (invalidTotals > 0) { throw new MSROException("Validation wf has found " + invalidTotals + " invalid profiles"); }

		return Arc.DEFAULT_ARC;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return new ProgressProvider() {

			@Override

			public boolean isInaccurate() {

				return false;

			}

			@Override

			public int getTotalValue() {

				return total;

			}

			@Override

			public int getCurrentValue() {

				return current;

			}

		};

	}

}

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

import java.util.Map;

import java.util.Map.Entry;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

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

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

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

public class BlackboardWorkflowJobListener extends AbstractBlackboardJobListener {

	/**

	 * logger.

	 */

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

	/**

	 * workflow engine.

	 */

	private Engine engine;

	/**

	 * workflow node token.

	 */

	private NodeToken token;

	@Override

	protected void onDone(final BlackboardJob job) {

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

		complete(job, Arc.DEFAULT_ARC);

	}

	public BlackboardWorkflowJobListener(final Engine engine, final NodeToken token) {

		super();

		this.engine = engine;

		this.token = token;

	}

	@Override

	final public void processJob(final BlackboardJob job) {

		token.getEnv().setTransientAttribute(WorkflowsConstants.BLACKBOARD_JOB, job);

		super.processJob(job);

	}

	@Override

	final protected void onFailed(final BlackboardJob job) {

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

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

		token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_ERROR, job.getError());

		complete(job, "failed");

	}

	final protected void complete(final BlackboardJob job, final String arc) {

		final Env env = token.getEnv();

		populateEnv(env, job.getParameters());

		engine.complete(token, arc);

		engine.executeQueuedArcTokens(token.getProcess());

	}

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

		for (Entry<String, String> entry : responseParams.entrySet()) {

			env.setAttribute(WorkflowsConstants.BLACKBOARD_PARAM_PREFIX + entry.getKey(), entry.getValue());

		}

	}

	@Override

	protected void onOngoing(final BlackboardJob job) {

		token.getEnv().setAttribute(WorkflowsConstants.BLACKBOARD_IS_GOING, true);

	}

	public Engine getEngine() {

		return engine;

	}

	public void setEngine(final Engine engine) {

		this.engine = engine;

	}

	public NodeToken getToken() {

		return token;

	}

	public void setToken(final NodeToken token) {

		this.token = token;

	}

}

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

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

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

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

public class DeleteIndexJobNode extends BlackboardJobNode {

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

	private String indexId;

	private String defaultIndexId;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		log.info("deleting index id = " + getIndexId());

		job.setAction("DELETE");

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

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

	}

	public String getIndexId() {

		return indexId;

	}

	public void setIndexId(final String indexId) {

		this.indexId = indexId;

	}

	public String getDefaultIndexId() {

		return defaultIndexId;

	}

	public void setDefaultIndexId(final String defaultIndexId) {

		this.defaultIndexId = defaultIndexId;

	}

}

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

import java.io.IOException;

import javax.annotation.Resource;

import org.apache.commons.io.IOUtils;

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 ExecuteSqlJobNode extends AsyncJobNode {

	private String db;

	private String dbParam;

	private String dbProperty;

	private String sql;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

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

		return dbProperty;

	}

	public void setDbProperty(final String dbProperty) {

		this.dbProperty = dbProperty;

	}

	public String getSql() {

		return sql;

	}

	public void setSql(final String sql) {

		this.sql = sql;

	}

}

package eu.dnetlib.msro.workflows.nodes;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

public class SetEnvParameterJobNode extends SimpleJobNode {

	private String parameterName;

	private String parameterValue;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		token.getEnv().setAttribute(parameterName, parameterValue);

		return Arc.DEFAULT_ARC;

	}

	public String getParameterName() {

		return parameterName;

	}

	public void setParameterName(final String parameterName) {

		this.parameterName = parameterName;

	}

	public String getParameterValue() {

		return parameterValue;

	}

	public void setParameterValue(final String parameterValue) {

		this.parameterValue = parameterValue;

	}

}

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

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

public class OngoingBlackboardWorkflowJobListener extends BlackboardWorkflowJobListener {

	public OngoingBlackboardWorkflowJobListener(Engine engine, NodeToken token) {

		super(engine, token);

	}

	@Override

	protected void onOngoing(BlackboardJob job) {

		super.onOngoing(job);

		complete(job, Arc.DEFAULT_ARC);

	}

}

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

import static java.lang.String.format;

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

public class SetFormatInfoJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	private String format;

	private String formatParam = "format";

	private String layout;

	private String layoutParam = "layout";

	private String interpretation;

	private String interpretationParam = "interpretation";

	/**

	 * {@inheritDoc}

	 * 

	 * @see com.googlecode.sarasvati.mem.MemNode#execute(com.googlecode.sarasvati.Engine, com.googlecode.sarasvati.NodeToken)

	 */

	@Override

	public String execute(final NodeToken token) {

		log.info(format("setting mdFromat: %s, layout: %s, interpretation: %s", getFormat(), getLayout(), getInterpretation()));

		token.getEnv().setAttribute(getFormatParam(), getFormat());

		token.getEnv().setAttribute(getLayoutParam(), getLayout());

		token.getEnv().setAttribute(getInterpretationParam(), getInterpretation());

		return Arc.DEFAULT_ARC;

	}

	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 String getInterpretation() {

		return interpretation;

	}

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

	public String getFormatParam() {

		return formatParam;

	}

	public void setFormatParam(final String formatParam) {

		this.formatParam = formatParam;

	}

	public String getLayoutParam() {

		return layoutParam;

	}

	public void setLayoutParam(final String layoutParam) {

		this.layoutParam = layoutParam;

	}

	public String getInterpretationParam() {

		return interpretationParam;

	}

	public void setInterpretationParam(final String interpretationParam) {

		this.interpretationParam = interpretationParam;

	}

}

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

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Iterator;

import java.util.Map;

import javax.annotation.Resource;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.common.logging.DnetLogger;

import eu.dnetlib.msro.rmi.MSROException;

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

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

public class FindDateRangeForIncrementalHarvestingJobNode extends SimpleJobNode {

	private String fromDateParam;

	private String untilDateParam;

	private long ONE_DAY = 1000 * 60 * 60 * 24;	

	@Resource(name = "msroWorkflowLogger")

	private DnetLogger dnetLogger;