merge bug19487 into default

Author: Matthias Radestock

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

@@ -215,17 +215,29 @@ public synchronized void quiescingRpc(Method m, RpcContinuation k)
     @Override public String toString() {
         return "AMQChannel(" + _connection + "," + _channelNumber + ")";
     }
-
+    
     /**
      * Protected API - respond, in the driver thread, to a {@link ShutdownSignalException}.
      * @param signal the signal to handle
+     * @param ignoreClosed the flag indicating whether to ignore the AlreadyClosedException
+     *                     thrown when the channel is already closed
+     * @param notifyRpc the flag indicating whether any remaining rpc continuation should be
+     *                  notified with the given signal
      */
-    public void processShutdownSignal(ShutdownSignalException signal) {
-        synchronized (this) {
-            ensureIsOpen(); // invariant: we should never be shut down more than once per instance
-            _shutdownCause = signal;
+    public void processShutdownSignal(ShutdownSignalException signal,
+                                      boolean ignoreClosed,
+                                      boolean notifyRpc) {
+        try {
+            synchronized (this) {
+                if (!ignoreClosed)
+                    ensureIsOpen(); // invariant: we should never be shut down more than once per instance
+                if (isOpen())
+                    _shutdownCause = signal;
+            }
+        } finally {
+            if (notifyRpc)
+                notifyOutstandingRpc(signal);
         }
-        notifyOutstandingRpc(signal);
     }
 
     public void notifyOutstandingRpc(ShutdownSignalException signal) {

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

@@ -103,6 +103,9 @@ public class AMQConnection extends ShutdownNotifierComponent implements Connecti
      */
     public BlockingCell<Object> _appContinuation = new BlockingCell<Object>();
 
+    /** Flag indicating whether the client received Connection.Close message from the broker */
+    public boolean _brokerInitiatedShutdown = false;
+    
     /**
      * Protected API - respond, in the driver thread, to a ShutdownSignal.
      * @param channelNumber the number of the channel to disconnect
@@ -191,6 +194,7 @@ public AMQConnection(ConnectionParameters params,
         _missedHeartbeats = 0;
         _heartbeat = 0;
         _exceptionHandler = exceptionHandler;
+        _brokerInitiatedShutdown = false;
 
         new MainLoop(); // start the main loop going
 
@@ -457,26 +461,15 @@ public MainLoop() {
                     }
                 }
             } catch (EOFException ex) {
-                if (isOpen()) {
-                    System.err.println("AMQConnection.mainLoop: connection close");
-                    shutdown(ex, false, ex);
-                }
+                if (!_brokerInitiatedShutdown)
+                    shutdown(ex, false, ex, true);
             } catch (Throwable ex) {
                 _exceptionHandler.handleUnexpectedConnectionDriverException(AMQConnection.this,
                                                                             ex);
-                if (isOpen()) {
-                    shutdown(ex, false, ex);
-                }
+                shutdown(ex, false, ex, true);
             } finally {
                 // Finally, shut down our underlying data connection.
                 _frameHandler.close();
-                
-                // Set shutdown exception for any outstanding rpc,
-                // so that it does not wait infinitely for Connection.CloseOk.
-                // This can only happen when the broker closed the socket
-                // unexpectedly.
-                _channel0.notifyOutstandingRpc(_shutdownCause);
-                
                 _appContinuation.set(null);
                 notifyListeners();
             }
@@ -562,18 +555,22 @@ public boolean processControlCommand(Command c)
     }
 
     public void handleConnectionClose(Command closeCommand) {
-        shutdown(closeCommand, false, null);
+        ShutdownSignalException sse = shutdown(closeCommand, false, null, false);
         try {
             _channel0.quiescingTransmit(new AMQImpl.Connection.CloseOk());
         } catch (IOException ioe) {
             Utility.emptyStatement();
         }
-        _heartbeat = 0; // Do not try to send heartbeats after CloseOk   
-        new SocketCloseWait();
+        _heartbeat = 0; // Do not try to send heartbeats after CloseOk
+        _brokerInitiatedShutdown = true;
+        new SocketCloseWait(sse);
     }
 
     private class SocketCloseWait extends Thread {
-        public SocketCloseWait() {
+        private ShutdownSignalException cause;
+        
+        public SocketCloseWait(ShutdownSignalException sse) {
+            cause = sse;
             start();
         }
 
@@ -586,6 +583,7 @@ public SocketCloseWait() {
                 _frameHandler.close();
             } finally {
                 _running = false;
+                _channel0.notifyOutstandingRpc(cause);
             }
         }
     }
@@ -594,27 +592,26 @@ public SocketCloseWait() {
      * 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.
+     * 
+     * @return a shutdown signal built using the given arguments
      */
-    public void shutdown(Object reason,
+    public ShutdownSignalException shutdown(Object reason,
                          boolean initiatedByApplication,
-                         Throwable cause)
+                         Throwable cause,
+                         boolean notifyRpc)
     {
-        try {
-            synchronized (this) {
+        ShutdownSignalException sse = new ShutdownSignalException(true,initiatedByApplication,
+                                                                  reason, this);
+        sse.initCause(cause);
+        synchronized (this) {
+            if (initiatedByApplication)
                 ensureIsOpen(); // invariant: we should never be shut down more than once per instance
-                ShutdownSignalException sse = new ShutdownSignalException(true,
-                                                             initiatedByApplication,
-                                                             reason, this);
-                sse.initCause(cause);
+            if (isOpen())
                 _shutdownCause = sse;
-            }
-
-            _channel0.processShutdownSignal(_shutdownCause);
-        } catch (AlreadyClosedException ace) {
-            if (initiatedByApplication)
-                throw ace;
         }
-        _channelManager.handleSignal(_shutdownCause);
+        _channel0.processShutdownSignal(sse, !initiatedByApplication, notifyRpc);
+        _channelManager.handleSignal(sse);
+        return sse;
     }
 
     public void close()
@@ -690,7 +687,7 @@ public void close(int closeCode,
         try {
             AMQImpl.Connection.Close reason =
                 new AMQImpl.Connection.Close(closeCode, closeMessage, 0, 0);
-            shutdown(reason, initiatedByApplication, cause);
+            shutdown(reason, initiatedByApplication, cause, true);
             AMQChannel.SimpleBlockingRpcContinuation k =
                 new AMQChannel.SimpleBlockingRpcContinuation();
             _channel0.quiescingRpc(reason, k);

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

@@ -68,12 +68,7 @@ public void handleSignal(ShutdownSignalException signal) {
         }
         for (AMQChannel channel : channels) {
             disconnectChannel(channel.getChannelNumber());
-            try {
-                channel.processShutdownSignal(signal);
-            } catch (ShutdownSignalException sse) {
-                // Notify channels that are waiting for rpc
-                channel.notifyOutstandingRpc(signal);
-            }
+            channel.processShutdownSignal(signal, true, true);
         }
     }
 

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

@@ -146,8 +146,11 @@ public void broadcastShutdownSignal(ShutdownSignalException signal) {
      * Protected API - overridden to broadcast the signal to all
      * consumers before calling the superclass's method.
      */
-    @Override public void processShutdownSignal(ShutdownSignalException signal) {
-        super.processShutdownSignal(signal);
+    @Override public void processShutdownSignal(ShutdownSignalException signal,
+                                                boolean ignoreClosed,
+                                                boolean notifyRpc)
+    {
+        super.processShutdownSignal(signal, ignoreClosed, notifyRpc);
         broadcastShutdownSignal(signal);
     }
 
@@ -223,7 +226,7 @@ public void releaseChannelNumber() {
                                                                              command,
                                                                              this);
                 synchronized(this) {
-                    processShutdownSignal(signal);
+                    processShutdownSignal(signal, true, true);
                     quiescingTransmit(new Channel.CloseOk());
                 }
                 notifyListeners();
@@ -289,7 +292,7 @@ public void close(int closeCode,
         // Synchronize the block below to avoid race conditions in case
         // connnection wants to send Connection-CloseOK
         synchronized(this) {
-            processShutdownSignal(signal);
+            processShutdownSignal(signal, !initiatedByApplication, true);
             quiescingRpc(reason, k);
         }