D-Net

		final Long dateFromFilter = token.getFullEnv().getAttribute("DateFromFilter", Long.class);

		if (dateFromFilter!= null) {

			log.info("From Filter activated, from:"+dateFromFilter);

		}

		final W3CEndpointReference epr = mdStoreService.deliverMDRecords(getMdId(), dateFromFilter!=null?""+dateFromFilter:"", "", "");

		final String overrideStoringType = token.getFullEnv().getAttribute("transformationType");

		job.getParameters().put("storingType", overrideStoringType!= null? overrideStoringType: getStoringType());

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.common.logging.DnetLogger;

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

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

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

import java.util.Iterator;

import java.util.List;

import java.util.Map;

public class IncrementalTransformationJobNode extends SimpleJobNode {

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

    private String transformationType;

    @Autowired

    private DnetLogger dnetLogger;

    @Autowired

    private UniqueServiceLocator locator;

    private static final DateUtils dateUtils = new DateUtils();

    @Override

    protected String execute(NodeToken nodeToken) throws Exception {

        if ("incremental".equalsIgnoreCase(transformationType)) {

            final String currentWfProfileId = findCurrentWfProfileId(nodeToken);

            final Long lastSuccessEndDate = findLastSuccessEndDate(currentWfProfileId);

            if (lastSuccessEndDate<0) {

                log.info("Last success date <0 the transformation will be forced to REFRESH");

                nodeToken.getFullEnv().setAttribute("transformationType", "REFRESH");

                return Arc.DEFAULT_ARC;

            }

            log.info("Last succes date "+ lastSuccessEndDate);

            final long lastUpdateDate = getLastTransformationRuleUpdate(getTransformationId(currentWfProfileId));

            log.info("last update date of the trId is "+lastUpdateDate);

            if (lastUpdateDate> lastSuccessEndDate) {

                log.info("Transformation Rule has benn updated, hence the transformation process will be forced to REFRESH");

                nodeToken.getFullEnv().setAttribute("transformationType", "REFRESH");

                return Arc.DEFAULT_ARC;

            }

            nodeToken.getFullEnv().setAttribute("transformationType", "INCREMENTAL");

            nodeToken.getFullEnv().setAttribute("DateFromFilter", lastSuccessEndDate);

            log.info("Transformation type setted to INCREMENTAL");

            return Arc.DEFAULT_ARC;

        }

        nodeToken.getFullEnv().setAttribute("transformationType", "REFRESH");

        log.info("Transformation type setted to REFRESH");

        return Arc.DEFAULT_ARC;

    }

    private Long findLastSuccessEndDate(String profId) {

        long res = -1;

        final Iterator<Map<String, String>> iter = dnetLogger.find(WorkflowsConstants.SYSTEM_WF_PROFILE_ID, profId);

        while (iter.hasNext()) {

            final Map<String, String> map = iter.next();

            log.debug("Iterating on the logs");

            if ("true".equalsIgnoreCase(map.get(WorkflowsConstants.SYSTEM_COMPLETED_SUCCESSFULLY))) {

                final long curr = NumberUtils.toLong(map.get(WorkflowsConstants.SYSTEM_END_DATE), -1);

                if (curr > res) {

                    res = curr;

                }

            }

        }

        return res;

    }

    private String findCurrentWfProfileId(NodeToken token) throws MSROException {

        log.debug("Start to find the current profile Id");

        final String p1 = token.getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

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

            log.debug("The profile Id found is "+p1);

            return p1;

        }

        final String p2 = token.getFullEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

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

            log.debug("The profile Id found is "+p2);

            return p2;

        }

        final String p3 = token.getProcess().getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

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

            log.debug("The profile Id found is "+p3);

            return p3;

        }

        throw new MSROException("Missing property in env: " + WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

    }

    private String getTransformationId(final String workflowId) throws ISLookUpException {

        final String query="for $x in collection('/db/DRIVER/WorkflowDSResources/WorkflowDSResourceType') where $x//RESOURCE_IDENTIFIER/@value='%s' " +

                "return $x//PARAM[./@category='TRANSFORMATION_RULE_ID']/text()";

        final ISLookUpService lookUpService = locator.getService(ISLookUpService.class);

        final String queryInstance = String.format(query, workflowId);

        log.debug("Query to find the Transformation Rule");

        List<String> transformationId = lookUpService.quickSearchProfile(queryInstance);

        if(transformationId== null || transformationId.isEmpty())

            throw new RuntimeException("Error unable to find the Transformation rule ID on workflow profile "+workflowId);

        return transformationId.get(0);

    }

    private Long getLastTransformationRuleUpdate(final String trId) throws ISLookUpException {

        final String query = "for $x in collection('/db/DRIVER/TransformationRuleDSResources/TransformationRuleDSResourceType') where $x//RESOURCE_IDENTIFIER/@value='%s' return $x//DATE_OF_CREATION/@value/string()";

        log.debug("retrieve creation date from transformation ID "+trId);

        final ISLookUpService lookUpService = locator.getService(ISLookUpService.class);

        final String queryInstance = String.format(query, trId);

        log.debug("Query to find the Transformation Rule");

        List<String> currentDate = lookUpService.quickSearchProfile(queryInstance);

        if(currentDate== null || currentDate.isEmpty())

            throw new RuntimeException("Error unable to find the creation date of the  Transformation rule "+trId);

        return dateUtils.parse(currentDate.get(0)).getTime();

    }

    public void setTransformationType(String transformationType) {

        this.transformationType = transformationType;

    }

    public String getTransformationType() {

        return transformationType;

    }

}

		<ARC to="selectIncrementalTransformation"/>

<NODE name="selectIncrementalTransformation" type="IncrementalTransformation">

	<DESCRIPTION>Fetch records from MDStore</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="transformationType" managedBy="user" function="validValues(['REFRESH','INCREMENTAL'])"></PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="fetchOriginals"/>

	</ARCS>

</NODE>

	<bean id ="wfNodeIncrementalTransformation"

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

		  scope="prototype"

	/>

	<version>6.3.0-solr7-SNAPSHOT</version>