D-Net

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

import java.net.URI;

import java.util.HashMap;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.web.client.RestTemplate;

import org.springframework.web.util.UriComponentsBuilder;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.mdstore.manager.common.model.MDStoreVersion;

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

public class EndReadingMDStoreJobNode extends SimpleJobNode {

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

    /* Spring managed params */

    private String mdStoreManagerUrl;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        final String url = getMdStoreManagerUrl() + "/version/{versionId}/endReading";

        final Map<String, Object> params = new HashMap<>();

        params.put("versionId", token.getEnv().getAttribute("reading_versionId"));

        final URI uri = UriComponentsBuilder.fromUriString(url)

                .buildAndExpand(params)

                .toUri();

        final RestTemplate restTemplate = new RestTemplate();

        restTemplate.getForObject(uri, MDStoreVersion.class);

        log.info("mdstore version read complete ");

        return Arc.DEFAULT_ARC;

    }

    public String getMdStoreManagerUrl() {

        return mdStoreManagerUrl;

    }

    @Required

    public void setMdStoreManagerUrl(final String mdStoreManagerUrl) {

        this.mdStoreManagerUrl = mdStoreManagerUrl;

    }

}

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

import java.net.URI;

import java.util.HashMap;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.web.client.RestTemplate;

import org.springframework.web.util.UriComponentsBuilder;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.mdstore.manager.common.model.MDStoreVersion;

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

public class PrepareMDStoreVersionJobNode extends SimpleJobNode {

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

    private String mdId;

    /* Spring managed params */

    private String mdStoreManagerUrl;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        final String url = getMdStoreManagerUrl() + "/mdstore/{mdId}/newVersion";

        final Map<String, Object> params = new HashMap<>();

        params.put("mdId", getMdId());

        final URI uri = UriComponentsBuilder.fromUriString(url)

                .buildAndExpand(params)

                .toUri();

        final RestTemplate restTemplate = new RestTemplate();

        final MDStoreVersion version = restTemplate.getForObject(uri, MDStoreVersion.class);

        log.info("mdstore version prepared " + version.getId());

        token.getEnv().setAttribute("mdId", mdId);

        token.getEnv().setAttribute("versionId", version.getId());

        return Arc.DEFAULT_ARC;

    }

    public String getMdStoreManagerUrl() {

        return mdStoreManagerUrl;

    }

    @Required

    public void setMdStoreManagerUrl(final String mdStoreManagerUrl) {

        this.mdStoreManagerUrl = mdStoreManagerUrl;

    }

    public String getMdId() {

        return mdId;

    }

    public void setMdId(final String mdId) {

        this.mdId = mdId;

    }

}

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

import eu.dnetlib.message.Message;

import eu.dnetlib.message.MessageManager;

import eu.dnetlib.message.MessageType;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.concurrent.LinkedBlockingQueue;

public class DnetMessageManager {

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

    private MessageManager manager;

    private LinkedBlockingQueue<Message> messages = new LinkedBlockingQueue<>();

    private final  Map<String, Message> onGonginMessages = new HashMap<>();

    private final Map<String, List<Message>> reportMessages = new HashMap<>();

    private String messageQueueServer;

    private String username;

    private String password;

    @Value("${dnet.openaire.messageManager.ongoingMessageQueueName}")

    private String onGoingMessageQueue;

    @Value("${dnet.openaire.messageManager.reportMessageQueueName}")

    private String reportMessageQueue;

    public DnetMessageManager() {

    }

    private String createReportId(final String wfId, final String jobName) {

        return String.format("%s::%s", wfId, jobName);

    }

    public void startListeningMessage() throws Exception {

        if (manager == null && StringUtils.isNotBlank(messageQueueServer) && StringUtils.isNotBlank(reportMessageQueue)) {

            manager = new MessageManager(messageQueueServer, username, password, messages);

            manager.startConsumingMessage(onGoingMessageQueue, true, false);

            manager.startConsumingMessage(reportMessageQueue, true, false);

            Runnable r = () -> {

                while (true) {

                    try {

                        Message currentMessage = messages.take();

                        if (currentMessage.getType() == MessageType.ONGOING) {

                            synchronized (onGonginMessages) {

                                onGonginMessages.put(currentMessage.getWorkflowId(), currentMessage);

                            }

                        } else {

                            synchronized (reportMessages) {

                                if (!reportMessages.containsKey(currentMessage.getWorkflowId()))

                                {

                                    reportMessages.put(currentMessage.getWorkflowId(), new ArrayList<>());

                                }

                                reportMessages.get(currentMessage.getWorkflowId()).add(currentMessage);

                            }

                        }

                    } catch (InterruptedException e) {

                        log.error("An error occured on retrieving messages from the blocking queue",e);

                        throw new RuntimeException("An error occured on retrieving messages from the blocking queue",e);

                    }

                }

            };

            new Thread(r).start();

        }

    }

