QPID-6374 : Always call exception listener from connection task pool, rather than from within the failover mutex

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1658689 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 39 insertions, 16 deletions

diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
index c7fcde824a..5518435b94 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
@@ -1349,16 +1349,25 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
 
     private void deliverJMSExceptionToExceptionListenerOrLog(final JMSException je, final Throwable cause)
     {
-        // deliver the exception if there is a listener
-        ExceptionListener exceptionListener = getExceptionListenerNoCheck();
+        final ExceptionListener exceptionListener = getExceptionListenerNoCheck();
         if (exceptionListener != null)
         {
-            exceptionListener.onException(je);
+            performConnectionTask(new Runnable()
+                                    {
+                                        @Override
+                                        public void run()
+                                        {
+                                            // deliver the exception if there is a listener
+                                                exceptionListener.onException(je);
+                                        }
+                                    });
         }
         else
         {
             _logger.error("Throwable Received but no listener set: " + cause);
         }
+
+
     }
 
     private boolean hardError(Throwable cause)
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
index 4e9164c3b0..fdeab7ae70 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
@@ -291,7 +291,7 @@ public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, Connec
 
     public void closed(Connection conn)
     {
-        ConnectionException exc = exception;
+        final ConnectionException exc = exception;
         exception = null;
 
         if (exc == null)
@@ -299,7 +299,7 @@ public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, Connec
             return;
         }
 
-        ConnectionClose close = exc.getClose();
+        final ConnectionClose close = exc.getClose();
         if (close == null || close.getReplyCode() == ConnectionCloseCode.CONNECTION_FORCED)
         {
             _conn.getProtocolHandler().setFailoverLatch(new CountDownLatch(1));
@@ -332,23 +332,31 @@ public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, Connec
 
         _conn.setClosed();
 
-        ExceptionListener listener = _conn.getExceptionListenerNoCheck();
+        final ExceptionListener listener = _conn.getExceptionListenerNoCheck();
         if (listener == null)
         {
             _logger.error("connection exception: " + conn, exc);
         }
         else
         {
-            String code = null;
-            if (close != null)
+            _conn.performConnectionTask(new Runnable()
             {
-                code = close.getReplyCode().toString();
-            }
+                @Override
+                public void run()
+                {
+                    String code = null;
+                    if (close != null)
+                    {
+                        code = close.getReplyCode().toString();
+                    }
+
+                    JMSException ex = new JMSException(exc.getMessage(), code);
+                    ex.setLinkedException(exc);
+                    ex.initCause(exc);
+                    listener.onException(ex);
+                }
+            });
 
-            JMSException ex = new JMSException(exc.getMessage(), code);
-            ex.setLinkedException(exc);
-            ex.initCause(exc);
-            listener.onException(ex);
         }
     }
 
diff --git a/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java b/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java
index 009598d8a4..ceb2a323ca 100644
--- a/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java
+++ b/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java
@@ -20,14 +20,14 @@
  */
 package org.apache.qpid.client;
 
+import java.io.IOException;
+
 import org.apache.qpid.AMQException;
 import org.apache.qpid.client.state.AMQState;
 import org.apache.qpid.framing.ProtocolVersion;
 import org.apache.qpid.jms.BrokerDetails;
 import org.apache.qpid.url.URLSyntaxException;
 
-import java.io.IOException;
-
 public class MockAMQConnection extends AMQConnection
 {
     public MockAMQConnection(String broker, String username, String password, String clientName, String virtualHost)
@@ -60,4 +60,10 @@ public class MockAMQConnection extends AMQConnection
     {
         return super.getDelegate();
     }
+
+    @Override
+    public void performConnectionTask(final Runnable task)
+    {
+        task.run();
+    }
 }