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;

    }

}