D-Net

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

import java.io.StringReader;

import java.util.Set;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.datasources.LocalOpenaireDatasourceManager;

import eu.dnetlib.enabling.resultset.MappedResultSetFactory;

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.Element;

import org.dom4j.io.SAXReader;

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

public class FilterManagedDatasourcesJobNode extends SimpleJobNode {

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

	private String inputEprParam;

	private String outputEprParam;

	@Autowired

	private LocalOpenaireDatasourceManager dsManager;

	@Autowired

	private MappedResultSetFactory mappedResultSetFactory;

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

		final Set<String> managedDatasources = dsManager.listManagedDatasourceIds();

		log.info(String.format("found %s managed datasources", managedDatasources.size()));

		final W3CEndpointReference outputEpr = mappedResultSetFactory.createMappedResultSet(inputEpr, s -> filterManaged(s, managedDatasources));

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

		return Arc.DEFAULT_ARC;

	}

	/**

	 * Extracts the datasource id from the input record and checks its existence in the given set.

	 * @param data

	 * @param filter

	 * @return The

	 * @throws IllegalStateException

	 */

	private String filterManaged(final String data, Set<String> filter) throws IllegalStateException {

		try {

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

			final String dsId = doc.valueOf("/record/metadata/ROWS/ROW[@table = 'dsm_datasources']/FIELD[@name = 'id']/text()");

			if (filter.contains(dsId)) {

				doc.selectSingleNode("/record/metadata/ROWS").detach();

				((Element) doc.selectSingleNode("/record/metadata")).addElement("ROWS");

				return doc.asXML();

			} else return data;

		} catch (DocumentException e) {

			throw new IllegalStateException(e);

		}

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

}

		<WORKFLOW_NAME>Aggregate Metadata (datasource and organization entities) [Ingestion]</WORKFLOW_NAME>

			<FIELD name="Content">datasource and organization entities</FIELD>

					<PARAM name="wfName" managedBy="system" required="true" type="string">Aggregate Metadata (datasource and organization entities) [Ingestion]</PARAM>

		<ARC to="filterManaged"/>

<NODE name="filterManaged" type="FilterManagedDatasources">

    <DESCRIPTION>Filter DB rows according to the managed flag state</DESCRIPTION>

    <PARAMETERS>

        <PARAM managedBy="system" name="inputEprParam" required="true" type="string">cleaned_row_epr</PARAM>

        <PARAM managedBy="system" name="outputEprParam" required="true" type="string">filtered_row_epr</PARAM>

    </PARAMETERS>

    <ARCS>

        <ARC to="storeRowRecords"/>

    </ARCS>

</NODE>

	<bean id="wfNodeFilterManagedDatasources"

	      class="eu.dnetlib.msro.openaireplus.workflows.nodes.FilterManagedDatasourcesJobNode"

	      scope="prototype"/>