D-Net

package eu.dnetlib.msro.message;

import eu.dnetlib.message.Message;

import eu.dnetlib.message.MessageManager;

import eu.dnetlib.message.MessageType;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import java.util.HashMap;

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, Message> reportMessages = new HashMap<>();

    private String messageQueueServer;

    private String username;

    private String password;

    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) {

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

            manager.startConsumingMessage("dev_ongoing", true, false);

            manager.startConsumingMessage("dev_report", 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) {

                                reportMessages.put(createReportId(currentMessage.getWorkflowId(), currentMessage.getJobName()), 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 Message getReport(final String workflowId, final String jobName) {

        final String reportId = createReportId(workflowId, jobName);

        return getMessage(reportMessages, reportId);

    }

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

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

        job.getParameters().put("wfId",wfId);

        Runnable r = () -> {

          while (ongoing) {

              Message mess = dnetMessageManager.getOnGoingMessages(wfId);

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

                  try {

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

                      Thread.sleep(1000);

                  } catch (Throwable 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;

                Message report = dnetMessageManager.getReport(wfId, "Collection");

                if (report!= null)

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

    }

}

					       type="string">Aggregate Metadata (publications) from PubsRepository [Hadoop]

<?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.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="wfNodeSubmitDnetHadoopJobNode"

		  class="eu.dnetlib.msro.openaireplus.workflows.nodes.hadoop.SubmitDnetHadoopJobNode"

		  scope="prototype" />

	<version>7.0.0-hadoop-SNAPSHOT</version>