Added channel.close response processing even if !isOpen()

o  copied handling code in ChannelN.processAsync() to !isOpen() case
o  improved atomicity of operations in shutdown processing
o  spelling and comment adjustments

Author: Steve Powell

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

@@ -259,10 +259,10 @@ public void processShutdownSignal(ShutdownSignalException signal,
                                       boolean notifyRpc) {
         try {
             synchronized (_channelMutex) {
-                if (!ignoreClosed)
-                    ensureIsOpen(); // invariant: we should never be shut down more than once per instance
-                if (isOpen())
-                    _shutdownCause = signal;
+                if (!setShutdownCauseIfOpen(signal)) {
+                    if (!ignoreClosed) 
+                        throw new AlreadyClosedException("Attempt to use closed channel", this);
+                }
 
                 _channelMutex.notifyAll();
             }

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

@@ -333,9 +333,9 @@ public void start()
                                                          heartbeat));
         // 0.9.1: insist [on not being redirected] is deprecated, but
         // still in generated code; just pass a dummy value here
-        Method res = _channel0.exnWrappingRpc(new AMQImpl.Connection.Open(_virtualHost,
-                                                                          "",
-                                                                          false)).getMethod();
+        _channel0.exnWrappingRpc(new AMQImpl.Connection.Open(_virtualHost,
+                                                            "",
+                                                            false)).getMethod();
         return;
     }
 
