You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ta...@apache.org on 2018/04/10 19:37:54 UTC
qpid-jms git commit: QPIDJMS-376 Fire JMS ExceptionListener event on
consumer remote close
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 <ta...@gmail.com>
Authored: Tue Apr 10 15:37:34 2018 -0400
Committer: Timothy Bish <ta...@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-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org