D-Net

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

import java.util.Map;

import java.util.Optional;

import java.util.stream.Collectors;

import eu.dnetlib.dhp.model.mdstore.Provenance;

import eu.dnetlib.enabling.datasources.common.Datasource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.collector.worker.model.ApiDescriptor;

import eu.dnetlib.enabling.datasources.common.ApiParam;

import eu.dnetlib.enabling.datasources.common.LocalDatasourceManager;

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

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

public class PrepareEnvCollectHadoopJobNode extends SimpleJobNode {

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

    @Autowired

    private LocalDatasourceManager<?, ?> dsManager;

    private String hdfsBasePath;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        // param 1 : hdfs path

        // param 2 : api descriptor (json)

        // param 3 : nameNode

        final String dsId = token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ID);

        final String apiId = token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);

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

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

        final Optional<ApiDescriptor> opt = dsManager.getApis(dsId)

                .stream()

                .filter(a -> a.getId().equals(apiId))

                .map(a -> {

                    final ApiDescriptor res = new ApiDescriptor();

                    res.setBaseUrl(a.getBaseurl());

                    res.setId(a.getId());

                    res.setProtocol(a.getProtocol());

                    res.getParams().put("metadata_identifier_path", a.getMetadataIdentifierPath());

                    res.getParams().putAll(a.getApiParams()

                            .stream()

                            .map(o -> (ApiParam) o)

                            .collect(Collectors.toMap(ApiParam::getParam, ApiParam::getValue)));

                    return res;

                })

                .findFirst();

        if (opt.isPresent()) {

            final ApiDescriptor api = opt.get();

            final String hdfsPath = String.format("%s/%s/%s/store", hdfsBasePath, mdId, versionId);

            final String seqFilePath = String.format("%s/%s/%s/seqFile", hdfsBasePath, mdId, versionId);

            token.getEnv().setAttribute("apiDescription", new Gson().toJson(api));

            token.getEnv().setAttribute("mdStorePath", hdfsPath);

            token.getEnv().setAttribute("sequenceFilePath", seqFilePath);

            final Provenance provenance = new Provenance();

            provenance.setDatasourceId(dsId);

            final Datasource<?, ?> ds = dsManager.getDs(dsId);

            provenance.setDatasourceName(ds.getOfficialname());

            provenance.setNsPrefix(ds.getNamespaceprefix());

            token.getEnv().setAttribute("dataSourceInfo", new Gson().toJson(provenance));

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

            token.getEnv().setAttribute("identifierPath",api.getParams().get("metadata_identifier_path"));

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

            token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE_BASEURL, api.getBaseUrl());

            token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_PREFIX + "protocol", api.getProtocol());

            final Map<String, String> params = api.getParams();

            if (params != null) {

                for(Map.Entry<String, String> e : params.entrySet()) {

                    token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_PREFIX + e.getKey(), e.getValue());

                }

            }

            return Arc.DEFAULT_ARC;

        } else {

            return "abort";

        }

    }

    public String getHdfsBasePath() {

        return hdfsBasePath;

    }

    public void setHdfsBasePath(String hdfsBasePath) {

        this.hdfsBasePath = hdfsBasePath;

    }

}