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 ruleId;

    @Autowired

    private UniqueServiceLocator serviceLocator;

    private String mdstoreInput;

    private String mdstoreOutput;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        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("mdstoreInput", getMdstoreInput());

        token.getEnv().setAttribute("mdstoreOutput", getMdstoreOutput());

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

        return Arc.DEFAULT_ARC;

    }

    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;

    }

    public String getMdstoreInput() {

        return mdstoreInput;

    }

    public void setMdstoreInput(String mdstoreInput) {

        this.mdstoreInput = mdstoreInput;

    }

    public String getMdstoreOutput() {

        return mdstoreOutput;

    }

    public void setMdstoreOutput(String mdstoreOutput) {

        this.mdstoreOutput = mdstoreOutput;

    }

}