Merge default into bug18743

Author: Hubert Plociniczak

.hgignore

@@ -0,0 +1,5 @@
+syntax: glob
+*~
+
+syntax: regexp
+^build/

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
@@ -433,38 +433,4 @@ 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

@@ -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.
@@ -157,39 +157,4 @@ public interface Connection { // rename to AMQPConnection later, this is a tempo
      * complete. If timeout is reached socket is forced to close.
      */
     void abort(int timeout);
-
-    /**
-     * 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/ShutdownListener.java

@@ -4,6 +4,6 @@
 
 public interface ShutdownListener extends EventListener {
 
-	public void service(ShutdownSignalException cause);
+	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/impl/AMQChannel.java

@@ -42,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 +54,6 @@ public abstract class AMQChannel {
 
     /** The current outstanding RPC request, if any. (Could become a queue in future.) */
     public RpcContinuation _activeRpc = null;
-    
-    /** 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.
@@ -172,24 +169,6 @@ public synchronized RpcContinuation nextOutstandingRpc()
         return result;
     }
 
-    /**
-     * 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 AlreadyClosedException
     {
@@ -250,15 +229,15 @@ public AMQCommand quiescingRpc(Method m,
     @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
      */
     public void processShutdownSignal(ShutdownSignalException signal) {
         synchronized (this) {
             ensureIsOpen(); // invariant: we should never be shut down more than once per instance
-            _cause = signal;
+            _shutdownCause = signal;
         }
         RpcContinuation k = nextOutstandingRpc();
         if (k != null) {

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

@@ -27,17 +27,9 @@
 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 java.util.concurrent.TimeoutException;
 
-import java.io.EOFException;
-import java.io.IOException;
-import java.net.SocketException;
-import java.util.Map;
-
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Address;
 import com.rabbitmq.client.AlreadyClosedException;
@@ -47,7 +39,6 @@
 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.BlockingCell;
 import com.rabbitmq.utility.Utility;
@@ -70,13 +61,12 @@
  * 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;
-    
+
     /** Timeout used while waiting for a connection.close-ok (milliseconds) */
     public static final int CONNECTION_CLOSING_TIMEOUT = 10000;
-
 
     private static final Version clientVersion =
         new Version(AMQP.PROTOCOL.MAJOR, AMQP.PROTOCOL.MINOR);
@@ -100,23 +90,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;
 
-    /** List of all shutdown listeners associated with the connection */
-    public List<ShutdownListener> listeners
-            = Collections.synchronizedList(new LinkedList<ShutdownListener>());
-    
     public BlockingCell<Object> appContinuation = new BlockingCell<Object>();
 
     /**
@@ -127,14 +106,6 @@ 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 AlreadyClosedException
     {
@@ -492,7 +463,7 @@ public MainLoop() {
                     shutdown(ex, false, ex);
                 }
             }
-            
+
             // Finally, shut down our underlying data connection.
             _frameHandler.close();
 
@@ -724,55 +695,6 @@ public void close(int closeCode,
         }
         notifyListeners();
     }
-    
-    /**
-     * Private API - notify the listeners attached to this connection
-     * @see com.rabbitmq.client.ShutdownListener
-     */
-    public void notifyListeners()
-    {
-        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() {
         return "amqp://" + _params.getUserName() + "@" + getHost() + ":" + getPort() + _params.getVirtualHost();