After considerable investigation with Matthias, worked out that the ExceptionHandler probably shouldn't be being used, because the DefaultEH does stuff which is probably useful, or at least we don't understand the implications of not using it. SSL Handshake seems to be done lazily - at the point at which data is first exchanged on the socket. Thus the socket opening is not enough to generate the SSL Handshake Exceptions. Now because of the race between the user thread and the MainLoop as to who first really accesses the socket (could be the user thread sending the header, or the main loop blocking on read), it was not deterministic which thread would get he ssl handshake exception.

So slightly tidied the AMQConnection class so that it doesn't do so many bad things in its constructor, and that we make sure we send the header first, before starting up the MainLoop thread. Thus if the MainLoop thread gets started then we know the SSL Handshake has been done. This means that the exception comes out in the right thread.

Thus also reverted the factory so that you can't set the ExceptionHandler. All tests pass. There were some associated changes elsewhere, which eclipse found.

Author: Matthew Sackman

src/com/rabbitmq/client/ConnectionFactory.java

@@ -41,8 +41,6 @@
 import javax.net.ssl.TrustManager;
 
 import com.rabbitmq.client.impl.AMQConnection;
-import com.rabbitmq.client.impl.DefaultExceptionHandler;
-import com.rabbitmq.client.impl.ExceptionHandler;
 import com.rabbitmq.client.impl.FrameHandler;
 import com.rabbitmq.client.impl.SocketFrameHandler;
 
