Fixed problems mentioned in QA.

Added notifyRpc argument to processShutdownSignal to enable
late notification in the server initiated shutdown.
Added _brokerInitiatedShutdown to identify this special case
and do not report false exception in the main loop.

Author: Hubert Plociniczak

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

@@ -215,22 +215,40 @@ 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) {
+    public void processShutdownSignal(ShutdownSignalException signal, boolean ignoreClosed, boolean notifyRpc) {
         try {
             synchronized (this) {
                 ensureIsOpen(); // invariant: we should never be shut down more than once per instance
                 _shutdownCause = signal;
             }
+        } catch (AlreadyClosedException ace) {
+            if (!ignoreClosed)
+                throw ace;
         } finally {
-            notifyOutstandingRpc(signal);
+            if (notifyRpc)
+                notifyOutstandingRpc(signal);
         }
     }
 
+    /**
+     * Protected API - respond, in the driver thread, to a {@link ShutdownSignalException}.
+     * The method behaves in the same way as processShutdownSignal(ShutdownSignalException, boolean, boolean)
+     * with the only difference that any continuing rpc is always notified of the shutdown signal.
+     * @param signal the signal to handle
+     */
+    public void processShutdownSignal(ShutdownSignalException signal, boolean ignoreClosed) {
+        processShutdownSignal(signal, ignoreClosed, true);
+    }
+    
     public void notifyOutstandingRpc(ShutdownSignalException signal) {
         RpcContinuation k = nextOutstandingRpc();
         if (k != null) {

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,15 +461,15 @@ public MainLoop() {
                     }
                 }
             } catch (EOFException ex) {
-                shutdown(ex, false, ex);
+                if (!_brokerInitiatedShutdown)
+                    shutdown(ex, false, ex, true);
             } catch (Throwable ex) {
                 _exceptionHandler.handleUnexpectedConnectionDriverException(AMQConnection.this,
                                                                             ex);
-                shutdown(ex, false, ex);
+                shutdown(ex, false, ex, true);
             } finally {
                 // Finally, shut down our underlying data connection.
                 _frameHandler.close();
-                
                 _appContinuation.set(null);
                 notifyListeners();
             }
@@ -551,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();
         }
 
@@ -575,6 +583,7 @@ public SocketCloseWait() {
                 _frameHandler.close();
             } finally {
                 _running = false;
+                _channel0.notifyOutstandingRpc(cause);
             }
         }
     }
@@ -583,12 +592,14 @@ 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)
     {
-        
         ShutdownSignalException sse = new ShutdownSignalException(true,initiatedByApplication,
                                                                   reason, this);
         sse.initCause(cause);
@@ -600,15 +611,10 @@ public void shutdown(Object reason,
         } catch (AlreadyClosedException ace) {
             if (initiatedByApplication)
                 throw ace;
-        } finally {
-            try {
-                _channel0.processShutdownSignal(sse);
-            } catch (AlreadyClosedException ace) {
-                if (initiatedByApplication)
-                    throw ace;
-            }
         }
+        _channel0.processShutdownSignal(sse, !initiatedByApplication, notifyRpc);
         _channelManager.handleSignal(sse);
+        return sse;
     }
 
     /**
@@ -687,7 +693,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,11 +68,7 @@ public void handleSignal(ShutdownSignalException signal) {
         }
         for (AMQChannel channel : channels) {
             disconnectChannel(channel.getChannelNumber());
-            try {
-                channel.processShutdownSignal(signal);
-            } catch (ShutdownSignalException sse) {
-                // Ignore
-            }
+            channel.processShutdownSignal(signal, true);
         }
     }
 

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

@@ -145,8 +145,8 @@ 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) {
+        super.processShutdownSignal(signal, ignoreClosed);
         broadcastShutdownSignal(signal);
     }
 
@@ -222,7 +222,7 @@ public void releaseChannelNumber() {
                                                                              command,
                                                                              this);
                 synchronized(this) {
-                    processShutdownSignal(signal);
+                    processShutdownSignal(signal, true);
                     quiescingTransmit(new Channel.CloseOk());
                 }
                 notifyListeners();
@@ -285,7 +285,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);
             quiescingRpc(reason, k);
         }