merge default into bug20337

Author: Alexandru Scvortov

README-EXAMPLES

@@ -63,7 +63,7 @@ HelloServer - Acts as an RPC server over AMQP.
 HelloJsonClient - Performs a simple JSON-RPC call over AMQP.
 HelloJsonServer - Acts as a JSON-RPC server over AMQP.
 LogTail - Tails the server logs.
-SendString - Sends a user supplied message over AMQP.
+SendString - Sends a user-supplied message over AMQP.
 
 
 More Complex Examples

src/com/rabbitmq/client/ShutdownListener.java

@@ -19,8 +19,14 @@
 
 import java.util.EventListener;
 
+/**
+ * A ShutdownListener receives information about the shutdown of connections and
+ * channels. Note that when a connection is shut down, its associated channels are also
+ * considered shut down and their ShutdownListeners will be notified (with the same cause).
+ *
+ * @see ShutdownNotifier
+ * @see ShutdownSignalException
+ */
 public interface ShutdownListener extends EventListener {
-
     public void shutdownCompleted(ShutdownSignalException cause);
-
 }

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;
 
@@ -175,7 +179,7 @@ public Map<String, Object> getServerProperties() {
      * @param password for <code><b>username</b></code>
      * @param frameHandler for sending and receiving frames on this connection
      * @param executor thread pool service for consumer threads for channels on this connection
-     * @param virtualHost
+     * @param virtualHost virtual host of this connection
      * @param clientProperties client info used in negotiating with the server
      * @param requestedFrameMax max size of frame offered
      * @param requestedChannelMax max number of channels offered
@@ -211,7 +215,7 @@ public AMQConnection(String username,
      * @param password for <code><b>username</b></code>
      * @param frameHandler for sending and receiving frames on this connection
      * @param executor thread pool service for consumer threads for channels on this connection
-     * @param virtualHost
+     * @param virtualHost virtual host of this connection
      * @param clientProperties client info used in negotiating with the server
      * @param requestedFrameMax max size of frame offered
      * @param requestedChannelMax max number of channels offered
@@ -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;
         }
@@ -623,10 +636,14 @@ public SocketCloseWait(ShutdownSignalException sse) {
     }
 
     /**
-     * Protected API - causes all attached channels to terminate with
-     * a ShutdownSignal built from the argument, and stops this
-     * connection from accepting further work from the application.
-     *
+     * Protected API - causes all attached channels to terminate (shutdown) with a ShutdownSignal
+     * built from the argument, and stops this connection from accepting further work from the
+     * application. {@link com.rabbitmq.client.ShutdownListener ShutdownListener}s for the
+     * connection are notified when the main loop terminates.
+     * @param reason object being shutdown
+     * @param initiatedByApplication true if caused by a client command
+     * @param cause trigger exception which caused shutdown
+     * @param notifyRpc true if outstanding rpc should be informed of shutdown
      * @return a shutdown signal built using the given arguments
      */
     public ShutdownSignalException shutdown(Object reason,
@@ -722,6 +739,7 @@ public void close(int closeCode,
         close(closeCode, closeMessage, initiatedByApplication, cause, -1, false);
     }
 
+    // TODO: Make this private
     /**
      * Protected API - Close this connection with the given code, message, source
      * and timeout value for all the close operations to complete.

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

@@ -30,7 +30,6 @@
 /**
  * Manages a set of channels, indexed by channel number (<code><b>1.._channelMax</b></code>).
  */
-
 public final class ChannelManager {
     private static final int SHUTDOWN_TIMEOUT_SECONDS = 10;
 
@@ -77,6 +76,10 @@ public ChannelN getChannel(int channelNumber) {
         }
     }
 
+    /**
+     * Handle shutdown. All the managed {@link com.rabbitmq.client.Channel Channel}s are shutdown.
+     * @param signal reason for shutdown
+     */
     public void handleSignal(ShutdownSignalException signal) {
         Set<ChannelN> channels;
         synchronized(this.monitor) {
@@ -86,6 +89,7 @@ public void handleSignal(ShutdownSignalException signal) {
             releaseChannelNumber(channel);
             channel.processShutdownSignal(signal, true, true);
             shutdownSet.add(channel.getShutdownLatch());
+            channel.notifyListeners();
         }
         scheduleShutdownProcessing();
     }

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

@@ -24,6 +24,11 @@
 import com.rabbitmq.client.ShutdownNotifier;
 import com.rabbitmq.client.ShutdownSignalException;
 
+/**
+ * A class that manages {@link ShutdownListener}s and remembers the reason for a shutdown. Both
+ * {@link com.rabbitmq.client.Channel Channel}s and {@link com.rabbitmq.client.Connection
+ * Connection}s have shutdown listeners.
+ */
 public class ShutdownNotifierComponent implements ShutdownNotifier {
 
     /** Monitor for shutdown listeners and shutdownCause */
@@ -87,6 +92,11 @@ public boolean isOpen() {
         }
     }
 
+    /**
+     * Internal: this is the means of registering shutdown.
+     * @param sse the reason for the shutdown
+     * @return <code>true</code> if the component is open; <code>false</code> otherwise.
+     */
     public boolean setShutdownCauseIfOpen(ShutdownSignalException sse) {
         synchronized (this.monitor) {
             if (isOpen()) {

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;
         }

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

@@ -32,7 +32,7 @@ public class CloseInMainLoop extends BrokerTestCase{
 
   class SpecialConnection extends AMQConnection{
     private AtomicBoolean validShutdown = new AtomicBoolean(false);
-    
+
     public boolean hadValidShutdown(){
       if(isOpen()) throw new IllegalStateException("hadValidShutdown called while connection is still open");
       return validShutdown.get();
@@ -61,6 +61,7 @@ public void handleConsumerException(Channel channel,
                                                     String consumerTag,
                                                     String methodName) {
                     try {
+                        // TODO: change this to call 4-parameter close and make 6-parm one private
                       ((AMQConnection) channel.getConnection())
                           .close(AMQP.INTERNAL_ERROR,
                                  "Internal error in Consumer " + consumerTag,

test/src/com/rabbitmq/client/test/functional/UnexpectedFrames.java

@@ -145,12 +145,6 @@ private void expectUnexpectedFrameError(Confuser confuser)
         //NB: the frame confuser relies on the encoding of the
         //method field to be at least 8 bytes long
         channel.basicPublish("", "routing key", null, "Hello".getBytes());
-        // TODO remove when bug 24086 is fixed.
-        try {
-            Thread.sleep(100);
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
         expectError(AMQP.UNEXPECTED_FRAME);
     }