send messages for DLE directly to DLE queue

reimer-atb · reimer-atb · commit 744740e86335 · 2022-04-21T17:32:29.000+02:00
not through topic exchange
because DLE does not connect to topic exchange
diff --git a/context-monitoring/src/main/java/de/atb/context/monitoring/monitors/messagebroker/util/MessageBrokerUtil.java b/context-monitoring/src/main/java/de/atb/context/monitoring/monitors/messagebroker/util/MessageBrokerUtil.java
@@ -14,11 +14,6 @@
  * #L%
  */
 
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.UUID;
-import java.util.concurrent.TimeoutException;
-
 import com.google.gson.Gson;
 import com.rabbitmq.client.BuiltinExchangeType;
 import com.rabbitmq.client.CancelCallback;
@@ -31,6 +26,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+import java.util.concurrent.TimeoutException;
+
 /**
  * Helper class wrapping methods for interacting with message broker.
  */
@@ -58,22 +58,7 @@ public static Channel connectToTopicExchange(final String host,
                                                  final String userName,
                                                  final String password,
                                                  final String exchange) throws IOException, TimeoutException {
-        LOGGER.info("Connecting to messagebroker {}:{} with user {}", host, port, userName != null ? userName : "<null>");
-        final ConnectionFactory factory = new ConnectionFactory();
-
-        factory.setHost(host);
-        factory.setPort(port);
-
-        if (StringUtils.isNotBlank(userName)) {
-            factory.setUsername(userName);
-        }
-        if (StringUtils.isNotBlank(password)) {
-            factory.setPassword(password);
-        }
-
-        factory.setAutomaticRecoveryEnabled(true);
-
-        final Connection connection = factory.newConnection();
+        final Connection connection = getConnection(host, port, userName, password);
 
         final Channel channel = connection.createChannel();
 
@@ -84,40 +69,53 @@ public static Channel connectToTopicExchange(final String host,
     }
 
     /**
-     * Connect to the message broker specified by {@code host} and {@code port}.
+     * Connect to message broker. Message broker details and credentials are specified in the given {@code dataSource}.
      * Create the given {@code exchange} if it does not exist yet.
      *
-     * @param host     the host where the message broker is running
-     * @param port     the port where the message broker is listening
-     * @param exchange the topic exchange's name
+     * @param dataSource the {@link MessageBrokerDataSource} containing the message broker connection details
      * @return a {@link Channel} object representing the established connection to the message broker
      * @throws IOException      in case of error
      * @throws TimeoutException in case of error
      * @see MessageBrokerUtil#connectToTopicExchange(String, int, String, String, String)
      */
-    public static Channel connectToTopicExchange(final String host,
-                                                 final int port,
-                                                 final String exchange) throws IOException, TimeoutException {
-        return connectToTopicExchange(host, port, null, null, exchange);
+    public static Channel connectToTopicExchange(final MessageBrokerDataSource dataSource)
+        throws IOException, TimeoutException {
+        return connectToTopicExchange(
+            dataSource.getMessageBrokerServer(),
+            dataSource.getMessageBrokerPort(),
+            dataSource.getUserName(),
+            dataSource.getPassword(),
+            dataSource.getExchange()
+        );
     }
 
     /**
-     * Connect to message broker. Message broker details and credentials are specified in the given {@code dataSource}.
-     * Create the given {@code exchange} if it does not exist yet.
+     * Connect to the message broker specified by {@code host} and {@code port}
+     * with the credentials specified by {@code userName} and {@code password}.
+     * Create the given {@code queue} if it does not exist yet.
      *
-     * @param dataSource the {@link MessageBrokerDataSource} containing the message broker connection details
+     * @param host     the host where the message broker is running
+     * @param port     the port where the message broker is listening
+     * @param userName the username to use when connecting to message broker - optional
+     * @param password the password to use when connecting to message broker - optional
+     * @param queue    the queue's name
      * @return a {@link Channel} object representing the established connection to the message broker
      * @throws IOException      in case of error
      * @throws TimeoutException in case of error
-     * @see MessageBrokerUtil#connectToTopicExchange(String, int, String, String, String)
      */
-    public static Channel connectToTopicExchange(final MessageBrokerDataSource dataSource)
-            throws IOException, TimeoutException {
-        return connectToTopicExchange(dataSource.getMessageBrokerServer(),
-                                      dataSource.getMessageBrokerPort(),
-                                      dataSource.getUserName(),
-                                      dataSource.getPassword(),
-                                      dataSource.getExchange());
+    public static Channel connectToQueue(final String host,
+                                         final int port,
+                                         final String userName,
+                                         final String password,
+                                         final String queue) throws IOException, TimeoutException {
+        final Connection connection = getConnection(host, port, userName, password);
+
+        final Channel channel = connection.createChannel();
+
+        LOGGER.info("Creating queue {}", queue);
+        channel.queueDeclare(queue, true, false, false, null);
+
+        return channel;
     }
 
     /**
@@ -165,12 +163,14 @@ public static void registerListenerOnTopic(final Channel channel,
                                                final MessageBrokerDataSource dataSource,
                                                final DeliverCallback deliverCallback,
                                                final CancelCallback cancelCallback) throws IOException {
-        registerListenerOnTopic(channel,
-                                dataSource.getExchange(),
-                                dataSource.getTopic(),
-                                dataSource.getId(),
-                                deliverCallback,
-                                cancelCallback);
+        registerListenerOnTopic(
+            channel,
+            dataSource.getExchange(),
+            dataSource.getTopic(),
+            dataSource.getId(),
+            deliverCallback,
+            cancelCallback
+        );
     }
 
     /**
@@ -191,14 +191,46 @@ public static void convertAndSendToTopic(final Channel channel,
                                              final String topic,
                                              final Object payload) throws IOException {
         final String jsonMessage = GSON.toJson(payload);
-        sendToTopic(channel, exchange, topic, jsonMessage);
-    }
-
-    private static void sendToTopic(final Channel channel,
-                                    final String exchange,
-                                    final String topic,
-                                    final String jsonMessage) throws IOException {
         LOGGER.info("Publishing message to topic {}/{}: {}", exchange, topic, jsonMessage);
         channel.basicPublish(exchange, topic, null, jsonMessage.getBytes(StandardCharsets.UTF_8));
     }
+
+    /**
+     * Converts the given {@code payload} object to a JSON string and sends it to the given {@code queue}.
+     * <p>
+     * Use {@link MessageBrokerUtil#connectToQueue(String, int, String, String, String)} to create {@link Channel}.
+     *
+     * @param channel the {@link Channel} object representing the established connection to the message broker
+     * @param queue   the queue's name
+     * @param payload the object to send
+     * @throws IOException in case of error
+     */
+    public static void convertAndSendToQueue(final Channel channel, final String queue, final Object payload)
+        throws IOException {
+        final String jsonMessage = GSON.toJson(payload);
+        LOGGER.info("Publishing message to queue {}: {}", queue, jsonMessage);
+        channel.basicPublish("", queue, null, jsonMessage.getBytes(StandardCharsets.UTF_8));
+    }
+
+    private static Connection getConnection(final String host,
+                                            final int port,
+                                            final String userName,
+                                            final String password) throws IOException, TimeoutException {
+        LOGGER.info("Connecting to messagebroker {}:{} with user {}", host, port, userName != null ? userName : "<null>");
+        final ConnectionFactory factory = new ConnectionFactory();
+
+        factory.setHost(host);
+        factory.setPort(port);
+
+        if (StringUtils.isNotBlank(userName)) {
+            factory.setUsername(userName);
+        }
+        if (StringUtils.isNotBlank(password)) {
+            factory.setPassword(password);
+        }
+
+        factory.setAutomaticRecoveryEnabled(true);
+
+        return factory.newConnection();
+    }
 }
diff --git a/smartclide-monitoring/src/main/java/eu/smartclide/contexthandling/dle/listener/DleGitMonitorProgressListener.java b/smartclide-monitoring/src/main/java/eu/smartclide/contexthandling/dle/listener/DleGitMonitorProgressListener.java
@@ -14,10 +14,6 @@
  * #L%
  */
 
-import java.io.IOException;
-import java.util.List;
-import java.util.concurrent.TimeoutException;
-
 import com.rabbitmq.client.Channel;
 import de.atb.context.monitoring.config.models.datasources.MessageBrokerDataSource;
 import de.atb.context.monitoring.events.MonitoringProgressListener;
@@ -31,19 +27,27 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+
 public class DleGitMonitorProgressListener implements MonitoringProgressListener<String, IMonitoringDataModel<?, ?>> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DleGitMonitorProgressListener.class);
 
-    private final String topic;
-    private final String exchange;
+    private final String dleQueue;
     private final Channel channel;
 
     public DleGitMonitorProgressListener(final MessageBrokerDataSource messageBrokerDataSource)
             throws IOException, TimeoutException {
-        exchange = messageBrokerDataSource.getExchange();
-        topic = messageBrokerDataSource.getDleTopic();
-        channel = MessageBrokerUtil.connectToTopicExchange(messageBrokerDataSource);
+        dleQueue = messageBrokerDataSource.getDleTopic();
+        channel = MessageBrokerUtil.connectToQueue(
+                messageBrokerDataSource.getMessageBrokerServer(),
+                messageBrokerDataSource.getMessageBrokerPort(),
+                messageBrokerDataSource.getUserName(),
+                messageBrokerDataSource.getPassword(),
+                dleQueue
+        );
     }
 
     @Override
@@ -71,20 +75,20 @@ public void documentAnalysed(final List<IMonitoringDataModel<?, ?>> analysed,
     private DleMessage convertToDleMessage(final GitMessage gitMessage) {
         return DleMessage.builder()
                 .monitor(CommitMessage.builder()
-                                 .user(gitMessage.getUser())
-                                 .branch(gitMessage.getBranch())
-                                 .files(gitMessage.getNoOfModifiedFiles())
-                                 .build())
+                        .user(gitMessage.getUser())
+                        .branch(gitMessage.getBranch())
+                        .files(gitMessage.getNoOfModifiedFiles())
+                        .build())
                 .build();
     }
 
     private void send(final DleMessage dleMessage) {
         try {
             // simulate that actual context-extraction will take some time
             Thread.sleep(1000);
-            MessageBrokerUtil.convertAndSendToTopic(channel, exchange, topic, dleMessage);
+            MessageBrokerUtil.convertAndSendToQueue(channel, dleQueue, dleMessage);
         } catch (Exception e) {
-            LOGGER.error("Failed to send {} to {}/{}", dleMessage, exchange, topic, e);
+            LOGGER.error("Failed to send {} to {}/{}", dleMessage, "", dleQueue, e);
         }
     }
 }
