Migrate branch bug18743

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

@@ -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

@@ -121,10 +121,75 @@ public interface Connection { // rename to AMQPConnection later, this is a tempo
     Channel createChannel(int channelNumber) throws IOException;
 
     /**
-     * Close this connection with the given code and message.
-     * @param closeCode code indicating the reason for closing the connection - see AMQP spec for a list of codes
-     * @param closeMessage optional message describing the reason for closing the connection
+     * Close this connection and all its channels.
+     * 
+     * This method will wait infinitely for all the close operations to
+     * complete.
+     * 
      * @throws IOException if an I/O problem is encountered
      */
-    void close(int closeCode, String closeMessage) throws IOException;
+    void close() throws IOException;
+    
+    /**
+     * Close this connection and all its channels
+     * 
+     * This method will wait with the given timeout for all the close
+     * operations to complete. If timeout is reached then socket is forced
+     * to close
+     * @param timeout timeout (in milioseconds) for completing all the close-related operations, use 0 for infinity
+     * @throws IOException if an I/O problem is encountered
+     */
+    void close(int timeout) throws IOException;
+    
+    /**
+     * Abort this connection and all its channels.
+     * 
+     * This method will force the connection to close. It will silently discard
+     * any exceptions enountered in close operations
+     */
+    void abort();
+    
+    /**
+     * Abort this connection and all its channels.
+     * 
+     * This method behaves in a similar way as abort(), with the only difference
+     * that it will wait with a provided timeout for all the close operations to
+     * 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

@@ -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

@@ -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

@@ -26,12 +26,13 @@
 package com.rabbitmq.client.impl;
 
 import java.io.IOException;
+import java.util.concurrent.TimeoutException;
 
+import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.Command;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.utility.BlockingValueOrException;
-import com.rabbitmq.utility.SingleShotLinearTimer;
 
 /**
  * Base class modelling an AMQ channel. Subclasses implement close()
@@ -53,9 +54,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.
@@ -117,6 +118,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 +165,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");
         }
     }
 
@@ -211,34 +231,11 @@ public synchronized void rpc(Method m, RpcContinuation k)
     public AMQCommand quiescingRpc(Method m,
                                    int timeoutMillisec,
                                    final AMQCommand timeoutReply)
-        throws IOException, ShutdownSignalException
+        throws IOException, ShutdownSignalException, TimeoutException
     {
-        SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation();
+    	TimeoutBlockingRpcContinuation k = new TimeoutBlockingRpcContinuation();
         transmitAndEnqueue(m, k);
-        if (timeoutMillisec != 0) {
-            SingleShotLinearTimer timer = new SingleShotLinearTimer();
-
-            Runnable task = new Runnable() {
-                    public void run() {
-                        // Timed out waiting for reply.
-                        // Simulate a reply.
-                        // TODO: Warn the user somehow??
-                        try {
-                            handleCompleteInboundCommand(timeoutReply);
-                        } catch (IOException ioe) {
-                            // Ignore.
-                        }
-                    }
-                };
-            timer.schedule(task, timeoutMillisec);
-            try {
-                return k.getReply();
-            } finally {
-                timer.cancel();
-            }
-        } else {
-            return k.getReply();
-        }
+        return k.getReply(timeoutMillisec);
     }
 
     /**
@@ -253,15 +250,15 @@ public void run() {
     @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
-            _isOpen = false;
+            _cause = signal;
         }
         RpcContinuation k = nextOutstandingRpc();
         if (k != null) {
@@ -309,4 +306,14 @@ public AMQCommand transformReply(AMQCommand command) {
             return command;
         }
     }
+    
+    public static class TimeoutBlockingRpcContinuation
+        extends SimpleBlockingRpcContinuation
+    {
+    	public AMQCommand getReply(int timeout)
+    	    throws ShutdownSignalException, TimeoutException
+    	{
+    		return _blocker.uninterruptibleGetValue(timeout);
+    	}
+    }
 }