Merge bug18061 into default

Author: tonyg

src/com/rabbitmq/client/AlreadyClosedException.java

@@ -0,0 +1,12 @@
+package com.rabbitmq.client;
+
+/*
+ * Thrown when application tries to perform an action on connection/channel
+ * which was already closed
+ */
+public class AlreadyClosedException extends ShutdownSignalException {
+	public AlreadyClosedException(String s)
+	{
+		super(true, true, s);
+	}
+}

src/com/rabbitmq/client/Channel.java

@@ -56,7 +56,7 @@
  * 
  */
 
-public interface Channel {
+public interface Channel extends ShutdownNotifier{
     /**
      * Retrieve this channel's channel number.
      * @return the channel number

src/com/rabbitmq/client/Connection.java

@@ -49,7 +49,7 @@
  * Current implementations are thread-safe for code at the client API level,
  * and in fact thread-safe internally except for code within RPC calls.
  */
-public interface Connection { // rename to AMQPConnection later, this is a temporary name
+public interface Connection extends ShutdownNotifier { // rename to AMQPConnection later, this is a temporary name
     /**
      * Retrieve the host.
      * @return the hostname of the peer we're connected to.

src/com/rabbitmq/client/ShutdownListener.java

@@ -0,0 +1,9 @@
+package com.rabbitmq.client;
+
+import java.util.EventListener;
+
+public interface ShutdownListener extends EventListener {
+	
+	public void shutdownCompleted(ShutdownSignalException cause);
+
+}

src/com/rabbitmq/client/ShutdownNotifier.java

@@ -0,0 +1,49 @@
+package com.rabbitmq.client;
+
+/**
+ * Interface for components that are shutdown capable and
+ * that allow listeners to be added for shutdown signals
+ * 
+ * @see ShutdownListener
+ * @see ShutdownSignalException
+ */
+public interface ShutdownNotifier {
+    /**
+     * Add shutdown listener.
+     * If the component is already closed, handler is fired immediately
+     * 
+     * @param listener {@link ShutdownListener} to the component
+     */
+    public void addShutdownListener(ShutdownListener listener);
+    
+    /**
+     * Remove shutdown listener for the component.
+     * 
+     * @param listener {@link ShutdownListener} to be removed
+     */
+    public void removeShutdownListener(ShutdownListener listener);
+    
+    /**
+     * Get the shutdown reason object
+     * @return ShutdownSignalException if component is closed, null otherwise
+     */
+    public ShutdownSignalException getCloseReason();
+    
+    /**
+     * Protected API - notify the listeners attached to the component
+     * @see com.rabbitmq.client.ShutdownListener
+     */
+    public void notifyListeners();
+    
+    /**
+     * Determine whether the component is currently open.
+     * Will return false if we are currently closing.
+     * Checking this method should be only for information,
+     * because of the race conditions - state can change after the call.
+     * Instead just execute and try to catch ShutdownSignalException
+     * and IOException
+     * 
+     * @return true when component is open, false otherwise
+     */
+    boolean isOpen();
+}

src/com/rabbitmq/client/ShutdownSignalException.java

@@ -28,7 +28,7 @@
  * Encapsulates a shutdown condition for a connection to an AMQP broker.
  */
 
-public class ShutdownSignalException extends Exception {
+public class ShutdownSignalException extends RuntimeException {
     /** True if the connection is shut down, or false if this signal refers to a channel */
     private final boolean _hardError;
 
@@ -59,11 +59,13 @@ public ShutdownSignalException(boolean hardError,
 
     /** @return true if this signals a connection error, or false if a channel error */
     public boolean isHardError() { return _hardError; }
+    
     /** @return true if this exception was caused by explicit application
      * action; false if it originated with the broker or as a result
      * of detectable non-deliberate application failure
      */
     public boolean isInitiatedByApplication() { return _initiatedByApplication; }
+    
     /** @return the reason object, if any */
     public Object getReason() { return _reason; }
 

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

@@ -27,6 +27,7 @@
 
 import java.io.IOException;
 
+import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.Command;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ShutdownSignalException;
@@ -41,7 +42,7 @@
  * @see ChannelN
  * @see Connection
  */
-public abstract class AMQChannel {
+public abstract class AMQChannel extends ShutdownNotifierComponent {
     /** The connection this channel is associated with. */
     public final AMQConnection _connection;
 
@@ -54,9 +55,6 @@ public abstract class AMQChannel {
     /** The current outstanding RPC request, if any. (Could become a queue in future.) */
     public RpcContinuation _activeRpc = null;
 
-    /** Indicates whether this channel is in a state to handle further activity. */
-    public volatile boolean _isOpen = true;
-
     /**
      * Construct a channel on the given connection, with the given channel number.
      * @param connection the underlying connection for this channel
@@ -117,6 +115,10 @@ public AMQCommand exnWrappingRpc(Method m)
     {
         try {
             return rpc(m);
+        } catch (AlreadyClosedException ace) {
+        	// Do not wrap it since it means that connection/channel
+        	// was closed in some action in the past
+        	throw ace;
         } catch (ShutdownSignalException ex) {
             throw wrap(ex);
         }
@@ -160,21 +162,18 @@ public synchronized void transmitAndEnqueue(Method m, RpcContinuation k)
         transmit(m);
     }
 
-    public synchronized RpcContinuation nextOutstandingRpc() {
+    public synchronized RpcContinuation nextOutstandingRpc()
+    {
         RpcContinuation result = _activeRpc;
         _activeRpc = null;
         return result;
     }
 
-    public boolean isOpen() {
-        return _isOpen;
-    }
-
     public void ensureIsOpen()
-        throws IllegalStateException
+        throws AlreadyClosedException
     {
         if (!isOpen()) {
-            throw new IllegalStateException("Attempt to use closed channel");
+            throw new AlreadyClosedException("Attempt to use closed channel");
         }
     }
 
@@ -261,7 +260,7 @@ public void run() {
     public void processShutdownSignal(ShutdownSignalException signal) {
         synchronized (this) {
             ensureIsOpen(); // invariant: we should never be shut down more than once per instance
-            _isOpen = false;
+            _shutdownCause = signal;
         }
         RpcContinuation k = nextOutstandingRpc();
         if (k != null) {

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

@@ -32,6 +32,7 @@
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Address;
+import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Command;
 import com.rabbitmq.client.Connection;
@@ -59,7 +60,7 @@
  * int ticket = ch1.accessRequest(realmName);
  * </pre>
  */
-public class AMQConnection implements Connection {
+public class AMQConnection extends ShutdownNotifierComponent implements Connection{
     /** Timeout used while waiting for AMQP handshaking to complete (milliseconds) */
     public static final int HANDSHAKE_TIMEOUT = 10000;
 
@@ -88,19 +89,12 @@ public class AMQConnection implements Connection {
     /** Flag controlling the main driver loop's termination */
     public volatile boolean _running = false;
 
-    /**
-     * When this value is null, the connection is in an "open"
-     * state. When non-null, the connection is in "closed" state, and
-     * this value indicates the circumstances of the shutdown.
-     */
-    public volatile ShutdownSignalException _shutdownCause = null;
-
     /** Maximum frame length, or zero if no limit is set */
     public int _frameMax;
 
     /** Handler for (otherwise-unhandled) exceptions that crop up in the mainloop. */
     public final ExceptionHandler _exceptionHandler;
-
+    
     /**
      * Protected API - respond, in the driver thread, to a ShutdownSignal.
      * @param channelNumber the number of the channel to disconnect
@@ -109,15 +103,11 @@ public final void disconnectChannel(int channelNumber) {
         _channelManager.disconnectChannel(channelNumber);
     }
 
-    public boolean isOpen() {
-        return _shutdownCause == null;
-    }
-
     public void ensureIsOpen()
-        throws IllegalStateException
+        throws AlreadyClosedException
     {
         if (!isOpen()) {
-            throw new IllegalStateException("Attempt to use closed connection");
+            throw new AlreadyClosedException("Attempt to use closed connection");
         }
     }
 
@@ -340,15 +330,13 @@ public Address[] open(final ConnectionParameters params, boolean insist)
             _frameHandler.setTimeout(HANDSHAKE_TIMEOUT);
             _frameHandler.sendHeader();
 
-            if (!isOpen()) {
-                // See bug 17389. The MainLoop could have shut down already in
-                // which case we don't want to wait forever for a reply.
-                
-                // There is no race if the MainLoop shuts down after enqueuing
-                // the RPC because if that happens the channel will correctly
-                // pass the exception into RPC, waking it up.
-                throw _shutdownCause;
-            }
+            // See bug 17389. The MainLoop could have shut down already in
+            // which case we don't want to wait forever for a reply.
+            
+            // There is no race if the MainLoop shuts down after enqueuing
+            // the RPC because if that happens the channel will correctly
+            // pass the exception into RPC, waking it up.
+            ensureIsOpen();
 
             AMQP.Connection.Start connStart =
                 (AMQP.Connection.Start) connStartBlocker.getReply().getMethod();
@@ -561,6 +549,7 @@ public void handleConnectionClose(Command closeCommand) {
             Utility.emptyStatement();
         }
         shutdown(closeCommand, false, null);
+        notifyListeners();
     }
 
     /**
@@ -616,6 +605,7 @@ public void close(int closeCode,
         } finally {
             _running = false;
         }
+        notifyListeners();
     }
 
     @Override public String toString() {