    public List<Message> getReport(final String workflowId) {

        return getMessages(reportMessages, workflowId);

    }

    private List<Message> getMessages(final Map<String, List<Message>> messageMap, String reportId) {

        if (messageMap.containsKey(reportId)) {

            List<Message> m = messageMap.get(reportId);

            messageMap.remove(reportId);

            return m;

        }

        return null;

    }

    private Message getMessage(final Map<String, Message> messageMap, String reportId) {

        if (messageMap.containsKey(reportId)) {

            Message m = messageMap.get(reportId);

            messageMap.remove(reportId);

            return m;

        }

        return null;

    }

    public Message getOnGoingMessages(final String workflowId) {

        return getMessage(onGonginMessages, workflowId);

    }

    public String getMessageQueueServer() {

        return messageQueueServer;

    }

    @Required

    public void setMessageQueueServer(String messageQueueServer) {

        this.messageQueueServer = messageQueueServer;

    }

    public String getUsername() {

        return username;

    }

    @Required

    public void setUsername(String username) {

        this.username = username;

    }

    public String getPassword() {

        return password;

    }

    @Required

    public void setPassword(String password) {

        this.password = password;

    }

}

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

import java.net.URI;

import java.util.HashMap;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.web.client.RestTemplate;

import org.springframework.web.util.UriComponentsBuilder;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.mdstore.manager.common.model.MDStoreVersion;

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

public class StartReadingMDStoreJobNode extends SimpleJobNode {

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

    private String mdId;

    /* Spring managed params */

    private String mdStoreManagerUrl;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        final String url = getMdStoreManagerUrl() + "/mdstore/{mdId}/startReading";

        final Map<String, Object> params = new HashMap<>();

        params.put("mdId", getMdId());

        final URI uri = UriComponentsBuilder.fromUriString(url)

                .buildAndExpand(params)

                .toUri();

        final RestTemplate restTemplate = new RestTemplate();

        final MDStoreVersion version = restTemplate.getForObject(uri, MDStoreVersion.class);

        log.info("mdstore version ready to read " + version.getId());

        token.getEnv().setAttribute("reading_mdId", mdId);

        token.getEnv().setAttribute("reading_versionId", version.getId());

        return Arc.DEFAULT_ARC;

    }

    public String getMdStoreManagerUrl() {

        return mdStoreManagerUrl;

    }

    @Required

    public void setMdStoreManagerUrl(final String mdStoreManagerUrl) {

        this.mdStoreManagerUrl = mdStoreManagerUrl;

    }

    public String getMdId() {

        return mdId;

    }

    public void setMdId(final String mdId) {

        this.mdId = mdId;

    }

}

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;

    }

}

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

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.message.Message;

import eu.dnetlib.msro.openaireplus.workflows.nodes.dhp.message.DnetMessageManager;

import eu.dnetlib.msro.workflows.hadoop.SubmitHadoopJobNode;

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

import eu.dnetlib.msro.workflows.nodes.blackboard.BlackboardWorkflowJobListener;

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import java.util.List;

import java.util.Map;

public class SubmitDnetHadoopJobNode extends SubmitHadoopJobNode implements ProgressProvider, ProgressJobNode {

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

    @Autowired

    DnetMessageManager dnetMessageManager;

    private boolean ongoing = true;

    private int currentValue;

    private String wfId;

    @Override

    protected void prepareJob(BlackboardJob job, NodeToken token) throws Exception {

        this.wfId = token.getProcess().getEnv().getAttribute("system:processId");

        Runnable r = () -> {

            while (ongoing) {

                Message mess = dnetMessageManager.getOnGoingMessages(wfId);

                if (mess != null && mess.getBody() != null && mess.getBody().containsKey("ongoing")) {

                    try {

                        this.currentValue = Integer.parseInt(mess.getBody().get("ongoing"));

                        Thread.sleep(1000);

                    } catch (Throwable e) {

                        log.error("Error ono receiving messages ", e);

                    }

                }

            }

        };

        new Thread(r).start();

        super.prepareJob(job, token);

    }

    @Override

    protected BlackboardWorkflowJobListener generateBlackboardListener(Engine engine, NodeToken token) {

        return new BlackboardWorkflowJobListener(engine, token) {

            @Override

            protected void onFailed(final BlackboardJob job) {

                ongoing = false;

                log.warn("Blackboard workflow node FAILED: " + job.getError());

                token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_HAS_FAILED, true);

                token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_ERROR, job.getError());

                complete(job, "abort");

            }

            @Override