@@ -610,11 +610,9 @@ public ShutdownSignalException shutdown(Object reason,
         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
-            if (isOpen())
-                _shutdownCause = sse;
+        if (!setShutdownCauseIfOpen(sse)) {
+            if (initiatedByApplication) 
+                throw new AlreadyClosedException("Attempt to use closed connection", this);
         }
 
         // stop any heartbeating

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

@@ -56,7 +56,6 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.TimeoutException;
 
 
@@ -256,12 +255,8 @@ public void releaseChannelNumber() {
                                                                              command,
                                                                              this);
                 synchronized (_channelMutex) {
-                    try {
-                        processShutdownSignal(signal, true, false);
-                        quiescingTransmit(new Channel.CloseOk());
-                    } finally {
-                        notifyOutstandingRpc(signal);
-                    }
+                    processShutdownSignal(signal, true, true);
+                    quiescingTransmit(new Channel.CloseOk());
                 }
                 notifyListeners();
                 return true;
@@ -358,13 +353,20 @@ public void releaseChannelNumber() {
                 return false;
             }
         } else {
-            // We're in quiescing mode.
+            // We're in quiescing mode == !isOpen()
 
             if (method instanceof Channel.Close) {
-                // We're already shutting down, so just send back an ok.
+                // We are already shutting down, but we cannot assume no Rpc is waiting.
+                releaseChannelNumber();
+                ShutdownSignalException signal = new ShutdownSignalException(false,
+                                                                             false,
+                                                                             command,
+                                                                             this);
                 synchronized (_channelMutex) {
+                    processShutdownSignal(signal, true, true);
                     quiescingTransmit(new Channel.CloseOk());
                 }
+                notifyListeners();
                 return true;
             } else if (method instanceof Channel.CloseOk) {
                 // We're quiescing, and we see a channel.close-ok:
@@ -422,7 +424,7 @@ public void close(int closeCode,
         throws IOException
     {
         // First, notify all our dependents that we are shutting down.
-        // This clears _isOpen, so no further work from the
+        // This clears isOpen(), so no further work from the
         // application side will be accepted, and any inbound commands
         // will be discarded (unless they're channel.close-oks).
         Channel.Close reason = new Channel.Close(closeCode, closeMessage, 0, 0);
@@ -445,8 +447,8 @@ public void close(int closeCode,
             }
 
             // Now that we're in quiescing state, channel.close was sent and
-            // we wait for the reply. We ignore the result. (It's always
-            // close-ok.)
+            // we wait for the reply. We ignore the result.
+            // (It's NOT always close-ok.)
             notify = true;
             k.getReply(-1);
         } catch (TimeoutException ise) {

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

@@ -255,13 +255,12 @@ public DataOutputStream getOutputStream() {
 
     @Override public String toString() {
         StringBuffer sb = new StringBuffer();
-        sb.append("Frame(" + type + ", " + channel + ", ");
+        sb.append("Frame(type=").append(type).append(", channel=").append(channel).append(", ");
         if (accumulator == null) {
-            sb.append(payload.length + " bytes of payload");
+            sb.append(payload.length).append(" bytes of payload)");
         } else {
-            sb.append(accumulator.size() + " bytes of accumulator");
+            sb.append(accumulator.size()).append(" bytes of accumulator)");
         }
-        sb.append(")");
         return sb.toString();
     }
 
@@ -339,7 +338,7 @@ else if(value instanceof byte[]) {
             acc += 4 + ((byte[])value).length;
         }
         else if(value instanceof List) {
-            acc += 4 + arraySize((List)value);
+            acc += 4 + arraySize((List<?>)value);
         }
         else if(value == null) {
         }
@@ -350,7 +349,7 @@ else if(value == null) {
     }
 
     /** Computes the AMQP wire-protocol length of an encoded field-array */
-    public static long arraySize(List values)
+    public static long arraySize(List<?> values)
         throws UnsupportedEncodingException
     {
         long acc = 0;

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

@@ -40,53 +40,71 @@
 
 public class ShutdownNotifierComponent implements ShutdownNotifier {
 
+    /** Monitor for listeners and shutdownCause */
+    private final Object monitor = new Object();
+
     /** List of all shutdown listeners associated with the component */
-    public List<ShutdownListener> listeners
-            = new ArrayList<ShutdownListener>();
+    private final List<ShutdownListener> shutdownListeners
+        = new ArrayList<ShutdownListener>();
 
     /**
      * When this value is null, the component is in an "open"
      * state. When non-null, the component is in "closed" state, and
      * this value indicates the circumstances of the shutdown.
      */
-    public volatile ShutdownSignalException _shutdownCause = null;
+    private volatile ShutdownSignalException shutdownCause = null;
 
     public void addShutdownListener(ShutdownListener listener)
     {
-        boolean closed = false;
-        synchronized(listeners) {
-            closed = !isOpen();
-            listeners.add(listener);
+        ShutdownSignalException sse = null;
+        synchronized(this.monitor) {
+            sse = this.shutdownCause;
+            this.shutdownListeners.add(listener);
         }
-        if (closed)
-            listener.shutdownCompleted(getCloseReason());
+        if (sse != null) // closed
+            listener.shutdownCompleted(sse);
     }
 
     public ShutdownSignalException getCloseReason() {
-        return _shutdownCause;
+        return this.shutdownCause;
     }
 
     public void notifyListeners()
     {
-        synchronized(listeners) {
-            for (ShutdownListener l: listeners)
-                try {
-                    l.shutdownCompleted(getCloseReason());
-                } catch (Exception e) {
-                    // FIXME: proper logging
-                }
+        ShutdownSignalException sse = null;
+        ShutdownListener[] sdls = null;
+        synchronized(this.monitor) {
+            sdls = this.shutdownListeners
+                .toArray(new ShutdownListener[this.shutdownListeners.size()]);
+            sse = this.shutdownCause;
+        }
+        for (ShutdownListener l: sdls) {
+            try {
+                l.shutdownCompleted(sse);
+            } catch (Exception e) {
+            // FIXME: proper logging
+            }
         }
     }
 
     public void removeShutdownListener(ShutdownListener listener)
     {
-        synchronized(listeners) {
-            listeners.remove(listener);
+        synchronized(this.monitor) {
+            this.shutdownListeners.remove(listener);
         }
     }
 
     public boolean isOpen() {
-        return _shutdownCause == null;
+        return this.shutdownCause == null;
     }
 
+    public boolean setShutdownCauseIfOpen(ShutdownSignalException sse) {
+        synchronized (this.monitor) {
+            if (isOpen()) {
+                this.shutdownCause = sse;
+                return true;
+            }
+            return false;
+        }
+    }
 }

src/com/rabbitmq/utility/BlockingCell.java

@@ -65,13 +65,13 @@ public synchronized T get() throws InterruptedException {
         }
         return _value;
     }
-    
+
     /**
      * Wait for a value, and when one arrives, return it (without clearing it). If there's
      * already a value present, there's no need to wait - the existing value is returned.
-     * If timeout is reached and value hasn't arrived, TimeoutException is thrown
+     * If timeout is reached and value hasn't arrived, TimeoutException is thrown.
      * 
-     * @param timeout timeout in miliseconds. -1 effectively means infinity
+     * @param timeout timeout in milliseconds. -1 effectively means infinity
      * @return the waited-for value
      * @throws InterruptedException if this thread is interrupted
      */
@@ -112,8 +112,8 @@ public synchronized T uninterruptibleGet() {
      * a value appears or until specified timeout is reached. If timeout is reached,
      * TimeoutException it thrown.
      * We also use System.nanoTime() to behave correctly when system clock jumps around.
-     *  
-     * @param timeout timeout in miliseconds. -1 effectively means infinity
+     * 
+     * @param timeout timeout in milliseconds. -1 effectively means infinity
      * @return the waited-for value
      */
     public synchronized T uninterruptibleGet(int timeout) throws TimeoutException {
@@ -141,7 +141,7 @@ public synchronized void set(T newValue) {
         }
         _value = newValue;
         _filled = true;
-        notify();
+        notifyAll();
     }
 
     /**
@@ -154,7 +154,6 @@ public synchronized boolean setIfUnset(T newValue) {
             return false;
         }
         set(newValue);
-        _filled = true;
         return true;
     }
 }