diff --git a/smartclide-monitoring/src/test/java/de/atb/context/monitoring/TestDataRetrieval.java b/smartclide-monitoring/src/test/java/de/atb/context/monitoring/TestDataRetrieval.java
@@ -1,15 +1,5 @@
 package de.atb.context.monitoring;
 
-import java.io.File;
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Path;
-import java.time.ZonedDateTime;
-import java.time.format.DateTimeFormatter;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeoutException;
-
 import com.rabbitmq.client.Channel;
 import de.atb.context.common.ContextPathUtils;
 import de.atb.context.common.util.ApplicationScenario;
@@ -31,6 +21,16 @@
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.RabbitMQContainer;
 
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+
 import static org.junit.Assert.assertEquals;
 
 /**
@@ -44,10 +44,9 @@ public class TestDataRetrieval {
     private static final Logger logger = LoggerFactory.getLogger(TestDataRetrieval.class);
 
     private static final String RABBITMQ_3_ALPINE = "rabbitmq:3-alpine";
-    private static final String EXCHANGE_NAME = "smartclide-monitoring";
+    private static final String EXCHANGE_NAME = "mom";
     private static final String ROUTING_KEY_MONITORING = "monitoring.git.commits";
-    private static final String ROUTING_KEY_DLE = "dle.git.commits";
-    private static final String QUEUE_PREFIX_DLE = "Fake-DLE";
+    private static final String QUEUE_NAME_DLE = "code_repo_recommendation_queue";
     private static final String DATASOURCE_GIT = "datasource-git";
     private static final String MONITORING_CONFIG_FILE_NAME = "monitoring-config.xml";
     private static final String AMI_REPOSITORY_ID = "AmI-repository";
@@ -66,10 +65,16 @@ public void setup() throws Exception {
         // setup message broker
         final String rabbitMQContainerHost = container.getHost();
         final Integer rabbitMQContainerAmqpPort = container.getAmqpPort();
-        channel = MessageBrokerUtil.connectToTopicExchange(rabbitMQContainerHost, rabbitMQContainerAmqpPort, EXCHANGE_NAME);
+        channel = MessageBrokerUtil.connectToTopicExchange(
+                rabbitMQContainerHost,
+                rabbitMQContainerAmqpPort,
+                null,
+                null,
+                EXCHANGE_NAME
+        );
 
         // setup fake DLE
-        createFakeDleListener();
+        createFakeDleListener(rabbitMQContainerHost, rabbitMQContainerAmqpPort);
 
         // write dynamically allocated message broker host and port to monitoring config file
         final Path monitoringConfigFilePath = ContextPathUtils.getConfigDirPath().resolve(MONITORING_CONFIG_FILE_NAME);
@@ -145,15 +150,20 @@ private void updateMessageBrokerDataSource(final Path monitoringConfig, final St
         persister.write(config, new File(monitoringConfig.toString()));
     }
 
-    private void createFakeDleListener() throws IOException {
-        MessageBrokerUtil.registerListenerOnTopic(
-                channel,
-                EXCHANGE_NAME,
-                ROUTING_KEY_DLE,
-                QUEUE_PREFIX_DLE,
+    private void createFakeDleListener(final String rabbitMQContainerHost, final Integer rabbitMQContainerAmqpPort)
+            throws IOException, TimeoutException {
+        final Channel channel = MessageBrokerUtil.connectToQueue(
+                rabbitMQContainerHost,
+                rabbitMQContainerAmqpPort,
+                null,
+                null,
+                QUEUE_NAME_DLE
+        );
+        channel.basicConsume(
+                QUEUE_NAME_DLE,
+                true,
                 (t, m) -> logger.info("DLE received message: {}", new String(m.getBody(), StandardCharsets.UTF_8)),
                 (t) -> logger.info("cancelled!")
         );
     }
-
 }
diff --git a/smartclide-monitoring/src/test/resources/config/monitoring-config.xml b/smartclide-monitoring/src/test/resources/config/monitoring-config.xml
@@ -11,7 +11,7 @@
         <datasource id="datasource-git" type="messagebroker"
                     monitor="de.atb.context.monitoring.monitors.GitMonitor"
                     uri=""
-                    options="server=localhost&amp;port=5672&amp;exchange=smartclide-monitoring&amp;topic=monitoring.git.*&amp;dle-topic=dle.git.commits"
+                    options="server=localhost&amp;port=5672&amp;exchange=mom&amp;topic=monitoring.git.*&amp;dle-topic=code_repo_recommendation_queue"
                     class="de.atb.context.monitoring.config.models.datasources.MessageBrokerDataSource"/>
     </datasources>
 