@@ -57,11 +55,6 @@ public class ConnectionFactory {
      * Holds the SocketFactory used to manufacture outbound sockets.
      */
     private SocketFactory _factory = SocketFactory.getDefault();
-
-    /**
-     * Holds the ExceptionHandler used in new AMQConnections
-     */
-    private ExceptionHandler _exceptionHandler = new DefaultExceptionHandler();
 
     /**
      * Instantiate a ConnectionFactory with a default set of parameters.
@@ -104,23 +97,6 @@ public void setSocketFactory(SocketFactory factory) {
         _factory = factory;
     }
 
-    /**
-     * Retrieve the ExceptionHandler used in new AMQConnections
-     * @return 
-     */
-    public ExceptionHandler getExceptionHandler() {
-        return _exceptionHandler;
-    }
-    
-    /**
-     * Set the ExceptionHandler used in new AMQConnections
-     * 
-     * @param handler The ExceptionHandler to use
-     */
-    public void setExceptionHandler(ExceptionHandler handler) {
-        _exceptionHandler = handler;
-    }
-    
     /**
      * Convenience method for setting up a SSL socket factory, using
      * the DEFAULT_SSL_PROTOCOL and a trusting TrustManager.
@@ -197,7 +173,10 @@ private Connection newConnection(Address[] addrs,
                         redirectCount = 0;
                     boolean allowRedirects = redirectCount < maxRedirects;
                     try {
-                        return new AMQConnection(_params, !allowRedirects, frameHandler, _exceptionHandler);
+                        AMQConnection conn = new AMQConnection(_params,
+                                    frameHandler);
+                        conn.startConnection(!allowRedirects);
+                        return conn;
                     } catch (RedirectException e) {
                         if (!allowRedirects) {
                             //this should never happen with a well-behaved server

src/com/rabbitmq/client/impl/AMQConnection.java

@@ -47,6 +47,7 @@
 import com.rabbitmq.client.MissedHeartbeatException;
 import com.rabbitmq.client.RedirectException;
 import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.impl.AMQChannel.SimpleBlockingRpcContinuation;
 import com.rabbitmq.utility.BlockingCell;
 import com.rabbitmq.utility.Utility;
 
@@ -167,31 +168,22 @@ public Address[] getKnownHosts() {
     /**
      * Construct a new connection to a broker.
      * @param params the initialization parameters for a connection
-     * @param insist true if broker redirects are disallowed
      * @param frameHandler interface to an object that will handle the frame I/O for this connection
-     * @throws RedirectException if the server is redirecting us to a different host/port
-     * @throws java.io.IOException if an error is encountered
      */
     public AMQConnection(ConnectionParameters params,
-                         boolean insist,
-                         FrameHandler frameHandler) throws RedirectException, IOException {
-        this(params, insist, frameHandler, new DefaultExceptionHandler());
+                         FrameHandler frameHandler) {
+        this(params, frameHandler, new DefaultExceptionHandler());
     }
 
     /**
      * Construct a new connection to a broker.
      * @param params the initialization parameters for a connection
-     * @param insist true if broker redirects are disallowed
      * @param frameHandler interface to an object that will handle the frame I/O for this connection
      * @param exceptionHandler interface to an object that will handle any special exceptions encountered while using this connection
-     * @throws RedirectException if the server is redirecting us to a different host/port
-     * @throws java.io.IOException if an error is encountered
      */
     public AMQConnection(ConnectionParameters params,
-                         boolean insist,
                          FrameHandler frameHandler,
                          ExceptionHandler exceptionHandler)
-        throws RedirectException, IOException
     {
         checkPreconditions();
         _params = params;
@@ -202,10 +194,36 @@ public AMQConnection(ConnectionParameters params,
         _heartbeat = 0;
         _exceptionHandler = exceptionHandler;
         _brokerInitiatedShutdown = false;
+    }
 
-        new MainLoop(); // start the main loop going
-
-        _knownHosts = open(_params, insist);
+    /**
+     * Start up the connection, including the MainLoop thread
+     * @param insist true if broker redirects are disallowed
+     * @throws RedirectException if the server is redirecting us to a different host/port
+     * @throws java.io.IOException if an error is encountered
+     */
+    public void startConnection(boolean insist)
+        throws IOException, RedirectException
+    {
+        // Make sure that the first thing we do is to send the header,
+        // which should cause any socket errors to show up for us, rather
+        // than risking them pop out in the MainLoop
+        AMQChannel.SimpleBlockingRpcContinuation connStartBlocker =
+            new AMQChannel.SimpleBlockingRpcContinuation();
+        // We enqueue an RPC continuation here without sending an RPC
+        // request, since the protocol specifies that after sending
+        // the version negotiation header, the client (connection
+        // initiator) is to wait for a connection.start method to
+        // arrive.
+        _channel0.enqueueRpc(connStartBlocker);
+        // The following two lines are akin to AMQChannel's
+        // transmit() method for this pseudo-RPC.
+        _frameHandler.setTimeout(HANDSHAKE_TIMEOUT);
+        _frameHandler.sendHeader();
+
+        new MainLoop().start(); // start the main loop going
+
+        _knownHosts = open(_params, insist, connStartBlocker);
     }
 
     /**
@@ -316,28 +334,16 @@ public Map<String, Object> buildClientPropertiesTable() {
      * calls Connection.Open and waits for the OpenOk. Sets heartbeat
      * and frame max values after tuning has taken place.
      * @param params the construction parameters for a Connection
+     * @param connStartBlocker the blocker we're waiting on for the start-ok
      * @return the known hosts that came back in the connection.open-ok
      * @throws RedirectException if the server asks us to redirect to
      *                           a different host/port.
      * @throws java.io.IOException if any other I/O error occurs
      */
-    public Address[] open(final ConnectionParameters params, boolean insist)
+    public Address[] open(final ConnectionParameters params, boolean insist, SimpleBlockingRpcContinuation connStartBlocker)
         throws RedirectException, IOException
     {
         try {
-            AMQChannel.SimpleBlockingRpcContinuation connStartBlocker =
-                new AMQChannel.SimpleBlockingRpcContinuation();
-            // We enqueue an RPC continuation here without sending an RPC
-            // request, since the protocol specifies that after sending
-            // the version negotiation header, the client (connection
-            // initiator) is to wait for a connection.start method to
-            // arrive.
-            _channel0.enqueueRpc(connStartBlocker);
-            // The following two lines are akin to AMQChannel's
-            // transmit() method for this pseudo-RPC.
-            _frameHandler.setTimeout(HANDSHAKE_TIMEOUT);
-            _frameHandler.sendHeader();
-
             // See bug 17389. The MainLoop could have shut down already in
             // which case we don't want to wait forever for a reply.
 
@@ -414,11 +420,6 @@ private static int negotiatedMaxValue(int clientValue, int serverValue) {
 
     private class MainLoop extends Thread {
 
-        /** Start the main loop going. */
-        public MainLoop() {
-            start();
-        }
-
         /**
          * Channel reader thread main loop. Reads a frame, and if it is
          * not a heartbeat frame, dispatches it to the channel it refers to.

test/src/com/rabbitmq/client/test/AMQConnectionTest.java

@@ -103,7 +103,7 @@ public void testConnectionSendsSingleHeaderAndTimesOut() {
         MyExceptionHandler handler = new MyExceptionHandler();
         assertEquals(0, _mockFrameHandler.countHeadersSent());
         try {
-            new AMQConnection(_params, false, _mockFrameHandler, handler);
+            new AMQConnection(_params, _mockFrameHandler, handler).startConnection(false);
             fail("Connection should have thrown exception");
         } catch(IOException signal) {
            // As expected 

test/src/com/rabbitmq/client/test/BrokenFramesTest.java

@@ -78,7 +78,7 @@ public void testNoMethod() throws Exception {
         myFrameHandler.setFrames(frames.iterator());
 
         try {
-            new AMQConnection(params, false, myFrameHandler);
+            new AMQConnection(params, myFrameHandler).startConnection(false);
         } catch (IOException e) {
             UnexpectedFrameError unexpectedFrameError = findUnexpectedFrameError(e);
             assertNotNull(unexpectedFrameError);
@@ -104,7 +104,7 @@ public void testMethodThenBody() throws Exception {
         myFrameHandler.setFrames(frames.iterator());
 
         try {
-            new AMQConnection(params, false, myFrameHandler);
+            new AMQConnection(params, myFrameHandler).startConnection(false);
         } catch (IOException e) {
             UnexpectedFrameError unexpectedFrameError = findUnexpectedFrameError(e);
             assertNotNull(unexpectedFrameError);

test/src/com/rabbitmq/client/test/ssl/BadVerifiedConnection.java

@@ -41,46 +41,20 @@
 
 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLHandshakeException;
 import javax.net.ssl.TrustManagerFactory;
 
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.client.Consumer;
-import com.rabbitmq.client.impl.ExceptionHandler;
 
 /**
  * Test for bug 19356 - SSL Support in rabbitmq
  *
  */
 public class BadVerifiedConnection extends UnverifiedConnection {
-    private boolean exceptionCaught = false;
-
     public void openConnection()
         throws IOException
     {
-        final Object lock = new Object();
-        
         connectionFactory = new ConnectionFactory();
-        connectionFactory.setExceptionHandler(new ExceptionHandler() {
-
-            public void handleConsumerException(Channel channel,
-                    Throwable exception, Consumer consumer, String consumerTag,
-                    String methodName) {
-            }
-
-            public void handleReturnListenerException(Channel channel,
-                    Throwable exception) {
-            }
-
-            public void handleUnexpectedConnectionDriverException(
-                    Connection conn, Throwable exception) {
-                synchronized (lock) {
-                    exceptionCaught = true;
-                    lock.notifyAll();
-                }
-            }});
-        
         try {
             String keystorePath = System.getProperty("keystore.empty.path");
             assertNotNull(keystorePath);
@@ -117,18 +91,9 @@ public void handleUnexpectedConnectionDriverException(
             try {
                 connection = connectionFactory.newConnection("localhost", 5671);
                 fail();
-            } catch (Throwable e) {
-                synchronized (lock) {
-                    while (! exceptionCaught) {
-                        try {
-                            lock.wait(10000);
-                            if (! exceptionCaught) {
-                                fail();
-                            }
-                        } catch (InterruptedException e1) {
-                        }
-                    }
-                }
+            } catch (SSLHandshakeException e) {
+            } catch (IOException e) {
+                fail();
             }
         }
     }