Migrate branch bug16247

Author: tonyg

src/com/rabbitmq/client/AlreadyClosedException.java

@@ -0,0 +1,19 @@
+package com.rabbitmq.client;
+
+public class AlreadyClosedException extends IllegalStateException {
+	
+	public AlreadyClosedException()
+	{
+		super();
+	}
+	
+	public AlreadyClosedException(Throwable e)
+	{
+		super(e);
+	}
+
+	public AlreadyClosedException(String s)
+	{
+		super(s);
+	}
+}

src/com/rabbitmq/client/Channel.java

@@ -433,4 +433,38 @@ Queue.DeclareOk queueDeclare(int ticket, String queue, boolean passive, boolean
      * @throws java.io.IOException if an error is encountered
      */
     Tx.RollbackOk txRollback() throws IOException;
+
+    /**
+     * Add shutdown listener to the channel
+     * 
+     * @param listener {@link ShutdownListener} for the channel
+     */
+    void addShutdownListener(ShutdownListener listener);
+    
+    /**
+     * Remove shutdown listener for the channel.
+     * 
+     * @param listener {@link ShutdownListener} to be removed
+     */
+    void removeShutdownListener(ShutdownListener listener);
+    
+    /**
+     * Get connection channel shutdown reason.
+     * Return null if channel is still open.
+     * @see com.rabbitmq.client.ShutdownCause 
+     * @return shutdown reason if channel is closed
+     */
+    ShutdownSignalException getCloseReason();
+    
+    /**
+     * Determine if channel is currently open.
+     * Will return false if we are currently closing or closed.
+     * Checking this method should be only for information,
+     * because of the race conditions - state can change after the call.
+     * Instead just execute and and try to catch AlreadyClosedException
+     * 
+     * @see com.rabbitmq.client.impl.AMQChannel#isOpen()
+     * @return true when channel is open, false otherwise
+     */
+    boolean isOpen();
 }

src/com/rabbitmq/client/Connection.java

@@ -127,4 +127,39 @@ public interface Connection { // rename to AMQPConnection later, this is a tempo
      * @throws IOException if an I/O problem is encountered
      */
     void close(int closeCode, String closeMessage) throws IOException;
+    
+    /**
+     * Add connection shutdown listener.
+     * If the connection is already closed handler is fired immediately
+     * 
+     * @param listener {@link ShutdownListener} to the connection
+     */
+    void addShutdownListener(ShutdownListener listener);
+    
+    /**
+     * Remove shutdown listener for the connection.
+     * 
+     * @param listener {@link ShutdownListener} to be removed
+     */
+    void removeShutdownListener(ShutdownListener listener);
+    
+    /**
+     * Retrieve connection close reason.
+     * 
+     * @see com.rabbitmq.client.ShutdownCause
+     * @return information about the cause of closing the connection, or null if connection is still open
+     */
+    ShutdownSignalException getCloseReason();
+    
+    /**
+     * Determine whether the connection 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 and try to catch AlreadyClosedException
+     * 
+     * @see com.rabbitmq.client.impl.AMQConnection#isOpen()
+     * @return true when connection is open, false otherwise
+     */
+    boolean isOpen();
 }

src/com/rabbitmq/client/RpcClient.java

@@ -190,7 +190,8 @@ public byte[] primitiveCall(AMQP.BasicProperties props, byte[] message)
             ShutdownSignalException wrapper =
                 new ShutdownSignalException(sig.isHardError(),
                                             sig.isInitiatedByApplication(),
-                                            sig.getReason());
+                                            sig.getReason(),
+                                            sig.getReference());
             wrapper.initCause(sig);
             throw wrapper;
         } else {

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 service(ShutdownSignalException cause);
+
+}

src/com/rabbitmq/client/ShutdownSignalException.java

