D-Net

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

import eu.dnetlib.dhp.utils.DHPUtils;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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;

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

public class PrepareEnvTransformHadoopJobNode extends SimpleJobNode {

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

    private String hdfsBasePath;

    private String ruleId;

    @Autowired

    private UniqueServiceLocator serviceLocator;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        final String reading_mdId = token.getEnv().getAttribute("reading_mdId");

        final String reading_versionId = token.getEnv().getAttribute("reading_versionId");

        final String writing_mdId = token.getEnv().getAttribute("mdId");

        final String writing_versionId = token.getEnv().getAttribute("versionId");

        final String hdfsNativePath = String.format("%s/%s/%s/store", hdfsBasePath, reading_mdId, reading_versionId);

        final String hdfsTransformedPath = String.format("%s/%s/%s/store", hdfsBasePath, writing_mdId, writing_versionId);

        final String trRule = serviceLocator.getService(ISLookUpService.class).getResourceProfile(getRuleId());

        token.getEnv().setAttribute("timestamp", "" + System.currentTimeMillis());

        token.getEnv().setAttribute("workflowId", token.getProcess().getEnv().getAttribute("system:processId"));

        token.getEnv().setAttribute("mdstoreInputPath", hdfsNativePath);

        token.getEnv().setAttribute("mdstoreOutputPath", hdfsTransformedPath);

        token.getEnv().setAttribute("transformationRule", DHPUtils.compressString(trRule));

        return Arc.DEFAULT_ARC;

    }

    public String getHdfsBasePath() {

        return hdfsBasePath;

    }

    public void setHdfsBasePath(final String hdfsBasePath) {

        this.hdfsBasePath = hdfsBasePath;

    }

    public String getRuleId() {

        return ruleId;

    }

    public void setRuleId(String ruleId) {

        this.ruleId = ruleId;

    }

    public UniqueServiceLocator getServiceLocator() {

        return serviceLocator;

    }

    public void setServiceLocator(UniqueServiceLocator serviceLocator) {

        this.serviceLocator = serviceLocator;

    }

}

    <bean id="wfNodePrepareEnvTransformHadoopJobNode"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.dhp.PrepareEnvTransformHadoopJobNode"

          scope="prototype" />