Merge bug24651

Author: Simon MacMullen

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

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.SocketException;
+import java.net.SocketTimeoutException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -108,6 +109,9 @@ public static final Map<String, Object> defaultClientProperties() {
     /** Flag indicating whether the client received Connection.Close message from the broker */
     private volatile boolean _brokerInitiatedShutdown;
 
+    /** Flag indicating we are still negotiating the connection in start */
+    private volatile boolean _inConnectionNegotiation;
+
     /** Manages heart-beat sending for this connection */
     private final HeartbeatSender _heartbeatSender;
 
@@ -248,6 +252,8 @@ public AMQConnection(String username,
 
         this._heartbeatSender = new HeartbeatSender(frameHandler);
         this._brokerInitiatedShutdown = false;
+
+        this._inConnectionNegotiation = true; // we start out waiting for the first protocol response
     }
 
     /**
@@ -385,6 +391,9 @@ public void start()
             throw AMQChannel.wrap(sse);
         }
 
+        // We can now respond to errors having finished tailoring the connection
+        this._inConnectionNegotiation = false;
+
         return;
     }
 
@@ -538,7 +547,11 @@ private class MainLoop extends Thread {
      * Called when a frame-read operation times out
      * @throws MissedHeartbeatException if heart-beats have been missed
      */
-    private void handleSocketTimeout() throws MissedHeartbeatException {
+    private void handleSocketTimeout() throws SocketTimeoutException {
+        if (_inConnectionNegotiation) {
+            throw new SocketTimeoutException("Timeout during Connection negotiation");
+        }
+
         if (_heartbeat == 0) { // No heart-beating
             return;
         }

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

@@ -44,8 +44,8 @@
 public class AMQConnectionTest extends TestCase {
     // private static final String CLOSE_MESSAGE = "terminated by test";
 
-    /** 
-     * Build a suite of tests 
+    /**
+     * Build a suite of tests
      * @return the test suite for this class
      */
     public static TestSuite suite() {
@@ -79,7 +79,7 @@ public static TestSuite suite() {
     }
 
     /** Check the AMQConnection does send exactly 1 initial header, and deal correctly with
-     * the frame handler throwing an exception when we try to read data 
+     * the frame handler throwing an exception when we try to read data
      */
     public void testConnectionSendsSingleHeaderAndTimesOut() {
         IOException exception = new SocketTimeoutException();
@@ -100,14 +100,14 @@ public void testConnectionSendsSingleHeaderAndTimesOut() {
                     handler).start();
             fail("Connection should have thrown exception");
         } catch(IOException signal) {
-           // As expected 
+           // As expected
         }
         assertEquals(1, _mockFrameHandler.countHeadersSent());
         // _connection.close(0, CLOSE_MESSAGE);
         List<Throwable> exceptionList = handler.getHandledExceptions();
         assertEquals(Collections.<Throwable>singletonList(exception), exceptionList);
     }
-   
+
     /** Check we can open a connection once, but not twice.
      * @throws IOException */
 //    public void testCanOpenConnectionOnceOnly() throws IOException {
@@ -121,18 +121,49 @@ public void testConnectionSendsSingleHeaderAndTimesOut() {
 //        }
 //    }
 
-// add test that we time out if no initial Start command is received,
-// setting a timeout and having the FrameHandler return null
-    
+    /**
+     * Test that we catch timeout between connect and negotiation of the connection being finished.
+     */
+    public void testConnectionHangInNegotiation() {
+        this._mockFrameHandler.setTimeoutCount(10); // to limit hang
+        MyExceptionHandler handler = new MyExceptionHandler();
+        assertEquals(0, this._mockFrameHandler.countHeadersSent());
+        try {
+            new AMQConnection(factory.getUsername(),
+                    factory.getPassword(),
+                    this._mockFrameHandler,
+                    Executors.newFixedThreadPool(1),
+                    factory.getVirtualHost(),
+                    factory.getClientProperties(),
+                    factory.getRequestedFrameMax(),
+                    factory.getRequestedChannelMax(),
+                    factory.getRequestedHeartbeat(),
+                    factory.getSaslConfig(),
+                    handler).start();
+            fail("Connection should have thrown exception");
+        } catch(IOException signal) {
+           // As expected
+        }
+        assertEquals(1, this._mockFrameHandler.countHeadersSent());
+        // _connection.close(0, CLOSE_MESSAGE);
+        List<Throwable> exceptionList = handler.getHandledExceptions();
+        assertEquals("Only one exception expected", 1, exceptionList.size());
+        assertEquals("Wrong type of exception returned.", SocketTimeoutException.class, exceptionList.get(0).getClass());
+    }
+
     /** Mock frame handler to facilitate testing. */
     private static class MockFrameHandler implements FrameHandler {
         /** How many times has sendHeader() been called? */
         private int _numHeadersSent;
-        
+
+        private int timeout;
+
         /** An optional exception for us to throw on reading frames */
         private IOException _exceptionOnReadingFrames;
 
-        /** count how many headers we've sent 
+        private int timeoutCount = 0;
+
+        /** count how many headers we've sent
          * @return the number of sent headers
          */
         public int countHeadersSent() {
@@ -143,20 +174,27 @@ public void setExceptionOnReadingFrames(IOException exception) {
             _exceptionOnReadingFrames = exception;
         }
 
+        public void setTimeoutCount(int timeoutCount) {
+            this.timeoutCount = timeoutCount;
+        }
+
         public Frame readFrame() throws IOException {
             if (_exceptionOnReadingFrames != null) {
                 throw _exceptionOnReadingFrames;
             }
-            return null;
-            // throw new SocketTimeoutException(); // simulate a socket timeout
+            if (this.timeoutCount > 0) {
+                if (--this.timeoutCount == 0)
+                    throw new IOException("Mock Framehandler: too many timeouts.");
+            }
+            return null; // simulate a socket timeout
         }
 
         public void sendHeader() throws IOException {
-            _numHeadersSent++;            
+            _numHeadersSent++;
         }
 
         public void setTimeout(int timeoutMs) throws SocketException {
-            // no need to implement this: don't bother changing the timeout
+            this.timeout = timeoutMs;
         }
 
         public void writeFrame(Frame frame) throws IOException {
@@ -168,7 +206,7 @@ public void close() {
         }
 
         public int getTimeout() throws SocketException {
-            return 0;
+            return this.timeout;
         }
 
         public InetAddress getAddress() {
@@ -180,7 +218,7 @@ public int getPort() {
         }
     }
 
-    /** Mock frame handler to facilitate testing. */
+    /** Exception handler to facilitate testing. */
     private class MyExceptionHandler implements ExceptionHandler {
         private List<Throwable> _handledExceptions = new ArrayList<Throwable>();
 
@@ -208,7 +246,7 @@ public void handleConsumerException(Channel ch,
         {
             fail("handleConsumerException " + consumerTag + " " + methodName + ": " + ex);
         }
-        
+
         public List<Throwable> getHandledExceptions() {
             return _handledExceptions;
         }