D-Net

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;

    }

}