@@ -26,6 +26,10 @@
 
 /**
  * Encapsulates a shutdown condition for a connection to an AMQP broker.
+ * Depending on HardError when calling
+ * {@link com.rabbitmq.client.ShutdownSignalException#getReference()} we will
+ * either get a reference to the Connection or Channel instance that fired
+ * this exception.
  */
 
 public class ShutdownSignalException extends Exception {
@@ -42,30 +46,41 @@ public class ShutdownSignalException extends Exception {
     /** Possible explanation */
     private final Object _reason;
 
+    /** Either Channel or Connection instance, depending on _hardError */
+    private final Object _ref;
+
     /**
      * Construct a ShutdownSignalException from the arguments.
      * @param hardError the relevant hard error
      * @param initiatedByApplication if the shutdown was client-initiated
      * @param reason Object describing the origin of the exception
+     * @param ref Reference to Connection or Channel that fired the signal
      */
     public ShutdownSignalException(boolean hardError,   
                                    boolean initiatedByApplication,
-                                   Object reason)
+                                   Object reason, Object ref)
     {
         this._hardError = hardError;
         this._initiatedByApplication = initiatedByApplication;
         this._reason = reason;
+        // Depending on hardError what we got is either Connection or Channel reference
+        this._ref = ref;
     }
 
     /** @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; }
+    
+    /** @return Reference to Connection or Channel object that fired the signal **/
+    public Object getReference() { return _ref; }
 
     public String toString() {
         return super.toString() + " (" +

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

@@ -27,8 +27,10 @@
 
 import java.io.IOException;
 
+import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.Command;
 import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ShutdownListener;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.utility.BlockingValueOrException;
 import com.rabbitmq.utility.SingleShotLinearTimer;
@@ -53,9 +55,9 @@ 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;
+    
+    /** Reason for closing the channel, null if still open */
+    public volatile ShutdownSignalException _cause;
 
     /**
      * Construct a channel on the given connection, with the given channel number.
@@ -160,21 +162,36 @@ 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 API - Indicates whether this channel is in an open state
+     * @return true if channel is open, false otherwise
+     */
+    public boolean isOpen()
+    {
+        return _cause == null;
+    }
+    
+    /**
+     * Public API - Get the reason for closing the channel 
+     * @return object having information about the shutdown, or null if still open
+     */
+    public ShutdownSignalException getCloseReason() 
+    {
+    	return _cause;
     }
 
     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 +278,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;
+            _cause = signal;
         }
         RpcContinuation k = nextOutstandingRpc();
         if (k != null) {

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

@@ -28,16 +28,21 @@
 import java.io.EOFException;
 import java.io.IOException;
 import java.net.SocketException;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 
 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;
 import com.rabbitmq.client.ConnectionParameters;
 import com.rabbitmq.client.MissedHeartbeatException;
 import com.rabbitmq.client.RedirectException;
+import com.rabbitmq.client.ShutdownListener;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.utility.Utility;
 
@@ -100,7 +105,11 @@ public class AMQConnection implements Connection {
 
     /** Handler for (otherwise-unhandled) exceptions that crop up in the mainloop. */
     public final ExceptionHandler _exceptionHandler;
-
+    
+    /** List of all shutdown listeners associated with the connection */
+    public List<ShutdownListener> listeners
+            = Collections.synchronizedList(new LinkedList<ShutdownListener>());
+    
     /**
      * Protected API - respond, in the driver thread, to a ShutdownSignal.
      * @param channelNumber the number of the channel to disconnect
@@ -109,15 +118,19 @@ public final void disconnectChannel(int channelNumber) {
         _channelManager.disconnectChannel(channelNumber);
     }
 
+    /**
+     * Public API - Determine whether the connection is open
+     * @return true if haven't yet received shutdown signal, false otherwise
+     */
     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");
         }
     }
 
@@ -576,7 +589,7 @@ public void shutdown(Object reason,
             ensureIsOpen(); // invariant: we should never be shut down more than once per instance
             _shutdownCause = new ShutdownSignalException(true,
                                                          initiatedByApplication,
-                                                         reason);
+                                                         reason, this);
         }
         if (cause != null) {
             _shutdownCause.initCause(cause);
@@ -616,6 +629,48 @@ public void close(int closeCode,
         } finally {
             _running = false;
         }
+        
+        synchronized(listeners) {
+        	for (ShutdownListener l: listeners)
+        		l.service(getCloseReason());
+        }
+    }
+    
+    /**
+     * Public API - Add shutdown listener fired when closing the connection
+     * @see com.rabbitmq.client.Connection#addShutdownListener()
+     */
+    public void addShutdownListener(ShutdownListener listener)
+    {
+    	
+    	boolean closed = false;
+    	synchronized(listeners) {
+    		closed = !isOpen();
+    		listeners.add(listener);
+    	}
+    	if (closed)
+    		listener.service(_shutdownCause);	
+    }
+    
+    /**
+     * Public API - Remove shutdown listener for this connection
+     * Removing only the first found object
+     * @see com.rabbitmq.client.Connection#removeShutdownListener()
+     */
+    public void removeShutdownListener(ShutdownListener listener)
+    {
+    	synchronized(listeners) {
+    		listeners.remove(listener);
+    	}
+    }
+    
+    /**
+     * Public API - Get reason for shutdown, or null if open
+     * @see com.rabbitmq.client.Connection#getShutdownReason()
+     */
+    public ShutdownSignalException getCloseReason()
+    {
+    	return _shutdownCause;
     }
 
     @Override public String toString() {