Added code changes so that REMReM Publish should not die even (#248)

* Added code changes so that REMReM Publish should not die even
if RabbitMQ is unavailable when starting up

Author: Vaishnavi-Enugala

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 2.0.26
+- Added code changes so that REMReM Publish should not die if RabbitMQ is unavailable when starting up.
+
 ## 2.0.25
 - Implemented "publisher confirms " in REMReM so that this  can be used to get confirmation about the messages sent to MB.
 - Implemented configurable parameters for TCP connection timeout against LDAP and MB

pom.xml

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

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

@@ -14,11 +14,22 @@
 */
 package com.ericsson.eiffel.remrem.publish.exception;
 
+import com.ericsson.eiffel.remrem.publish.helper.RMQBeanConnectionFactory;
+
 public class RemRemPublishException extends Exception {
 
     private static final long serialVersionUID = 1L;
 
     public RemRemPublishException(String message) {
         super(message);
     }
+
+    public RemRemPublishException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public RemRemPublishException(String message, RMQBeanConnectionFactory factory,
+            Throwable cause) {
+        super(message + factory.getHost() + ":" + factory.getPort(), cause);
+    }
 }

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

@@ -53,6 +53,7 @@ public class RabbitMqProperties {
     private Integer channelsCount;
     private boolean createExchangeIfNotExisting;
     private Integer tcpTimeOut;
+    private boolean hasExchange = false;
 //  built in tcp connection timeout value for MB in milliseconds.
     public static final Integer DEFAULT_TCP_TIMEOUT = 60000;
     private Long waitForConfirmsTimeOut;
@@ -88,7 +89,10 @@ public String getExchangeName() {
     }
 
     public void setExchangeName(String exchangeName) {
-        this.exchangeName = exchangeName;
+        if (!exchangeName.equals(this.exchangeName)) {
+            this.exchangeName = exchangeName;
+            this.hasExchange = false;
+        }
     }
 
     public Integer getPort() {
@@ -183,7 +187,7 @@ public void setRabbitConnection(Connection rabbitConnection) {
         this.rabbitConnection = rabbitConnection;
     }
 
-    public void init() throws RemRemPublishException {
+    public void init() {
         log.info("RabbitMqProperties init ...");
         if (Boolean.getBoolean(PropertiesConfig.CLI_MODE)) {
             initCli();
@@ -238,7 +242,13 @@ public void init() throws RemRemPublishException {
         } catch (NoSuchAlgorithmException e) {
             log.error(e.getMessage(), e);            
         }
-        checkAndCreateExchangeIfNeeded();
+        try {
+            //The exception can be safely handled here as there is a check for existence of exchange is done before each publish.
+            checkAndCreateExchangeIfNeeded();
+        } catch (RemRemPublishException e) {
+            log.error("Error occured while setting up the RabbitMq Connection. "+e.getMessage());
+            e.printStackTrace();
+        }
     }
 
     /**
@@ -264,7 +274,8 @@ public void createRabbitMqConnection() throws RemRemPublishException {
             }
         } catch (IOException | TimeoutException e) {
             log.error(e.getMessage(), e);
-            throw new RemRemPublishException("Failed to create connection for Rabbitmq ::" + factory.getHost() + ":" + factory.getPort());
+            throw new RemRemPublishException("Failed to create connection for Rabbitmq :: ", factory,
+                    e);
         }
     }
 
@@ -364,19 +375,26 @@ public void checkAndCreateExchangeIfNeeded() throws RemRemPublishException {
                 try {
                     connection = factory.newConnection();
                 } catch (final IOException | TimeoutException e) {
-                    throw new RemRemPublishException("Exception occurred while creating Rabbitmq connection ::" + factory.getHost() + ":" + factory.getPort() + e.getMessage());
+                    throw new RemRemPublishException(
+                            "Exception occurred while creating Rabbitmq connection :: ", factory, e);
                 }
                 Channel channel = null;
                 try {
                     channel = connection.createChannel();
                 } catch (final IOException e) {
-                    throw new RemRemPublishException("Exception occurred while creating Channel with Rabbitmq connection ::" + factory.getHost() + ":" + factory.getPort() + e.getMessage());
+                    throw new RemRemPublishException(
+                            "Exception occurred while creating Channel with Rabbitmq connection ::",
+                            factory, e);
                 }
                 try {
                     channel.exchangeDeclare(exchangeName, "topic", true);
+                    log.info("Exchange {} is created",exchangeName);
+                    hasExchange = true;
                 } catch (final IOException e) {
                     log.info(exchangeName + "failed to create an exchange");
-                    throw new RemRemPublishException("Unable to create Exchange with Rabbitmq connection " + exchangeName + factory.getHost() + ":" + factory.getPort() + e.getMessage());
+                    throw new RemRemPublishException(
+                            "Unable to create Exchange with Rabbitmq connection " + exchangeName,
+                            factory, e);
                 } finally {
                     if (channel == null || channel.isOpen()) {
                         try {
@@ -405,25 +423,32 @@ public void checkAndCreateExchangeIfNeeded() throws RemRemPublishException {
      * @throws IOException
      */
     private boolean hasExchange() throws RemRemPublishException {
-        log.info("Exchange is: " + exchangeName);
+        if(hasExchange) {
+           log.info("Exchange is: {}", exchangeName);
+           return true;
+        }
+
         Connection connection;
         try {
             connection = factory.newConnection();
         } catch (final IOException | TimeoutException e) {
-            throw new RemRemPublishException("Exception occurred while creating Rabbitmq connection ::" + factory.getHost() + factory.getPort() + e.getMessage());
+            throw new RemRemPublishException(
+                    "Exception occurred while creating Rabbitmq connection :: ", factory, e);
         }
         Channel channel = null;
         try {
             channel = connection.createChannel();
         } catch (final IOException e) {
-            log.info("Exchange " + exchangeName + " does not Exist");
-            throw new RemRemPublishException("Exception occurred while creating Channel with Rabbitmq connection ::" + factory.getHost() + factory.getPort() + e.getMessage());
+            throw new RemRemPublishException(
+                    "Exception occurred while creating Channel with Rabbitmq connection :: ",
+                    factory, e);
         }
         try {
             channel.exchangeDeclarePassive(exchangeName);
-            return true;
+            hasExchange = true;
+            return hasExchange;
         } catch (final IOException e) {
-            log.info("Exchange " + exchangeName + " does not Exist");
+            log.info("Exchange " + exchangeName + " was not created");
             return false;
         } finally {
             if (channel != null && channel.isOpen()) {
@@ -437,7 +462,8 @@ private boolean hasExchange() throws RemRemPublishException {
         }
     }
 
-    /**
+
+	/**
      * This method is used to publish the message to RabbitMQ
      * @param routingKey
      * @param msg is Eiffel Event
@@ -449,6 +475,7 @@ private boolean hasExchange() throws RemRemPublishException {
     public void send(String routingKey, String msg)
             throws IOException, NackException, TimeoutException, RemRemPublishException, IllegalArgumentException {
             Channel channel = giveMeRandomChannel();
+            checkAndCreateExchangeIfNeeded();
             channel.addShutdownListener(new ShutdownListener() {
                 public void shutdownCompleted(ShutdownSignalException cause) {
                     // Beware that proper synchronization is needed here
@@ -483,13 +510,14 @@ public void shutdownCompleted(ShutdownSignalException cause) {
             throw new TimeoutException("Timeout waiting for ACK " + e.getMessage());
         } catch (IllegalArgumentException e) {
             log.error("Failed to publish message due to " + e.getMessage());
-            throw new IllegalArgumentException("DomainId limit exceeded " + e.getMessage());
+            throw new IllegalArgumentException("DomainId limit exceeded " + e.getMessage(), e);
         } 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 RemRemPublishException("Channel was closed for Rabbitmq connection :: ",
+                        factory, e);
             }
-            throw new IOException("Failed to publish message due to " + e.getMessage());
+            throw new IOException("Failed to publish message due to " + e.getMessage(), e);
         }
     }
 

publish-service/src/test/java/com/ericsson/eiffel/remrem/publish/service/MessageServiceRMQImplUnitTest.java

@@ -86,19 +86,19 @@ public class MessageServiceRMQImplUnitTest {
      * exchange it will throw an Exception.
      */
     @Test
-    public void testCreateExchangeIfNotExistingEnable() throws RemRemPublishException {
-        boolean ExceptionOccured = false;
+    public void testCreateExchangeIfNotExistingEnable() {
+        boolean exceptionOccured = false;
         rabbitmqProtocolProperties.setExchangeName("nonexistexchangename");
         rabbitmqProtocolProperties.setCreateExchangeIfNotExisting(createExchangeIfNotExisting);
         try {
-            rabbitmqProtocolProperties.init();
+            rabbitmqProtocolProperties.checkAndCreateExchangeIfNeeded();
         } catch (RemRemPublishException e) {
-            ExceptionOccured = true;
+            exceptionOccured = true;
         } finally {
             rabbitmqProtocolProperties.setExchangeName(exchangeName);
             rabbitmqProtocolProperties.init();
         }
-        assertFalse("An exception occured while creating a exchange", ExceptionOccured);
+        assertFalse("An exception occured while creating a exchange", exceptionOccured);
     }
 
     /**
@@ -110,7 +110,7 @@ public void testCreateExchangeIfNotExistingEnable() throws RemRemPublishExceptio
     public void testCreateExchangeIfNotExistingDisable() throws RemRemPublishException  {
         rabbitmqProtocolProperties.setExchangeName("test76888");
         rabbitmqProtocolProperties.setCreateExchangeIfNotExisting(false);
-        rabbitmqProtocolProperties.init();
+        rabbitmqProtocolProperties.checkAndCreateExchangeIfNeeded();
 
         rabbitmqProtocolProperties.setExchangeName(exchangeName);
         rabbitmqProtocolProperties.init();