Repository: qpid-jms
Updated Branches:
  refs/heads/master f05c184cd -> 18c017da8


QPIDJMS-376 Fire JMS ExceptionListener event on consumer remote close

When a MessageConsumer has a MessageListener fire the JMS
ExceptionListener to signal the error as the consumer would not
otherwise indicate to the user it had closed and would not be handling
messages.

Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/18c017da
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/18c017da
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/18c017da

Branch: refs/heads/master
Commit: 18c017da8cd41d21fc4a8a763dd9797cadd6c37a
Parents: f05c184
Author: Timothy Bish <tabish...@gmail.com>
Authored: Tue Apr 10 15:37:34 2018 -0400
Committer: Timothy Bish <tabish...@gmail.com>
Committed: Tue Apr 10 15:37:34 2018 -0400

----------------------------------------------------------------------
 .../java/org/apache/qpid/jms/JmsSession.java    |  3 +
 .../integration/ConsumerIntegrationTest.java    | 79 ++++++++++++++++++++
 2 files changed, 82 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/18c017da/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java 
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
index a168df5..192cfc8 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
@@ -355,6 +355,9 @@ public class JmsSession implements AutoCloseable, Session, 
QueueSession, TopicSe
         LOG.info("A JMS MessageConsumer has been closed: {}", resource);
 
         JmsMessageConsumer consumer = consumers.get(resource.getId());
+        if (consumer.hasMessageListener()) {
+            connection.onAsyncException(JmsExceptionSupport.create(cause));
+        }
 
         try {
             if (consumer != null) {

http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/18c017da/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
----------------------------------------------------------------------
diff --git 
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
 
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
index 8dc074a..e8a506a 100644
--- 
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
+++ 
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
@@ -21,6 +21,7 @@ package org.apache.qpid.jms.integration;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
@@ -141,7 +142,15 @@ public class ConsumerIntegrationTest extends 
QpidJmsTestCase {
 
         try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
             final AtomicBoolean consumerClosed = new AtomicBoolean();
+            final AtomicBoolean exceptionFired = new AtomicBoolean();
+
             JmsConnection connection = (JmsConnection) 
testFixture.establishConnecton(testPeer);
+            connection.setExceptionListener(new ExceptionListener() {
+                @Override
+                public void onException(JMSException exception) {
+                    exceptionFired.set(true);
+                }
+            });
             connection.addConnectionListener(new 
JmsDefaultConnectionListener() {
                 @Override
                 public void onConsumerClosed(MessageConsumer consumer, 
Throwable exception) {
@@ -181,6 +190,76 @@ public class ConsumerIntegrationTest extends 
QpidJmsTestCase {
             }, 10000, 10));
 
             assertTrue("Consumer closed callback didn't trigger", 
consumerClosed.get());
+            assertFalse("JMS Exception listener shouldn't fire with no 
MessageListener", exceptionFired.get());
+
+            // Try closing it explicitly, should effectively no-op in client.
+            // The test peer will throw during close if it sends anything.
+            consumer.close();
+        }
+    }
+
+    @Test(timeout = 20000)
+    public void 
testRemotelyCloseConsumerWithMessageListenerFiresJMSExceptionListener() throws 
Exception {
+        final String BREAD_CRUMB = "ErrorMessage";
+
+        try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+            final AtomicBoolean consumerClosed = new AtomicBoolean();
+            final AtomicBoolean exceptionFired = new AtomicBoolean();
+
+            JmsConnection connection = (JmsConnection) 
testFixture.establishConnecton(testPeer);
+            connection.setExceptionListener(new ExceptionListener() {
+                @Override
+                public void onException(JMSException exception) {
+                    LOG.trace("JMS ExceptionListener: ", exception);
+                    exceptionFired.set(true);
+                }
+            });
+            connection.addConnectionListener(new 
JmsDefaultConnectionListener() {
+                @Override
+                public void onConsumerClosed(MessageConsumer consumer, 
Throwable exception) {
+                    consumerClosed.set(true);
+                }
+            });
+
+            testPeer.expectBegin();
+            Session session = connection.createSession(false, 
Session.AUTO_ACKNOWLEDGE);
+
+            // Create a consumer, then remotely end it afterwards.
+            testPeer.expectReceiverAttach();
+            testPeer.expectLinkFlow();
+            testPeer.remotelyDetachLastOpenedLinkOnLastOpenedSession(true, 
true, AmqpError.RESOURCE_DELETED, BREAD_CRUMB, 10);
+
+            Queue queue = session.createQueue("myQueue");
+            final MessageConsumer consumer = session.createConsumer(queue);
+            consumer.setMessageListener(new MessageListener() {
+
+                @Override
+                public void onMessage(Message message) {
+                }
+            });
+
+            // Verify the consumer gets marked closed
+            testPeer.waitForAllHandlersToComplete(1000);
+            assertTrue("consumer never closed.", Wait.waitFor(new 
Wait.Condition() {
+                @Override
+                public boolean isSatisified() throws Exception {
+                    try {
+                        consumer.getMessageListener();
+                    } catch (IllegalStateException jmsise) {
+                        if (jmsise.getCause() != null) {
+                            String message = jmsise.getCause().getMessage();
+                            return 
message.contains(AmqpError.RESOURCE_DELETED.toString()) &&
+                                   message.contains(BREAD_CRUMB);
+                        } else {
+                            return false;
+                        }
+                    }
+                    return false;
+                }
+            }, 10000, 10));
+
+            assertTrue("Consumer closed callback didn't trigger", 
consumerClosed.get());
+            assertTrue("JMS Exception listener shouldn't fire with no 
MessageListener", exceptionFired.get());
 
             // Try closing it explicitly, should effectively no-op in client.
             // The test peer will throw during close if it sends anything.


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org
For additional commands, e-mail: commits-h...@qpid.apache.org

Reply via email to