            protected void populateEnv(Env env, Map<String, String> responseParams) {

                ongoing = false;

                List<Message> reports = dnetMessageManager.getReport(wfId);

                if (reports == null) {

                    int numberOftries = 0;

                    try {

                        while (reports == null && numberOftries < 3) {

                            reports = dnetMessageManager.getReport(wfId);

                            Thread.sleep(3000 * numberOftries++);

                        }

                    } catch (InterruptedException e) {

                        log.error("Error on waiting report", e);

                    }

                }

                if (reports == null) {

                    throw new RuntimeException("Unable to get report for WorklowId " + wfId);

                }

                reports.forEach(it -> it.getBody().forEach(env::setAttribute));

            }

        };

    }

    @Override

    public int getTotalValue() {

        return 0;

    }

    @Override

    public int getCurrentValue() {

        return currentValue;

    }

    @Override

    public boolean isInaccurate() {

        return false;

    }

    @Override

    public ProgressProvider getProgressProvider() {

        return this;

    }

}

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

import java.net.URI;

import java.util.HashMap;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.web.client.RestTemplate;

import org.springframework.web.util.UriComponentsBuilder;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

public class AbortMDStoreVersionJobNode extends SimpleJobNode {

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

    /* Spring managed params */

    private String mdStoreManagerUrl;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        final String url = getMdStoreManagerUrl() + "/version/{versionId}?force=true";

        final Map<String, Object> params = new HashMap<>();

        params.put("versionId", token.getEnv().getAttribute("versionId"));

        final URI uri = UriComponentsBuilder.fromUriString(url)

                .buildAndExpand(params)

                .toUri();

        final RestTemplate restTemplate = new RestTemplate();

        restTemplate.delete(uri);

        log.info("mdstore version deleted ");

        return Arc.DEFAULT_ARC;

    }

    public String getMdStoreManagerUrl() {

        return mdStoreManagerUrl;

    }

    @Required

    public void setMdStoreManagerUrl(final String mdStoreManagerUrl) {

        this.mdStoreManagerUrl = mdStoreManagerUrl;

    }

}

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

import java.net.URI;

import java.util.HashMap;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.web.client.RestTemplate;

import org.springframework.web.util.UriComponentsBuilder;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

public class DeleteMDStoreHadoopJobNode extends SimpleJobNode {

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

    private String mdId;

    /* Spring managed params */

    private String mdStoreManagerUrl;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        final String url = getMdStoreManagerUrl() + "/mdstore/{mdId}";

        final Map<String, Object> params = new HashMap<>();

        params.put("mdId", getMdId());

        final URI uri = UriComponentsBuilder.fromUriString(url)

                .buildAndExpand(params)

                .toUri();

        final RestTemplate restTemplate = new RestTemplate();

        restTemplate.delete(uri);

        log.info("mdstor deleted: " + mdId);

        return Arc.DEFAULT_ARC;

    }

    public String getMdStoreManagerUrl() {

        return mdStoreManagerUrl;

    }

    @Required

    public void setMdStoreManagerUrl(final String mdStoreManagerUrl) {

        this.mdStoreManagerUrl = mdStoreManagerUrl;

    }

    public String getMdId() {

        return mdId;

    }

    public void setMdId(final String mdId) {

        this.mdId = mdId;

    }

}

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

import java.net.URI;

import java.util.HashMap;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.web.client.RestTemplate;

import org.springframework.web.util.UriComponentsBuilder;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

public class CommitMDStoreVersionJobNode extends SimpleJobNode {

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

    /* Spring managed params */

    private String mdStoreManagerUrl;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        final String url = getMdStoreManagerUrl() + "/version/{versionId}/commit/{size}";

        final Map<String, Object> params = new HashMap<>();

        params.put("versionId", token.getEnv().getAttribute("versionId"));

        params.put("size", token.getEnv().getAttribute("mdStoreSize"));

        final URI uri = UriComponentsBuilder.fromUriString(url)

                .buildAndExpand(params)

                .toUri();

        final RestTemplate restTemplate = new RestTemplate();

        restTemplate.getForObject(uri, Void.class);

        log.info("mdstore version committed ");

        return Arc.DEFAULT_ARC;

    }

    public String getMdStoreManagerUrl() {

        return mdStoreManagerUrl;

    }

    @Required

    public void setMdStoreManagerUrl(final String mdStoreManagerUrl) {

        this.mdStoreManagerUrl = mdStoreManagerUrl;

    }

}

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

import java.net.URI;

import java.util.HashMap;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.web.client.RestTemplate;

import org.springframework.web.util.UriComponentsBuilder;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.mdstore.manager.common.model.MDStoreWithInfo;

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

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

public class CreateMDStoreHadoopJobNode extends SimpleJobNode {

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

    /* Workflow params */

    private String format;

    private String layout;

    private String interpretation;

    private String outputPrefix = "mdstore";

    /* Spring managed params */

    private String mdStoreManagerUrl;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        final String url = getMdStoreManagerUrl() + "/new/{format}/{layout}/{interpretation}";

        final Map<String, Object> params = new HashMap<>();

        params.put("format", getFormat());

