Merged default into 18776

Author: 0x6e6562

src/com/rabbitmq/client/Channel.java

@@ -198,19 +198,6 @@ void basicPublish(int ticket, String exchange, String routingKey, boolean mandat
      */
     Exchange.DeclareOk exchangeDeclare(int ticket, String exchange, String type, boolean durable) throws IOException;
 
-    /**
-     * Actively declare a non-exclusive, non-autodelete queue
-     * The name of the new queue is held in the "queue" field of the {@link com.rabbitmq.client.AMQP.Queue.DeclareOk} result.
-     * @see com.rabbitmq.client.AMQP.Queue.Declare
-     * @see com.rabbitmq.client.AMQP.Queue.DeclareOk
-     * @param ticket an access ticket for the appropriate realm
-     * @param queue the name of the queue
-     * @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
-     * @return a declaration-confirm method to indicate the exchange was successfully declared
-     * @throws java.io.IOException if an error is encountered
-     */
-    Queue.DeclareOk queueDeclare(int ticket, String queue, boolean durable) throws IOException;
-
     /**
      * Declare an exchange, via an interface that allows the complete set of arguments
      * The name of the new queue is held in the "queue" field of the {@link com.rabbitmq.client.AMQP.Queue.DeclareOk} result.
@@ -250,6 +237,19 @@ Exchange.DeclareOk exchangeDeclare(int ticket, String exchange, String type, boo
      * @throws java.io.IOException if an error is encountered
      */
     Queue.DeclareOk queueDeclare(int ticket, String queue) throws IOException;
+    
+    /**
+     * Actively declare a non-exclusive, non-autodelete queue
+     * The name of the new queue is held in the "queue" field of the {@link com.rabbitmq.client.AMQP.Queue.DeclareOk} result.
+     * @see com.rabbitmq.client.AMQP.Queue.Declare
+     * @see com.rabbitmq.client.AMQP.Queue.DeclareOk
+     * @param ticket an access ticket for the appropriate realm
+     * @param queue the name of the queue
+     * @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
+     * @return a declaration-confirm method to indicate the exchange was successfully declared
+     * @throws java.io.IOException if an error is encountered
+     */
+    Queue.DeclareOk queueDeclare(int ticket, String queue, boolean durable) throws IOException;
 
     /**
      * Declare a queue
@@ -318,6 +318,33 @@ Queue.DeclareOk queueDeclare(int ticket, String queue, boolean passive, boolean
      * @throws java.io.IOException if an error is encountered
      */
     Queue.BindOk queueBind(int ticket, String queue, String exchange, String routingKey, Map<String, Object> arguments) throws IOException;
+    
+    /**
+     * Uninds a queue from an exchange, with no extra arguments.
+     * @see com.rabbitmq.client.AMQP.Queue.Unbind
+     * @see com.rabbitmq.client.AMQP.Queue.UnbindOk
+     * @param ticket an access ticket for the appropriate realm
+     * @param queue the name of the queue
+     * @param exchange the name of the exchange
+     * @param routingKey the routine key to use for the binding
+     * @return an unbinding-confirm method if the binding was successfully deleted
+     * @throws java.io.IOException if an error is encountered
+     */
+    Queue.UnbindOk queueUnbind(int ticket, String queue, String exchange, String routingKey) throws IOException;
+
+    /**
+     * Unbind a queue from an exchange.
+     * @see com.rabbitmq.client.AMQP.Queue.Unbind
+     * @see com.rabbitmq.client.AMQP.Queue.UnbindOk
+     * @param ticket an access ticket for the appropriate realm
+     * @param queue the name of the queue
+     * @param exchange the name of the exchange
+     * @param routingKey the routine key to use for the binding
+     * @param arguments other properties (binding parameters)
+     * @return an unbinding-confirm method if the binding was successfully deleted
+     * @throws java.io.IOException if an error is encountered
+     */
+    Queue.UnbindOk queueUnbind(int ticket, String queue, String exchange, String routingKey, Map<String, Object> arguments) throws IOException;
 
     /**
      * Retrieve a message from a queue using {@link com.rabbitmq.client.AMQP.Basic.Get}

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

@@ -215,17 +215,29 @@ public synchronized void quiescingRpc(Method m, RpcContinuation k)
     @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
+     * @param ignoreClosed the flag indicating whether to ignore the AlreadyClosedException
+     *                     thrown when the channel is already closed
+     * @param notifyRpc the flag indicating whether any remaining rpc continuation should be
+     *                  notified with the given signal
      */
-    public void processShutdownSignal(ShutdownSignalException signal) {
-        synchronized (this) {
-            ensureIsOpen(); // invariant: we should never be shut down more than once per instance
-            _shutdownCause = signal;
+    public void processShutdownSignal(ShutdownSignalException signal,
+                                      boolean ignoreClosed,
+                                      boolean notifyRpc) {
+        try {
+            synchronized (this) {
+                if (!ignoreClosed)
+                    ensureIsOpen(); // invariant: we should never be shut down more than once per instance
+                if (isOpen())
+                    _shutdownCause = signal;
+            }
+        } finally {
+            if (notifyRpc)
+                notifyOutstandingRpc(signal);
         }
-        notifyOutstandingRpc(signal);
     }
 
     public void notifyOutstandingRpc(ShutdownSignalException signal) {

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

@@ -103,6 +103,9 @@ public class AMQConnection extends ShutdownNotifierComponent implements Connecti
      */
     public BlockingCell<Object> _appContinuation = new BlockingCell<Object>();
 
+    /** Flag indicating whether the client received Connection.Close message from the broker */
+    public boolean _brokerInitiatedShutdown = false;
+    
     /**
      * Protected API - respond, in the driver thread, to a ShutdownSignal.
      * @param channelNumber the number of the channel to disconnect
@@ -191,6 +194,7 @@ public AMQConnection(ConnectionParameters params,
         _missedHeartbeats = 0;
         _heartbeat = 0;
         _exceptionHandler = exceptionHandler;
+        _brokerInitiatedShutdown = false;
 
         new MainLoop(); // start the main loop going
 
@@ -457,26 +461,15 @@ public MainLoop() {
                     }
                 }
             } catch (EOFException ex) {
-                if (isOpen()) {
-                    System.err.println("AMQConnection.mainLoop: connection close");
-                    shutdown(ex, false, ex);
-                }
+                if (!_brokerInitiatedShutdown)
+                    shutdown(ex, false, ex, true);
             } catch (Throwable ex) {
                 _exceptionHandler.handleUnexpectedConnectionDriverException(AMQConnection.this,
                                                                             ex);
-                if (isOpen()) {
-                    shutdown(ex, false, ex);
-                }
+                shutdown(ex, false, ex, true);
             } finally {
                 // Finally, shut down our underlying data connection.
                 _frameHandler.close();
-                
-                // Set shutdown exception for any outstanding rpc,
-                // so that it does not wait infinitely for Connection.CloseOk.
-                // This can only happen when the broker closed the socket
-                // unexpectedly.
-                _channel0.notifyOutstandingRpc(_shutdownCause);
-                
                 _appContinuation.set(null);
                 notifyListeners();
             }
@@ -562,18 +555,22 @@ public boolean processControlCommand(Command c)
     }
 
     public void handleConnectionClose(Command closeCommand) {
-        shutdown(closeCommand, false, null);
+        ShutdownSignalException sse = shutdown(closeCommand, false, null, false);
         try {
             _channel0.quiescingTransmit(new AMQImpl.Connection.CloseOk());
         } catch (IOException ioe) {
             Utility.emptyStatement();
         }
-        _heartbeat = 0; // Do not try to send heartbeats after CloseOk   
-        new SocketCloseWait();
+        _heartbeat = 0; // Do not try to send heartbeats after CloseOk
+        _brokerInitiatedShutdown = true;
+        new SocketCloseWait(sse);
     }
 
     private class SocketCloseWait extends Thread {
-        public SocketCloseWait() {
+        private ShutdownSignalException cause;
+        
+        public SocketCloseWait(ShutdownSignalException sse) {
+            cause = sse;
             start();
         }
 
@@ -586,6 +583,7 @@ public SocketCloseWait() {
                 _frameHandler.close();
             } finally {
                 _running = false;
+                _channel0.notifyOutstandingRpc(cause);
             }
         }
     }
@@ -594,27 +592,26 @@ public SocketCloseWait() {
      * Protected API - causes all attached channels to terminate with
      * a ShutdownSignal built from the argument, and stops this
      * connection from accepting further work from the application.
+     * 
+     * @return a shutdown signal built using the given arguments
      */
-    public void shutdown(Object reason,
+    public ShutdownSignalException shutdown(Object reason,
                          boolean initiatedByApplication,
-                         Throwable cause)
+                         Throwable cause,
+                         boolean notifyRpc)
     {
-        try {
-            synchronized (this) {
+        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
-                ShutdownSignalException sse = new ShutdownSignalException(true,
-                                                             initiatedByApplication,
-                                                             reason, this);
-                sse.initCause(cause);
+            if (isOpen())
                 _shutdownCause = sse;
-            }
-
-            _channel0.processShutdownSignal(_shutdownCause);
-        } catch (AlreadyClosedException ace) {
-            if (initiatedByApplication)
-                throw ace;
         }
-        _channelManager.handleSignal(_shutdownCause);
+        _channel0.processShutdownSignal(sse, !initiatedByApplication, notifyRpc);
+        _channelManager.handleSignal(sse);
+        return sse;
     }
 
     public void close()
@@ -690,7 +687,7 @@ public void close(int closeCode,
         try {
             AMQImpl.Connection.Close reason =
                 new AMQImpl.Connection.Close(closeCode, closeMessage, 0, 0);
-            shutdown(reason, initiatedByApplication, cause);
+            shutdown(reason, initiatedByApplication, cause, true);
             AMQChannel.SimpleBlockingRpcContinuation k =
                 new AMQChannel.SimpleBlockingRpcContinuation();
             _channel0.quiescingRpc(reason, k);

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

@@ -68,12 +68,7 @@ public void handleSignal(ShutdownSignalException signal) {
         }
         for (AMQChannel channel : channels) {
             disconnectChannel(channel.getChannelNumber());
-            try {
-                channel.processShutdownSignal(signal);
-            } catch (ShutdownSignalException sse) {
-                // Notify channels that are waiting for rpc
-                channel.notifyOutstandingRpc(signal);
-            }
+            channel.processShutdownSignal(signal, true, true);
         }
     }
 

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

@@ -146,8 +146,11 @@ public void broadcastShutdownSignal(ShutdownSignalException signal) {
      * Protected API - overridden to broadcast the signal to all
      * consumers before calling the superclass's method.
      */
-    @Override public void processShutdownSignal(ShutdownSignalException signal) {
-        super.processShutdownSignal(signal);
+    @Override public void processShutdownSignal(ShutdownSignalException signal,
+                                                boolean ignoreClosed,
+                                                boolean notifyRpc)
+    {
+        super.processShutdownSignal(signal, ignoreClosed, notifyRpc);
         broadcastShutdownSignal(signal);
     }
 
@@ -169,6 +172,22 @@ public void releaseChannelNumber() {
         // incoming commands except for a close-ok.
 
         Method method = command.getMethod();
+
+        if (method instanceof Channel.Close) {
+            // Channel should always respond to Channel.Close
+            // from the server
+            releaseChannelNumber();
+            ShutdownSignalException signal = new ShutdownSignalException(false,
+                                                                         false,
+                                                                         command,
+                                                                         this);
+            synchronized(this) {
+                processShutdownSignal(signal, true, true);
+                quiescingTransmit(new Channel.CloseOk());
+            }
+            notifyListeners();
+            return true;
+        }
         if (isOpen()) {
             // We're in normal running mode.
 
@@ -216,18 +235,6 @@ public void releaseChannelNumber() {
                     }
                 }
                 return true;
-            } else if (method instanceof Channel.Close) {
-                releaseChannelNumber();
-                ShutdownSignalException signal = new ShutdownSignalException(false,
-                                                                             false,
-                                                                             command,
-                                                                             this);
-                synchronized(this) {
-                    processShutdownSignal(signal);
-                    quiescingTransmit(new Channel.CloseOk());
-                }
-                notifyListeners();
-                return true;
             } else {
                 return false;
             }
@@ -289,7 +296,7 @@ public void close(int closeCode,
         // Synchronize the block below to avoid race conditions in case
         // connnection wants to send Connection-CloseOK
         synchronized(this) {
-            processShutdownSignal(signal);
+            processShutdownSignal(signal, !initiatedByApplication, true);
             quiescingRpc(reason, k);
         }
 
@@ -532,6 +539,20 @@ public Queue.BindOk queueBind(int ticket, String queue, String exchange,
                                           false, arguments)).getMethod();
     }
 
+    /**
+     * Public API - Unbind a queue from an exchange.
+     * @see com.rabbitmq.client.AMQP.Queue.Unbind
+     * @see com.rabbitmq.client.AMQP.Queue.UnbindOk
+     */
+    public Queue.UnbindOk queueUnbind(int ticket, String queue, String exchange,
+                                  String routingKey, Map<String, Object> arguments)
+        throws IOException
+    {
+        return (Queue.UnbindOk)
+            exnWrappingRpc(new Queue.Unbind(ticket, queue, exchange, routingKey,
+                                          arguments)).getMethod();
+    }
+
     /**
      * Public API - Bind a queue to an exchange, with no extra arguments.
      * @see com.rabbitmq.client.AMQP.Queue.Bind
@@ -543,6 +564,17 @@ public Queue.BindOk queueBind(int ticket, String queue, String exchange, String
         return queueBind(ticket, queue, exchange, routingKey, null);
     }
 
+    /**
+     * Public API - Unbind a queue from an exchange, with no extra arguments.
+     * @see com.rabbitmq.client.AMQP.Queue.Unbind
+     * @see com.rabbitmq.client.AMQP.Queue.UnbindOk
+     */
+    public Queue.UnbindOk queueUnbind(int ticket, String queue, String exchange, String routingKey)
+        throws IOException
+    {
+        return queueUnbind(ticket, queue, exchange, routingKey, null);
+    }
+
     /**
      * Public API - Retrieve a message from a queue using {@link com.rabbitmq.client.AMQP.Basic.Get}
      * @see com.rabbitmq.client.AMQP.Basic.Get