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

     * MDStore identifier

     */

    private String mdId;

     * REFRESH | INCREMENTAL

     */

    private String collectionMode;

     * XML | JSON, default = XML

     */

    private String metadataEncoding = "XML";

    private String oozieWfPath;

    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 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());

                            .stream()

                            .map(o -> (ApiParam) o)

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

                    return res;

                })

                .findFirst();

        if (opt.isPresent()) {

            token.getEnv().setAttribute("collectionMode", getCollectionMode());

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

            provenance.setDatasourceId(dsId);

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

            provenance.setDatasourceName(ds.getOfficialname());

            provenance.setNsPrefix(ds.getNamespaceprefix());

            log.info("datasource provenance: " + dsProvenance);

            token.getEnv().setAttribute("dataSourceInfo", dsProvenance);

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

        this.mdId = mdId;

        return collectionMode;

    public void setCollectionMode(String collectionMode) {

        this.collectionMode = collectionMode;

    }

    public String getOozieWfPath() {

        return oozieWfPath;

    }

    public void setOozieWfPath(String oozieWfPath) {

        this.oozieWfPath = oozieWfPath;

    }

    public String getMetadataEncoding() {

        return metadataEncoding;

    }

    public void setMetadataEncoding(String metadataEncoding) {

        this.metadataEncoding = metadataEncoding;

    }