        params.put("layout", getLayout());

        params.put("interpretation", getInterpretation());

        final URI uri = UriComponentsBuilder.fromUriString(url)

                .queryParam("dsName", token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_NAME))

                .queryParam("dsId", token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ID))

                .queryParam("apiId", token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE))

                .buildAndExpand(params)

                .toUri();

        final RestTemplate restTemplate = new RestTemplate();

        final MDStoreWithInfo result = restTemplate.getForObject(uri, MDStoreWithInfo.class);

        log.info("mdstore created " + result.toString());

        token.getEnv().setAttribute(getOutputPrefix() + "format", format);

        token.getEnv().setAttribute(getOutputPrefix() + "layout", layout);

        token.getEnv().setAttribute(getOutputPrefix() + "interpretation", interpretation);

        token.getEnv().setAttribute(getOutputPrefix() + "id", result.getId());

        return Arc.DEFAULT_ARC;

    }

    public String getFormat() {

        return format;

    }

    public void setFormat(final String format) {

        this.format = format;

    }

    public String getLayout() {

        return layout;

    }

    public void setLayout(final String layout) {

        this.layout = layout;

    }

    public String getInterpretation() {

        return interpretation;

    }

    public void setInterpretation(final String interpretation) {

        this.interpretation = interpretation;

    }

    public String getOutputPrefix() {

        return outputPrefix;

    }

    public void setOutputPrefix(final String outputPrefix) {

        this.outputPrefix = outputPrefix;

    }

    public String getMdStoreManagerUrl() {

        return mdStoreManagerUrl;

    }

    @Required

    public void setMdStoreManagerUrl(final String mdStoreManagerUrl) {

        this.mdStoreManagerUrl = mdStoreManagerUrl;

    }

}

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

import java.net.URI;

import java.util.Date;

import java.util.HashMap;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.web.client.RestTemplate;

import org.springframework.web.util.UriComponentsBuilder;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.mdstore.manager.common.model.MDStoreWithInfo;

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

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

public class MDStoreToApiExtraFieldHadoopJobNode extends SimpleJobNode {

    private String mdId;

    private String datasourceId;

    private String datasourceInterface;

    private String extraFieldForTotal;

    private String extraFieldForDate;

    private String extraFieldForMdId;

    /* Spring managed params */

    private String mdStoreManagerUrl;

    @Autowired

    private LocalDatasourceManager<?, ?> dsManager;

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

    @Override

    protected String execute(final NodeToken token) throws Exception {

        final String url = getMdStoreManagerUrl() + "/mdstore/{mdId}";

        final Map<String, Object> params = new HashMap<>();

        params.put("mdId", getMdId());

        final URI uri = UriComponentsBuilder.fromUriString(url)

                .buildAndExpand(params)

                .toUri();

        final RestTemplate restTemplate = new RestTemplate();

        final MDStoreWithInfo mdstore = restTemplate.getForObject(uri, MDStoreWithInfo.class);

        final int size = Math.toIntExact(mdstore.getSize());

        final Date date = mdstore.getLastUpdate();

        if (extraFieldForTotal.equals("last_collection_total")) {

            dsManager.setLastCollectionInfo(datasourceId, datasourceInterface, mdId, size, date);

        } else if (extraFieldForTotal.equals("last_aggregation_total")) {

            dsManager.setLastAggregationInfo(datasourceId, datasourceInterface, mdId, size, date);

        } else {

            log.warn("Invalid field for total: " + extraFieldForTotal);

        }

        return Arc.DEFAULT_ARC;

    }

    public String getMdId() {

        return mdId;

    }

    public void setMdId(final String mdId) {

        this.mdId = mdId;

    }

    public String getDatasourceId() {

        return datasourceId;

    }

    public void setDatasourceId(final String datasourceId) {

        this.datasourceId = datasourceId;

    }

    public String getDatasourceInterface() {

        return datasourceInterface;

    }

    public void setDatasourceInterface(final String datasourceInterface) {

        this.datasourceInterface = datasourceInterface;

    }

    public String getExtraFieldForTotal() {

        return extraFieldForTotal;

    }

    public void setExtraFieldForTotal(final String extraFieldForTotal) {

        this.extraFieldForTotal = extraFieldForTotal;

    }

    public String getExtraFieldForDate() {

        return extraFieldForDate;

    }

    public void setExtraFieldForDate(final String extraFieldForDate) {

        this.extraFieldForDate = extraFieldForDate;

    }

    public String getExtraFieldForMdId() {

        return extraFieldForMdId;

    }

    public void setExtraFieldForMdId(final String extraFieldForMdId) {

        this.extraFieldForMdId = extraFieldForMdId;

    }

    public String getMdStoreManagerUrl() {

        return mdStoreManagerUrl;

    }

    public void setMdStoreManagerUrl(final String mdStoreManagerUrl) {

        this.mdStoreManagerUrl = mdStoreManagerUrl;

    }

}

