D-Net

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

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.msro.rmi.MSROException;

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

import org.apache.commons.lang3.StringUtils;

import java.util.List;

import java.util.Map;

public class ExtractOutputPathJobNode extends SimpleJobNode {

    private String hdfsOutputPathParam;

    @Override

    protected String execute(NodeToken token) throws Exception {

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

        if (StringUtils.isBlank(sets))

            throw  new MSROException("¯\\\\_(ツ)_/¯ cannot find sets on env");

        final List<Map<String, String>> setsMap = new Gson().fromJson(sets, List.class);

        if (setsMap== null || setsMap.size() !=1)  {

            throw  new MSROException("¯\\\\_(ツ)_/¯ Sets map from json is wrong!");

        }

        final String path = setsMap.get(0).get("path");

        if (StringUtils.isEmpty(path)){

            throw new MSROException("Path is empty");

        }

        token.getEnv().setAttribute(getHdfsOutputPathParam(), path);

        return Arc.DEFAULT_ARC;

    }

    public String getHdfsOutputPathParam() {

        return hdfsOutputPathParam;

    }

    public void setHdfsOutputPathParam(String hdfsOutputPathParam) {

        this.hdfsOutputPathParam = hdfsOutputPathParam;

    }

}