1 files changed, 244 insertions, 0 deletions
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java
new file mode 100644
index 0000000000..141de1e5a8
--- /dev/null
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java
@@ -0,0 +1,244 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you 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 org.apache.qpid.test.unit.client.connection;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.Connection;
+import javax.jms.ExceptionListener;
+import javax.jms.IllegalStateException;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.Queue;
+import javax.jms.Session;
+
+import org.apache.qpid.AMQConnectionClosedException;
+import org.apache.qpid.client.AMQNoRouteException;
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+import org.apache.qpid.transport.ConnectionException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ExceptionListenerTest extends QpidBrokerTestCase
+{
+    private static final Logger LOGGER = LoggerFactory.getLogger(ExceptionListenerTest.class);
+
+    private volatile Throwable _lastExceptionListenerException = null;
+
+    public void testExceptionListenerHearsBrokerShutdown() throws  Exception
+    {
+        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(1);
+        final AtomicInteger exceptionCounter = new AtomicInteger(0);
+        final ExceptionListener listener = new ExceptionListener()
+        {
+            public void onException(JMSException exception)
+            {
+                exceptionCounter.incrementAndGet();
+                _lastExceptionListenerException = exception;
+                exceptionReceivedLatch.countDown();
+            }
+        };
+
+        Connection connection = getConnection();
+        connection.setExceptionListener(listener);
+
+        stopBroker();
+
+        exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
+
+        assertEquals("Unexpected number of exceptions received", 1, exceptionCounter.intValue());
+        LOGGER.debug("exception was", _lastExceptionListenerException);
+        assertNotNull("Exception should have cause", _lastExceptionListenerException.getCause());
+        Class<? extends Exception> expectedExceptionClass = isBroker010() ? ConnectionException.class : AMQConnectionClosedException.class;
+        assertEquals(expectedExceptionClass, _lastExceptionListenerException.getCause().getClass());
+    }
+
+    /**
+     * It is reasonable for an application to perform Connection#close within the exception
+     * listener.  This test verifies that close is allowed, and proceeds without generating
+     * further exceptions.
+     */
+    public void testExceptionListenerClosesConnection_IsAllowed() throws  Exception
+    {
+        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(1);
+        final Connection connection = getConnection();
+        final ExceptionListener listener = new ExceptionListener()
+        {
+            public void onException(JMSException exception)
+            {
+                try
+                {
+                    connection.close();
+                    // PASS
+                }
+                catch (Throwable t)
+                {
+                    _lastExceptionListenerException = t;
+                }
+                finally
+                {
+                    exceptionReceivedLatch.countDown();
+                }
+            }
+        };
+        connection.setExceptionListener(listener);
+
+
+        stopBroker();
+
+        boolean exceptionReceived = exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
+        assertTrue("Exception listener did not hear exception within timeout", exceptionReceived);
+        assertNull("Connection#close() should not have thrown exception", _lastExceptionListenerException);
+    }
+
+    /**
+     * Spring's SingleConnectionFactory installs an ExceptionListener that calls stop()
+     * and ignores any IllegalStateException that result.  This test serves to test this
+     * scenario.
+     */
+    public void testExceptionListenerStopsConnection_ThrowsIllegalStateException() throws  Exception
+    {
+        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(1);
+        final Connection connection = getConnection();
+        final ExceptionListener listener = new ExceptionListener()
+        {
+            public void onException(JMSException exception)
+            {
+                try
+                {
+                    connection.stop();
+                    fail("Exception not thrown");
+                }
+                catch (IllegalStateException ise)
+                {
+                    // PASS
+                }
+                catch (Throwable t)
+                {
+                    _lastExceptionListenerException = t;
+                }
+                finally
+                {
+                    exceptionReceivedLatch.countDown();
+                }
+            }
+        };
+        connection.setExceptionListener(listener);
+
+        stopBroker();
+
+        boolean exceptionReceived = exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
+        assertTrue("Exception listener did not hear exception within timeout", exceptionReceived);
+        assertNull("Connection#stop() should not have thrown unexpected exception", _lastExceptionListenerException);
+    }
+
+    /**
+     * This test reproduces a deadlock that was the subject of a support call. A Spring based
+     * application was using SingleConnectionFactory.  It installed an ExceptionListener that
+     * stops and closes the connection in response to any exception.  On receipt of a message
+     * the application would create a new session then send a response message (within onMessage).
+     * It appears that a misconfiguration in the application meant that some of these messages
+     * were bounced (no-route). Bounces are treated like connection exceptions and are passed
+     * back to the application via the ExceptionListener.  The deadlock occurred between the
+     * ExceptionListener's call to stop() and the MessageListener's attempt to create a new
+     * session.
+     */
+    public void testExceptionListenerConnectionStopDeadlock() throws  Exception
+    {
+        Queue messageQueue = getTestQueue();
+
+        Map<String, String> options = new HashMap<String, String>();
+        options.put(ConnectionURL.OPTIONS_CLOSE_WHEN_NO_ROUTE, Boolean.toString(false));
+
+        final Connection connection = getConnectionWithOptions(options);
+
+        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+        session.createConsumer(messageQueue).close(); // Create queue by side-effect
+
+        // Put 10 messages onto messageQueue
+        sendMessage(session, messageQueue, 10);
+
+        // Install an exception listener that stops/closes the connection on receipt of 2nd AMQNoRouteException.
+        // (Triggering on the 2nd (rather than 1st) seems to increase the probability that the test ends in deadlock,
+        // at least on my machine).
+        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(2);
+        final ExceptionListener listener = new ExceptionListener()
+        {
+            public void onException(JMSException exception)
+            {
+                try
+                {
+                    assertNotNull("JMS Exception must have cause", exception.getCause() );
+                    assertEquals("JMS Exception is of wrong type", AMQNoRouteException.class, exception.getCause().getClass());
+                    exceptionReceivedLatch.countDown();
+                    if (exceptionReceivedLatch.getCount() == 0)
+                    {
+                        connection.stop(); // ** Deadlock
+                        connection.close();
+                    }
+                }
+                catch (Throwable t)
+                {
+                    _lastExceptionListenerException = t;
+                }
+            }
+        };
+        connection.setExceptionListener(listener);
+
+        // Create a message listener that receives from testQueue and tries to forward them to unknown queue (thus
+        // provoking AMQNoRouteException exceptions to be delivered to the ExceptionListener).
+        final Queue unknownQueue = session.createQueue(getTestQueueName() + "_unknown");;
+        MessageListener redirectingMessageListener = new MessageListener()
+        {
+            @Override
+            public void onMessage(Message msg)
+            {
+                try
+                {
+                    Session mlSession = connection.createSession(true, Session.SESSION_TRANSACTED);  // ** Deadlock
+                    mlSession.createProducer(unknownQueue).send(msg);
+                    mlSession.commit();
+                }
+                catch (JMSException je)
+                {
+                    // Connection is closed by the listener, so exceptions here are expected.
+                    LOGGER.debug("Expected exception - message listener got exception", je);
+                }
+            }
+        };
+
+        MessageConsumer consumer = session.createConsumer(messageQueue);
+        consumer.setMessageListener(redirectingMessageListener);
+        connection.start();
+
+        // Await the 2nd exception
+        boolean exceptionReceived = exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
+        assertTrue("Exception listener did not hear exception within timeout", exceptionReceived);
+        assertNull("Exception listener should not have had experienced exception", _lastExceptionListenerException);
+    }
+}