<RESOURCE_PROFILE>

    <HEADER>

        <RESOURCE_IDENTIFIER value="513da605-e9b4-4150-9116-f82a34a8585f_V29ya2Zsb3dEU1Jlc291cmNlcy9Xb3JrZmxvd0RTUmVzb3VyY2VUeXBl"/>

        <RESOURCE_TYPE value="WorkflowDSResourceType"/>

        <RESOURCE_KIND value="WorkflowDSResources"/>

        <RESOURCE_URI value="value3"/>

        <DATE_OF_CREATION value="2019-04-03T16:44:41+02:00"/>

    </HEADER>

    <BODY>

        <WORKFLOW_NAME>Aggregate Metadata (publications) from PubsRepository [Hadoop]</WORKFLOW_NAME>

        <WORKFLOW_INFO>

            <FIELD name="Action">Aggregate Metadata</FIELD>

            <FIELD name="Consequence IS">Support</FIELD>

            <FIELD name="Datasource class">PubsRepository</FIELD>

            <FIELD name="Content">publications</FIELD>

        </WORKFLOW_INFO>

        <WORKFLOW_TYPE>REPO_HI</WORKFLOW_TYPE>

        <WORKFLOW_PRIORITY>20</WORKFLOW_PRIORITY>

        <CONFIGURATION start="manual">

            <NODE isStart="true" name="VerifyDatasource" type="VerifyDatasource">

                <DESCRIPTION>Verify if DS is pending</DESCRIPTION>

                <PARAMETERS>

                    <PARAM managedBy="system" name="expectedInterfaceTypologyPrefixes" required="false" type="string"/>

                    <PARAM managedBy="system" name="expectedCompliancePrefixes" required="false" type="string"/>

                </PARAMETERS>

                <ARCS>

                    <ARC to="createMetaWf"/>

                    <ARC name="validateDs" to="validateDs"/>

                </ARCS>

            </NODE>

            <NODE name="validateDs" type="ValidateDatasource">

                <DESCRIPTION>Validate DS</DESCRIPTION>

                <PARAMETERS/>

                <ARCS>

                    <ARC to="createMetaWf"/>

                </ARCS>

            </NODE>

            <NODE name="createMetaWf" type="RegisterMetaWf">

                <DESCRIPTION>Create MetaWorkflow</DESCRIPTION>

                <PARAMETERS>

                    <PARAM managedBy="system" name="wfName" required="true" type="string">Aggregate Metadata (publications) from PubsRepository [Hadoop]

                    </PARAM>

                </PARAMETERS>

                <ARCS>

                    <ARC to="createDc"/>

                    <ARC to="createOaf"/>

                </ARCS>

            </NODE>

            <NODE name="createDc" type="CreateMDStoreHadoop">

                <DESCRIPTION>Create oai_dc store</DESCRIPTION>

                <PARAMETERS>

                    <PARAM managedBy="system" name="format" required="true" type="string">oai_dc</PARAM>

                    <PARAM managedBy="system" name="interpretation" required="true" type="string">native</PARAM>

                    <PARAM managedBy="system" name="layout" required="true" type="string">store</PARAM>

                    <PARAM managedBy="system" name="outputPrefix" required="true" type="string">harv_</PARAM>

                </PARAMETERS>

                <ARCS>

                    <ARC to="updateMetaWf"/>

                </ARCS>

            </NODE>

            <NODE name="createOaf" type="CreateMDStoreHadoop">

                <DESCRIPTION>Create OAF store</DESCRIPTION>

                <PARAMETERS>

                    <PARAM managedBy="system" name="format" required="true" type="string">OAF</PARAM>

                    <PARAM managedBy="system" name="interpretation" required="true" type="string">cleaned</PARAM>

                    <PARAM managedBy="system" name="layout" required="true" type="string">store</PARAM>

                    <PARAM managedBy="system" name="outputPrefix" required="true" type="string">tran_</PARAM>

                </PARAMETERS>

                <ARCS>

                    <ARC to="updateMetaWf"/>

                </ARCS>

            </NODE>

            <NODE isJoin="true" name="updateMetaWf" type="UpdateMetaWf">

                <DESCRIPTION>Create MetaWorkflow</DESCRIPTION>

                <PARAMETERS>

                    <PARAM managedBy="system" name="beanName" required="true" type="string">metaWfPubsRepositoryHadoop</PARAM>

                </PARAMETERS>

                <ARCS>

                    <ARC to="updateMetaWfStatus"/>

                </ARCS>

            </NODE>

            <NODE name="updateMetaWfStatus" type="UpdateOpenaireMetaWfStatus">

                <DESCRIPTION>Update MetaWorkflow Status</DESCRIPTION>

                <PARAMETERS/>

                <ARCS>

                    <ARC to="success"/>

                </ARCS>

            </NODE>

        </CONFIGURATION>

        <STATUS/>

    </BODY>

