Merged bug23722 into default

Author: Emile Joubert

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

@@ -245,10 +245,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

@@ -317,9 +317,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;
     }
 
@@ -594,11 +594,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

@@ -42,7 +42,6 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.TimeoutException;
 
 
@@ -231,27 +230,16 @@ public void releaseChannelNumber() {
         // incoming commands except for a close and close-ok.
 
         Method method = command.getMethod();
+        // we deal with channel.close in the same way, regardless
+        if (method instanceof Channel.Close) {
+            asyncShutdown(command);
+            return true;
+        }
 
         if (isOpen()) {
             // We're in normal running mode.
 
-            if (method instanceof Channel.Close) {
-                releaseChannelNumber();
-                ShutdownSignalException signal = new ShutdownSignalException(false,
-                                                                             false,
-                                                                             command,
-                                                                             this);
-                synchronized (_channelMutex) {
-                    try {
-                        processShutdownSignal(signal, true, false);
-                        quiescingTransmit(new Channel.CloseOk());
-                    } finally {
-                        notifyOutstandingRpc(signal);
-                    }
-                }
-                notifyListeners();
-                return true;
-            } else if (method instanceof Basic.Deliver) {
+            if (method instanceof Basic.Deliver) {
                 Basic.Deliver m = (Basic.Deliver) method;
 
                 Consumer callback = _consumers.get(m.consumerTag);
@@ -355,15 +343,9 @@ 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.
-                synchronized (_channelMutex) {
-                    quiescingTransmit(new Channel.CloseOk());
-                }
-                return true;
-            } else if (method instanceof Channel.CloseOk) {
+            if (method instanceof Channel.CloseOk) {
                 // We're quiescing, and we see a channel.close-ok:
                 // this is our signal to leave quiescing mode and
                 // finally shut down for good. Let it be handled as an
@@ -378,6 +360,23 @@ public void releaseChannelNumber() {
         }
     }
 
+    private void asyncShutdown(Command command) throws IOException {
+        releaseChannelNumber();
+        ShutdownSignalException signal = new ShutdownSignalException(false,
+                                                                     false,
+                                                                     command,
+                                                                     this);
+        synchronized (_channelMutex) {
+            try {
+                processShutdownSignal(signal, true, false);
+                quiescingTransmit(new Channel.CloseOk());
+            } finally {
+                notifyOutstandingRpc(signal);
+            }
+        }
+        notifyListeners();
+    }
+
     /** Public API - {@inheritDoc} */
     public void close()
         throws IOException
@@ -419,7 +418,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);
@@ -442,8 +441,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) {
@@ -891,4 +890,5 @@ public Channel.FlowOk getFlow() {
     public long getNextPublishSeqNo() {
         return nextPublishSeqNo;
     }
-}
+    
+}

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

@@ -241,13 +241,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();
     }
 
@@ -325,7 +324,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) {
         }
@@ -336,7 +335,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

@@ -26,53 +26,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

@@ -51,13 +51,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
      */
@@ -98,8 +98,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 {
@@ -127,7 +127,7 @@ public synchronized void set(T newValue) {
         }
         _value = newValue;
         _filled = true;
-        notify();
+        notifyAll();
     }
 
     /**
@@ -140,7 +140,6 @@ public synchronized boolean setIfUnset(T newValue) {
             return false;
         }
         set(newValue);
-        _filled = true;
         return true;
     }
 }