Throw JMS Exception when listening on non-existing destination

[#159272755]

Fixes #62

(cherry picked from commit c89705a)

Author: acogoluegnes

src/main/java/com/rabbitmq/jms/client/MessageListenerConsumer.java

@@ -8,6 +8,7 @@
 import javax.jms.JMSException;
 import javax.jms.MessageListener;
 
+import com.rabbitmq.jms.util.RMQJMSException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -216,7 +217,7 @@ public void stop() {
     }
 
     @Override
-    public void start() {
+    public void start() throws Exception {
         String cT = this.getConsTag();
         logger.trace("consumerTag='{}'", cT);
         this.rejecting = false;
@@ -226,6 +227,7 @@ public void start() {
         } catch (Exception e) {
             this.completion.setComplete();  // just in case someone is waiting on it
             logger.error("basicConsume (consumerTag='{}') threw exception", cT, e);
+            throw new RMQJMSException("Error while starting consumer", e);
         }
     }
 

src/main/java/com/rabbitmq/jms/client/RMQMessageConsumer.java

@@ -17,6 +17,7 @@
 import javax.jms.Topic;
 import javax.jms.TopicSubscriber;
 
+import com.rabbitmq.jms.util.RMQJMSException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -179,15 +180,21 @@ public void setMessageListener(MessageListener messageListener) throws JMSExcept
         logger.trace("setting MessageListener({})", messageListener);
         this.removeListenerConsumer();  // if there is any
         this.messageListener = messageListener;
-        this.setNewListenerConsumer(messageListener); // if needed
+        try {
+            this.setNewListenerConsumer(messageListener); // if needed
+        } catch (JMSException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new RMQJMSException(e);
+        }
     }
 
     /**
      * Create a new RabitMQ Consumer, if necessary.
      * @param messageListener to drive from Consumer; no Consumer is created if this is null.
      * @throws IllegalStateException
      */
-    private void setNewListenerConsumer(MessageListener messageListener) throws IllegalStateException {
+    private void setNewListenerConsumer(MessageListener messageListener) throws Exception {
         if (messageListener != null) {
             MessageListenerConsumer mlConsumer =
               new MessageListenerConsumer(this,
@@ -404,7 +411,13 @@ void internalClose() throws JMSException {
         /* stop and remove any active subscription - waits for onMessage processing to finish */
         this.removeListenerConsumer();
 
-        this.abortables.abort(); // abort Consumers of both types that remain
+        try {
+            this.abortables.abort(); // abort Consumers of both types that remain
+        } catch (JMSException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new RMQJMSException(e);
+        }
 
         this.closed = true;
         this.closing = false;
@@ -471,7 +484,7 @@ private boolean getNoLocalNoException() {
      *
      * @throws InterruptedException if the thread is interrupted
      */
-    void pause() throws InterruptedException {
+    void pause() throws Exception {
         this.receiveManager.closeGate();
         this.receiveManager.waitToClear(new TimeTracker(STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS));
         this.abortables.stop();
@@ -484,7 +497,13 @@ void pause() throws InterruptedException {
      * @throws javax.jms.JMSException if the thread is interrupted
      */
     void resume() throws JMSException {
-        this.abortables.start(); // async listener restarted
+        try {
+            this.abortables.start(); // async listener restarted
+        } catch (JMSException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new RMQJMSException(e);
+        }
         this.receiveManager.openGate(); // sync listener allowed to run
     }
 

src/main/java/com/rabbitmq/jms/client/RMQSession.java

@@ -1182,9 +1182,14 @@ void pause() throws JMSException {
         for (RMQMessageConsumer consumer : this.consumers) {
             try {
                 consumer.pause();
+            } catch (JMSException e) {
+                throw e;
             } catch (InterruptedException x) {
                 logger.error("Consumer({}) pause interrupted", consumer, x);
                 throw new RMQJMSException(x);
+            } catch (Exception x) {
+                logger.error("Error while pausing consumer({})", consumer, x);
+                throw new RMQJMSException(x);
             }
         }
     }

src/main/java/com/rabbitmq/jms/util/Abortable.java

@@ -9,24 +9,24 @@ public interface Abortable {
      * Cause any other implementing threads to terminate fairly quickly, signalling abnormal termination
      * to its instigator, if necessary.
      * <p>
-     * Implementations of this method must be thread-safe, and throw no exceptions, except possibly {@link RuntimeException}s.
+     * Implementations of this method must be thread-safe.
      * </p>
      */
-    void abort();
+    void abort() throws Exception;
 
     /**
      * Cause any other implementing threads to stop temporarily.
      * <p>
-     * Implementations of this method must be thread-safe, and throw no exceptions, except possibly {@link RuntimeException}s.
+     * Implementations of this method must be thread-safe.
      * </p>
      */
-    void stop();
+    void stop() throws Exception;
 
     /**
      * Cause any other stopped threads to start.
      * <p>
-     * Implementations of this method must be thread-safe, and throw no exceptions, except possibly {@link RuntimeException}s.
+     * Implementations of this method must be thread-safe.
      * </p>
      */
-    void start();
+    void start() throws Exception;
 }

src/main/java/com/rabbitmq/jms/util/AbortableHolder.java

@@ -12,16 +12,16 @@ public class AbortableHolder implements Abortable {
     private final boolean[] flags = new boolean[] { false, false, false }; // to prevent infinite regress
 
     private enum Action {
-        ABORT(0) { @Override void doit(Abortable a) { a.abort(); } },
-        START(1) { @Override void doit(Abortable a) { a.start(); } },
-        STOP(2)  { @Override void doit(Abortable a) { a.stop();  } };
+        ABORT(0) { @Override void doit(Abortable a) throws Exception { a.abort(); } },
+        START(1) { @Override void doit(Abortable a) throws Exception { a.start(); } },
+        STOP(2)  { @Override void doit(Abortable a) throws Exception { a.stop();  } };
         private final int ind;
 
         Action(int ind) { this.ind = ind;  }
 
         int index()     { return this.ind; }
 
-        abstract void doit(Abortable a);
+        abstract void doit(Abortable a) throws Exception;
     };
 
     /**
@@ -38,13 +38,13 @@ public void remove(Abortable a) {
         this.abortableQueue.remove(a);
     }
 
-    public void abort() { act(Action.ABORT); }
+    public void abort() throws Exception { act(Action.ABORT); }
 
-    public void start() { act(Action.START); }
+    public void start() throws Exception { act(Action.START); }
 
-    public void stop()  { act(Action.STOP);  }
+    public void stop() throws Exception  { act(Action.STOP);  }
 
-    private void act(AbortableHolder.Action action) {
+    private void act(AbortableHolder.Action action) throws Exception {
         if (this.flags[action.index()]) return; // prevent infinite
         this.flags[action.index()] = true;      // regress
 

src/test/java/com/rabbitmq/integration/tests/ExceptionHandlingIT.java

@@ -0,0 +1,51 @@
+/* Copyright (c) 2018 Pivotal Software, Inc. All rights reserved. */
+
+package com.rabbitmq.integration.tests;
+
+import com.rabbitmq.jms.admin.RMQDestination;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.QueueConnection;
+import javax.jms.QueueConnectionFactory;
+import javax.jms.Session;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * To make sure an exception is thrown when starting listening on
+ * a non-existing destination.
+ */
+public class ExceptionHandlingIT {
+
+    QueueConnectionFactory connFactory;
+    QueueConnection queueConn;
+
+    @BeforeEach
+    public void beforeTests() throws Exception {
+        this.connFactory = (QueueConnectionFactory) AbstractTestConnectionFactory.getTestConnectionFactory()
+            .getConnectionFactory();
+        this.queueConn = connFactory.createQueueConnection();
+    }
+
+    @AfterEach
+    public void afterTests() throws Exception {
+        if (queueConn != null)
+            queueConn.close();
+    }
+
+    @Test
+    public void consumerOnNonExistingDestination() throws Exception {
+        queueConn.start();
+        String queueName = ExceptionHandlingIT.class.getSimpleName() + " " + System.currentTimeMillis();
+        RMQDestination queue = new RMQDestination(queueName, null, null, queueName);
+        Session session = queueConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageConsumer messageConsumer = session.createConsumer(queue);
+
+        assertThrows(JMSException.class, () -> messageConsumer.setMessageListener(msg -> {
+        }));
+    }
+}