</RESOURCE_PROFILE>

    <!-- Aggregate Metadata (publications) from PubsRepository [Hadoop] -->

    <bean id="metaWfPubsRepositoryHadoop"

          class="eu.dnetlib.msro.workflows.metawf.DatasourceMetaWorkflow"

          p:destroyWorkflowTemplate="classpath:/eu/dnetlib/msro/openaireplus/workflows/repo-hi/pubsRepositoryHadoop/repoBye.wf.st"

          scope="prototype">

        <property name="tree">

            <bean class="eu.dnetlib.msro.workflows.metawf.WorkflowTree"

                  p:name="collection"

                  p:template="classpath:/eu/dnetlib/msro/openaireplus/workflows/repo-hi/pubsRepositoryHadoop/collection.wf.st">

                <property name="children">

                    <list>

                        <bean class="eu.dnetlib.msro.workflows.metawf.WorkflowTree"

                              p:name="transform"

                              p:template="classpath:/eu/dnetlib/msro/openaireplus/workflows/repo-hi/pubsRepositoryHadoop/transform.wf.st" />

                    </list>

                </property>

            </bean>

        </property>

    </bean>

<NODE name="SET_INFO" isStart="true" type="SetProviderInfo">

	<DESCRIPTION>Set information about current provider</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="providerId" managedBy="system">$params.("dataprovider:id")$</PARAM>

		<PARAM required="true" type="string" name="providerName" managedBy="system">$params.("dataprovider:name")$</PARAM>

		<PARAM required="true" type="string" name="api" managedBy="system">$params.("dataprovider:interface")$</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="DeleteMetaWorkflow"/>

	</ARCS>

</NODE>

<NODE name="DeleteMetaWorkflow" type="DeleteOpenaireMetaWf">

	<DESCRIPTION>Delete the MetaWorkflow</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="metaWfId" managedBy="system">$params.("META_WORKFLOW_ID")$</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="DeleteMDStoreDC"/>

	</ARCS>

</NODE>

<NODE name="DeleteMDStoreDC" type="DeleteMDStoreHadoop">

	<DESCRIPTION>Delete the oai_dc mdstore</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="mdId" managedBy="system">$params.("harv_id")$</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="DeleteMDStoreOAF"/>

	</ARCS>

</NODE>

<NODE name="DeleteMDStoreOAF" type="DeleteMDStoreHadoop">

	<DESCRIPTION>Delete the OAF mdstore</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="mdId" managedBy="system">$params.("tran_id")$</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="RemoveApiExtraFields"/>

	</ARCS>

</NODE>

<NODE name="RemoveApiExtraFields" type="RemoveApiExtraFields">

	<DESCRIPTION>Reset the extrafields of the api</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="datasourceId" managedBy="system">$params.("dataprovider:id")$</PARAM>

		<PARAM required="true" type="string" name="datasourceInterface" managedBy="system">$params.("dataprovider:interface")$</PARAM>

		<PARAM required="true" type="string" name="fields" managedBy="system">last_collection_total, last_collection_date, last_collection_mdId, last_aggregation_total, last_aggregation_date, last_aggregation_mdId</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="success"/>

	</ARCS>

</NODE>

 <NODE name="SET_INFO" isStart="true" type="SetProviderInfo">

	<DESCRIPTION>Set information about current provider</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="providerId" managedBy="system">$params.("dataprovider:id")$</PARAM>

		<PARAM required="true" type="string" name="providerName" managedBy="system">$params.("dataprovider:name")$</PARAM>

		<PARAM required="true" type="string" name="api" managedBy="system">$params.("dataprovider:interface")$</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="obtainParams"/>

	</ARCS>

</NODE>

<NODE name="obtainParams" type="ObtainOpenaireDataSourceParams">

	<DESCRIPTION>Obtain data source params</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="providerId" managedBy="system">$params.("dataprovider:id")$</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="PREPARE_STORE_VERSION"/>

	</ARCS>

</NODE>

<NODE name="PREPARE_STORE_VERSION" type="PrepareMDStoreVersion">

	<DESCRIPTION>Prepare a new MdStore Version</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="mdId" managedBy="system">$params.("harv_id")$</PARAM>

	</PARAMETERS>

	<ARCS>

        <ARC to="PREPARE_ENV_COLLECTION"/>

    </ARCS>

</NODE>

<NODE name="PREPARE_ENV_COLLECTION" type="PrepareEnvCollectHadoopJobNode">

    <DESCRIPTION>Put in the environment all the variable needed to the collection oozie job </DESCRIPTION>

    <PARAMETERS>

        <PARAM managedBy="user" name="hdfsBasePath" required="true" type="string"></PARAM>

    </PARAMETERS>

    <ARCS>

        <ARC to="COLLECT_HADOOP"/>

    </ARCS>

