Implementing "publisher Confirms " in REMReM to get confirmations about the messages sent to MB. (#242)

* Implemented "publisher confirms " in REMReM so that this can be used
to get confirmation about the messages  sent to MB.

Author: Vaishnavi-Enugala

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 2.0.25
+- Implemented "publisher confirms " in REMReM so that this  can be used to get confirmation about the messages sent to MB.
+
 ## 2.0.24
 - Updated all the curl commands in documentation
 - Uplifted eiffel-remrem-parent version from 2.0.6 to 2.0.7.

pom.xml

@@ -9,7 +9,7 @@
         <version>2.0.8</version>
     </parent>
     <properties>
-        <eiffel-remrem-publish.version>2.0.24</eiffel-remrem-publish.version>
+        <eiffel-remrem-publish.version>2.0.25</eiffel-remrem-publish.version>
         <eiffel-remrem-semantics.version>2.2.3</eiffel-remrem-semantics.version>
     </properties>
     <artifactId>eiffel-remrem-publish</artifactId>

publish-cli/src/main/java/com/ericsson/eiffel/remrem/publish/cli/CliOptions.java

@@ -79,6 +79,7 @@ public static void createCLIOptions() {
         options.addOption("mp", "messaging_protocol", true, "name of messaging protocol to be used, e.g. eiffel3, eiffelsemantics, default is eiffelsemantics");
         options.addOption("domain", "domainId", true, "identifies the domain that produces the event");
         options.addOption("cc", "channelsCount", true, "Number of channels connected to message bus, default is 1");
+        options.addOption("wcto", "wait_for_confirms_timeOut", true, "the timeout for wait for confirms, default is 5000 ms/milliseconds");
         options.addOption("ud", "user_domain_suffix", true, "user domain suffix");
         options.addOption("v", "lists the versions of publish and all loaded protocols");
         options.addOption("tag", "tag", true, "tag to be used in routing key");
@@ -225,6 +226,12 @@ public static void handleMessageBusOptions() throws HandleMessageBusException {
             System.setProperty(key, channelsCount);
         }
 
+        if (commandLine.hasOption("wcto")) {
+            String timeOut = commandLine.getOptionValue("wcto");
+            String key = PropertiesConfig.WAIT_FOR_CONFIRMS_TIME_OUT;
+            System.setProperty(key,timeOut);
+        }
+
         if (commandLine.hasOption("tls")) {
             String tls_ver = commandLine.getOptionValue("tls");
             if (tls_ver == null) {
@@ -272,6 +279,7 @@ public static void clearSystemProperties() {
         System.clearProperty(PropertiesConfig.TLS);
         System.clearProperty(PropertiesConfig.DOMAIN_ID);
         System.clearProperty(PropertiesConfig.CHANNELS_COUNT);
+        System.clearProperty(PropertiesConfig.WAIT_FOR_CONFIRMS_TIME_OUT);
     }
 
     /**

publish-cli/src/test/java/com/ericsson/eiffel/remrem/publish/cli/CliOptionsUnitTests.java

@@ -125,6 +125,20 @@ public void testEnOptionFails() throws Exception {
         assertTrue(CliOptions.getErrorCodes().contains(code));
     }
 
+    public void testWctoOption() throws Exception {
+        String[] args = {"-f", "/a/b/c/test.file",  "test", "-wcto", "5000"};
+        CliOptions.parse(args);
+        assertTrue(CliOptions.getErrorCodes().isEmpty());
+    }
+
+    @Test
+    public void testWctoOptionFails() throws Exception {
+        String[] args = {"-f", "/a/b/c/test.file",  "test", "-wctof", "5000"};
+        CliOptions.parse(args);
+        int code = CLIExitCodes.CLI_MISSING_OPTION_EXCEPTION;
+        assertTrue(CliOptions.getErrorCodes().contains(code));
+    }
+
     public void testNpOption() throws Exception {
         String[] args = {"-f", "/a/b/c/test.file",  "test", "-np", "non_persistent"};
         CliOptions.parse(args);

publish-common/src/main/java/com/ericsson/eiffel/remrem/publish/config/PropertiesConfig.java

@@ -19,6 +19,7 @@ public class PropertiesConfig {
     public static final String MESSAGE_BUS_PORT = "com.ericsson.eiffel.remrem.publish.messagebus.port";
     public static final String VIRTUAL_HOST = "com.ericsson.eiffel.remrem.publish.messagebus.virtualhost";
     public static final String CHANNELS_COUNT = "com.ericsson.eiffel.remrem.publish.messagebus.channels";
+    public static final String WAIT_FOR_CONFIRMS_TIME_OUT = "com.ericsson.eiffel.remrem.publish.messagebus.waitforconfirmstimeout";
     public static final String TLS = "com.ericsson.eiffel.remrem.publish.messagebus.tls";
     public static final String EXCHANGE_NAME = "com.ericsson.eiffel.remrem.publish.exchange.name";
     public static final String USE_PERSISTENCE = "com.ericsson.eiffel.remrem.publish.use.persistence";
@@ -45,7 +46,10 @@ public class PropertiesConfig {
     public static final String INVALID_EXCHANGE_MESSAGE_SERVICE = " ExchangeName is not present, To create the exchange specify createExchangeIfNotExisting in application configuration";
 
     public static final String SERVER_DOWN = "Internal Server Error";
+    public static final String GATEWAY_TIMEOUT = "Gateway Timeout";
     public static final String SERVER_DOWN_MESSAGE = "RabbitMQ is down. Please try later";
+    public static final String MESSAGE_NACK = "Message is nacked";
+    public static final String TIMEOUT_WAITING_FOR_ACK = "Time out waiting for ACK";
     public static final String ROUTING_KEY_GENERATION_FAILED_CONTENT = "Could not prepare Routing key to publish message";
     public static final String UNSUCCESSFUL_EVENT_CONTENT = "Please check previous event and try again later";
     public static final String RABBITMQ_PROPERTIES_NOT_FOUND = "RabbitMQ properties not found";

publish-common/src/main/java/com/ericsson/eiffel/remrem/publish/config/RabbitMqPropertiesConfig.java

@@ -115,7 +115,14 @@ private void readSpringProperties() {
                 rabbitMqProperties.setCreateExchangeIfNotExisting(rabbitmqInstanceObject.get("createExchangeIfNotExisting").asBoolean());
                 rabbitMqProperties.setDomainId(rabbitmqInstanceObject.get("domainId").asText());
                 if((rabbitmqInstanceObject.get("channelsCount") != null) ) {
-                    rabbitMqProperties.setChannelsCount(Integer.parseInt(rabbitmqInstanceObject.get("channelsCount").asText()));  
+                    rabbitMqProperties.setChannelsCount(
+                            Integer.getInteger(rabbitmqInstanceObject.get("channelsCount").asText(),
+                                    RabbitMqProperties.DEFAULT_CHANNEL_COUNT));
+                }
+                if((rabbitmqInstanceObject.get("waitForConfirmsTimeOut") != null) ) {
+                    rabbitMqProperties.setWaitForConfirmsTimeOut(Long.getLong(
+                            rabbitmqInstanceObject.get("waitForConfirmsTimeOut").asText(),
+                            RabbitMqProperties.DEFAULT_WAIT_FOR_CONFIRMS_TIMEOUT));
                 }
                 rabbitMqPropertiesMap.put(protocol, rabbitMqProperties);
             }

publish-common/src/main/java/com/ericsson/eiffel/remrem/publish/exception/NackException.java

@@ -0,0 +1,29 @@
+/*
+    Copyright 2022 Ericsson AB.
+    For a full list of individual contributors, please see the commit history.
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+*/
+package com.ericsson.eiffel.remrem.publish.exception;
+
+import java.io.IOException;
+
+public class NackException extends IOException {
+
+    private static final long serialVersionUID = 1872149119179038290L;
+
+    public NackException(String message) {
+        super(message);
+    }
+    public NackException(String message , Throwable cause) {
+        super(message, cause);
+    }
+}

publish-common/src/main/java/com/ericsson/eiffel/remrem/publish/helper/RMQHelper.java

@@ -18,6 +18,7 @@
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeoutException;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
@@ -29,6 +30,7 @@
 import com.ericsson.eiffel.remrem.protocol.MsgService;
 import com.ericsson.eiffel.remrem.publish.config.PropertiesConfig;
 import com.ericsson.eiffel.remrem.publish.config.RabbitMqPropertiesConfig;
+import com.ericsson.eiffel.remrem.publish.exception.NackException;
 import com.ericsson.eiffel.remrem.publish.exception.RemRemPublishException;
 
 import ch.qos.logback.classic.Level;
@@ -81,28 +83,30 @@ public void rabbitMqPropertiesInit(String protocol) throws RemRemPublishExceptio
      * @param protocol name
      * @throws RemRemPublishException
      */
-    private void protocolInit(String protocol) throws RemRemPublishException{
-        rabbitMqPropertiesMap.get(protocol).setProtocol(protocol);
-        rabbitMqPropertiesMap.get(protocol).init();
+    private void protocolInit(String protocol) throws RemRemPublishException {
+        RabbitMqProperties rabbitmqProtocolProperties = rabbitMqPropertiesMap.get(protocol);
+        rabbitmqProtocolProperties.setProtocol(protocol);
+        rabbitmqProtocolProperties.init();
     }
 
-    public void send(String routingKey, String msg, MsgService msgService) throws IOException {
+    public void send(String routingKey, String msg, MsgService msgService) throws IOException, NackException, TimeoutException, RemRemPublishException {
         String protocol = msgService.getServiceName();
-        if(rabbitMqPropertiesMap.get(protocol) != null) {
-            rabbitMqPropertiesMap.get(protocol).send(routingKey,msg);
+        RabbitMqProperties rabbitmqProtocolProperties = rabbitMqPropertiesMap.get(protocol);
+        if (rabbitmqProtocolProperties != null) {
+            rabbitmqProtocolProperties.send(routingKey, msg);
         } else {
-            log.error("RabbitMq properties not configured for the protocol "+protocol);
+            log.error("RabbitMq properties not configured for the protocol " + protocol);
         }
     }
 
     @PreDestroy
     public void cleanUp() throws IOException {
         log.info("RMQHelper cleanUp ...");
         for(String protocol : rabbitMqPropertiesMap.keySet()) {
-            rabbitMqPropertiesMap.get(protocol).getRabbitConnection();
-            if (rabbitMqPropertiesMap.get(protocol).getRabbitConnection() != null){
-                rabbitMqPropertiesMap.get(protocol).getRabbitConnection().close();
-                rabbitMqPropertiesMap.get(protocol).setRabbitConnection(null);
+            RabbitMqProperties rabbitmqProtocolProperties = rabbitMqPropertiesMap.get(protocol);
+            if (rabbitmqProtocolProperties.getRabbitConnection() != null) {
+                rabbitmqProtocolProperties.getRabbitConnection().close();
+                rabbitmqProtocolProperties.setRabbitConnection(null);
             } else {
                 log.warn("rabbitConnection is null when cleanUp");
             }
@@ -118,4 +122,5 @@ private void handleLogging() {
             log.setLevel(Level.OFF);
         }
     }
+
 }

publish-common/src/main/java/com/ericsson/eiffel/remrem/publish/helper/RabbitMqProperties.java

@@ -26,6 +26,7 @@
 
 import com.ericsson.eiffel.remrem.publish.config.PropertiesConfig;
 import com.ericsson.eiffel.remrem.publish.exception.RemRemPublishException;
+import com.ericsson.eiffel.remrem.publish.exception.NackException;
 import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
@@ -51,6 +52,9 @@ public class RabbitMqProperties {
     private String domainId;
     private Integer channelsCount;
     private boolean createExchangeIfNotExisting;
+    private Long waitForConfirmsTimeOut;
+    public static final Long DEFAULT_WAIT_FOR_CONFIRMS_TIMEOUT = 5000L;
+    public static final Integer DEFAULT_CHANNEL_COUNT = 1;
 
     private Connection rabbitConnection;
     private String protocol;
@@ -59,6 +63,15 @@ public class RabbitMqProperties {
 
     Logger log = (Logger) LoggerFactory.getLogger(RMQHelper.class);
 
+
+    public Long getWaitForConfirmsTimeOut() {
+        return waitForConfirmsTimeOut;
+    }
+
+    public void setWaitForConfirmsTimeOut(Long waitForConfirmsTimeOut) {
+        this.waitForConfirmsTimeOut = waitForConfirmsTimeOut;
+    }
+
     public String getHost() {
         return host;
     }
@@ -219,20 +232,24 @@ public void init() throws RemRemPublishException {
 
     /**
      * This method is used to create Rabbitmq connection and channels
+     * @throws RemRemPublishException
      */
-    public void createRabbitMqConnection() {
+    public void createRabbitMqConnection() throws RemRemPublishException {
         try {
             rabbitConnection = factory.newConnection();
             log.info("Connected to RabbitMQ.");
             rabbitChannels = new ArrayList<>();
             if(channelsCount == null || channelsCount == 0 ) {
-                channelsCount = 1;
+                channelsCount = DEFAULT_CHANNEL_COUNT;
             }
             for (int i = 0; i < channelsCount; i++) {
-                rabbitChannels.add(rabbitConnection.createChannel());
+                Channel channel = rabbitConnection.createChannel();
+                channel.confirmSelect();
+                rabbitChannels.add(channel);
             }
         } catch (IOException | TimeoutException e) {
             log.error(e.getMessage(), e);
+            throw new RemRemPublishException("Failed to create connection for Rabbitmq ::" + factory.getHost() + ":" + factory.getPort());
         }
     }
 
@@ -276,6 +293,9 @@ private void initService() {
         if (channelsCount == null ) {
             channelsCount = Integer.getInteger(getValuesFromSystemProperties(protocol + ".rabbitmq.channelsCount"));
         }
+        if (waitForConfirmsTimeOut == null ) {
+            waitForConfirmsTimeOut = Long.getLong(getValuesFromSystemProperties(protocol + ".rabbitmq.waitForConfirmsTimeOut"));
+        }
     }
 
 
@@ -285,6 +305,7 @@ private void setValues() {
         virtualHost = getValuesFromSystemProperties(PropertiesConfig.VIRTUAL_HOST);
         domainId = getValuesFromSystemProperties(PropertiesConfig.DOMAIN_ID);
         channelsCount = Integer.getInteger(PropertiesConfig.CHANNELS_COUNT);
+        waitForConfirmsTimeOut = Long.getLong(PropertiesConfig.WAIT_FOR_CONFIRMS_TIME_OUT);
         tlsVer = getValuesFromSystemProperties(PropertiesConfig.TLS);
         exchangeName = getValuesFromSystemProperties(PropertiesConfig.EXCHANGE_NAME);
         usePersitance = Boolean.getBoolean(PropertiesConfig.USE_PERSISTENCE);
@@ -400,40 +421,60 @@ private boolean hasExchange() throws RemRemPublishException {
      * @param routingKey
      * @param msg is Eiffel Event
      * @throws IOException
+     * @throws NackException
+     * @throws TimeoutException
+     * @throws RemRemPublishException
      */
-    public void send(String routingKey, String msg) throws IOException {
-
-        Channel channel = giveMeRandomChannel();
-        channel.addShutdownListener(new ShutdownListener() {
-            public void shutdownCompleted(ShutdownSignalException cause) {
-                // Beware that proper synchronization is needed here
-                if (cause.isInitiatedByApplication()) {
-                    log.debug("Shutdown is initiated by application. Ignoring it.");
-                } else {
-                    log.error("Shutdown is NOT initiated by application.");
-                    log.error(cause.getMessage());
-                    boolean cliMode = Boolean.getBoolean(PropertiesConfig.CLI_MODE);
-                    if (cliMode) {
-                        System.exit(-3);
+    public void send(String routingKey, String msg)
+            throws IOException, NackException, TimeoutException, RemRemPublishException {
+            Channel channel = giveMeRandomChannel();
+            channel.addShutdownListener(new ShutdownListener() {
+                public void shutdownCompleted(ShutdownSignalException cause) {
+                    // Beware that proper synchronization is needed here
+                    if (cause.isInitiatedByApplication()) {
+                        log.debug("Shutdown is initiated by application. Ignoring it.");
+                    } else {
+                        log.error("Shutdown is NOT initiated by application.");
+                        log.error(cause.getMessage());
+                        boolean cliMode = Boolean.getBoolean(PropertiesConfig.CLI_MODE);
+                        if (cliMode) {
+                            System.exit(-3);
+                        }
                     }
                 }
+            });
+            BasicProperties msgProps = MessageProperties.BASIC;
+            if (usePersitance)
+                msgProps = MessageProperties.PERSISTENT_BASIC;
+        try {
+            channel.basicPublish(exchangeName, routingKey, msgProps, msg.getBytes());
+            log.info("Published message with size {} bytes on exchange '{}' with routing key '{}'",
+                    msg.getBytes().length, exchangeName, routingKey);
+            if (waitForConfirmsTimeOut == null || waitForConfirmsTimeOut == 0) {
+                waitForConfirmsTimeOut = DEFAULT_WAIT_FOR_CONFIRMS_TIMEOUT;
             }
-        });
-
-        BasicProperties msgProps = MessageProperties.BASIC;
-        if (usePersitance)
-            msgProps = MessageProperties.PERSISTENT_BASIC;
-
-        channel.basicPublish(exchangeName, routingKey, msgProps, msg.getBytes());
-        log.info("Published message with size {} bytes on exchange '{}' with routing key '{}'", msg.getBytes().length,
-                exchangeName, routingKey);
+            channel.waitForConfirmsOrDie(waitForConfirmsTimeOut);
+        } catch (InterruptedException | IOException e) {
+            log.error("Failed to publish message due to " + e.getMessage());
+            throw new NackException("The message is nacked due to " + e.getMessage(), e);
+        } catch (TimeoutException e) {
+            log.error("Failed to publish message due to " + e.getMessage());
+            throw new TimeoutException("Timeout waiting for ACK " + e.getMessage());
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+            if(!channel.isOpen()&& rabbitConnection.isOpen()){
+                throw new RemRemPublishException("Channel was closed for Rabbitmq connection ::" + factory.getHost() + factory.getPort());
+            }
+            throw new IOException("Failed to publish message due to " + e.getMessage());
+        }
     }
 
     /**
      * This method is used to give random channel
      * @return channel
+     * @throws RemRemPublishException
      */
-    private Channel giveMeRandomChannel() {
+    private Channel giveMeRandomChannel() throws RemRemPublishException {
         if ((rabbitConnection == null || !rabbitConnection.isOpen())) {
             createRabbitMqConnection();
         }