</NODE>

<NODE name="COLLECT_HADOOP" type="SubmitDnetHadoopJobNode">

    <DESCRIPTION>Start the Hadoop Job</DESCRIPTION>

    <PARAMETERS>

        <PARAM managedBy="system" name="hadoopJob" required="true" type="string">dnetHadoopCollection</PARAM>

        <PARAM managedBy="user" name="cluster" required="true" type="string">DHP</PARAM>

        <PARAM managedBy="system" name="envParams" required="true" type="string">

            {

                "apiDescription":"apiDescription",

                "mdStorePath":"mdStorePath",

                "sequenceFilePath":"sequenceFilePath",

                "dataSourceInfo":"dataSourceInfo"  ,

                "timestamp":"timestamp",

                "identifierPath":"identifierPath",

                "workflowId":"workflowId"

            }

        </PARAM>

    </PARAMETERS>

    <ARCS>

        <ARC to="COMMIT_VERSION"/>

        <ARC name="abort" to="ABORT_VERSION"/>

    </ARCS>

</NODE>

<NODE name="COMMIT_VERSION" type="CommitMDStoreVersion">

	<DESCRIPTION>Commit the mdstore version</DESCRIPTION>

	<PARAMETERS/>

	<ARCS>

		<ARC to="UPDATE_INFO"/>

	</ARCS>

</NODE>

<NODE name="ABORT_VERSION" type="AbortMDStoreVersion">

	<DESCRIPTION>Abort the mdstore version</DESCRIPTION>

	<PARAMETERS/>

	<ARCS>

		<ARC to="failure"/>

	</ARCS>

</NODE>

<NODE name="UPDATE_INFO" type="MDStoreToApiExtraFieldHadoop">

	<DESCRIPTION>Update datasouce API extra fields</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="mdId" managedBy="system">$params.("harv_id")$</PARAM>

		<PARAM required="true" type="string" name="datasourceId" managedBy="system">$params.("dataprovider:id")$</PARAM>

		<PARAM required="true" type="string" name="datasourceInterface" managedBy="system">$params.("dataprovider:interface")$</PARAM>

		<PARAM required="true" type="string" name="extraFieldForTotal" managedBy="system">last_collection_total</PARAM>

		<PARAM required="true" type="string" name="extraFieldForDate" managedBy="system">last_collection_date</PARAM>

		<PARAM required="true" type="string" name="extraFieldForMdId" managedBy="system">last_collection_mdId</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="success"/>

	</ARCS>

</NODE>

<NODE name="SET_INFO" isStart="true" type="SetProviderInfo">

	<DESCRIPTION>Set information about current provider</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="providerId" managedBy="system">$params.("dataprovider:id")$</PARAM>

		<PARAM required="true" type="string" name="providerName" managedBy="system">$params.("dataprovider:name")$</PARAM>

		<PARAM required="true" type="string" name="api" managedBy="system">$params.("dataprovider:interface")$</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="obtainParams" />

	</ARCS>

</NODE>

<NODE name="obtainParams" type="ObtainOpenaireDataSourceParams">

	<DESCRIPTION>Fetch records from MDStore</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="providerId" managedBy="system">$params.("dataprovider:id")$</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="START_READING"/>

	</ARCS>

</NODE>

<NODE name="START_READING" type="StartReadingMDStore">

	<DESCRIPTION>Start reading Hadoop MD Store</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="mdId" managedBy="system">$params.("harv_id")$</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="PREPARE_STORE_VERSION"/>

	</ARCS>

</NODE>

<NODE name="PREPARE_STORE_VERSION" type="PrepareMDStoreVersion">

	<DESCRIPTION>Prepare a new MdStore Version</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="mdId" managedBy="system">$params.("tran_id")$</PARAM>

	</PARAMETERS>

	<ARCS>

        <ARC to="PREPARE_ENV_TRANSFORMATION"/>

    </ARCS>

</NODE>

<NODE name="PREPARE_ENV_TRANSFORMATION" type="PrepareEnvTransformHadoopJobNode">

    <DESCRIPTION>Retrieve all the parameters needed to run the transformation workflow</DESCRIPTION>

    <PARAMETERS>

        <PARAM category="TRANSFORMATION_RULE_ID" function="listProfiles('TransformationRuleDSResourceType', '//TITLE')" managedBy="user" name="ruleId" required="true" type="string"></PARAM>

        <PARAM managedBy="user" name="hdfsBasePath" required="true" type="string"></PARAM>

    </PARAMETERS>

    <ARCS>

        <ARC to="TRANSFORM_HADOOP"/>

    </ARCS>

</NODE>

<NODE name="TRANSFORM_HADOOP" type="SubmitDnetHadoopJobNode">

    <DESCRIPTION>Start the Hadoop Job</DESCRIPTION>

    <PARAMETERS>

        <PARAM managedBy="system" name="hadoopJob" required="true" type="string">dnetHadoopTrasnformation</PARAM>

        <PARAM managedBy="user" name="cluster" required="true" type="string">DHP</PARAM>

        <PARAM managedBy="system" name="envParams" required="true" type="string">

            {

                "mdstoreInputPath":"mdstoreInputPath",

                "mdstoreOutputPath":"mdstoreOutputPath",

                "transformationRule":"transformationRule",

                "timestamp":"timestamp",

                "workflowId":"workflowId"

            }

        </PARAM>

    </PARAMETERS>

    <ARCS>

        <ARC to="COMMIT_VERSION"/>

        <ARC name="abort" to="ABORT_VERSION"/>

    </ARCS>

</NODE>

<NODE name="COMMIT_VERSION" type="CommitMDStoreVersion">

    <DESCRIPTION>Commit the mdstore version</DESCRIPTION>

    <PARAMETERS/>

    <ARCS>

        <ARC to="END_READING"/>

    </ARCS>

</NODE>

<NODE name="END_READING" type="EndReadingMDStore">

    <DESCRIPTION>End reading Hadoop MD Store</DESCRIPTION>

    <PARAMETERS/>

    <ARCS>

        <ARC to="UPDATE_INFO"/>

    </ARCS>

</NODE>

<NODE name="ABORT_VERSION" type="AbortMDStoreVersion">

    <DESCRIPTION>Abort the mdstore version</DESCRIPTION>

    <PARAMETERS/>

    <ARCS>

        <ARC to="END_READING_ABORT"/>

    </ARCS>

</NODE>

<NODE name="END_READING_ABORT" type="EndReadingMDStore">

    <DESCRIPTION>End reading Hadoop MD Store</DESCRIPTION>

    <PARAMETERS/>

    <ARCS>

        <ARC to="failure"/>

    </ARCS>

</NODE>

<NODE name="UPDATE_INFO" type="MDStoreToApiExtraFieldHadoop">

	<DESCRIPTION>Update datasouce API extra fields</DESCRIPTION>

	<PARAMETERS>

		<PARAM required="true" type="string" name="mdId" managedBy="system">$params.("tran_id")$</PARAM>

		<PARAM required="true" type="string" name="datasourceId" managedBy="system">$params.("dataprovider:id")$</PARAM>

		<PARAM required="true" type="string" name="datasourceInterface" managedBy="system">$params.("dataprovider:interface")$</PARAM>

		<PARAM required="true" type="string" name="extraFieldForTotal" managedBy="system">last_aggregation_total</PARAM>

		<PARAM required="true" type="string" name="extraFieldForDate" managedBy="system">last_aggregation_date</PARAM>

		<PARAM required="true" type="string" name="extraFieldForMdId" managedBy="system">last_aggregation_mdId</PARAM>

	</PARAMETERS>

	<ARCS>

		<ARC to="success"/>

	</ARCS>

</NODE>

dnet.openaire.messageManager.host =

dnet.openaire.messageManager.username =  

dnet.openaire.messageManager.password =  

dnet.openaire.messageManager.ongoingMessageQueueName = 

dnet.openaire.messageManager.reportMessageQueueName =

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

       xmlns:p="http://www.springframework.org/schema/p"

       xmlns="http://www.springframework.org/schema/beans"

       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

    <bean id="dnetMessageManager" class="eu.dnetlib.msro.openaireplus.workflows.nodes.dhp.message.DnetMessageManager"

          p:messageQueueServer="${dnet.openaire.messageManager.host}"

          p:username="${dnet.openaire.messageManager.username}"

          p:password="${dnet.openaire.messageManager.password}"

          init-method="startListeningMessage"

    ></bean>

</beans>

    <bean id="wfNodeUpdateActionSets"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.actions.UpdateSetsJobNode"

          scope="prototype"/>

    <bean id="wfNodePrepareActionSets"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.actions.PrepareActionSetsJobNode"

          scope="prototype"/>

    <bean id="wfNodeCleanActionSetsProfile"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.actions.CleanActionSetsProfileJobNode"

          p:xupdate="${dnet.actionmanager.garbage.sets.xupdate}"

          scope="prototype"/>

    <bean id="wfNodeGarbageSets"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.actions.GarbageSetsJobNode"

          scope="prototype"/>

    <bean id="wfNodeGarbageSetsHDFS"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.actions.GarbageSetsHDFSJobNode"

          scope="prototype"/>

    <bean id="wfNodePromoteActionsHDFS"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.actions.PromoteActionsHDFSJobNode"

          scope="prototype"/>

    <bean id="wfNodePromoteActions"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.actions.PromoteActionsJobNode"

          scope="prototype"/>

    <bean id="wfNodePrepareCopyTable"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.hbase.PrepareCopyTableJobNode"

          scope="prototype"/>