Merged default into bug19559

Author: Hubert Plociniczak

build.xml

@@ -44,17 +44,37 @@
   <target name="amqp-generate" depends="amqp-generate-check"
     unless="amqp.generate.notRequired" description="generate AMQP.java and AMQImpl.java from AMQP spec">
     <mkdir dir="${src.generated}/com/rabbitmq/client/"/>
-    <exec dir="." executable="python" output="${src.generated}/com/rabbitmq/client/AMQP.java">
+    <exec dir="." executable="python"
+          output="${src.generated}/com/rabbitmq/client/AMQP.java"
+    	  errorproperty="amqp.generate.error1"
+          resultproperty="amqp.generate.result1">
       <arg line="codegen.py"/>
       <arg line="header"/>
       <arg line="${AMQP_SPEC_JSON_PATH}"/>
     </exec>
+  	<fail message="Generation of AMQP.java failed with message:${line.separator}${amqp.generate.error1}">
+        <condition>
+            <not>
+                <equals arg1="${amqp.generate.result1}" arg2="0" />
+            </not>
+        </condition>
+    </fail>
     <mkdir dir="${src.generated}/com/rabbitmq/client/impl"/>
-    <exec dir="." executable="python" output="${src.generated}/com/rabbitmq/client/impl/AMQImpl.java">
+    <exec dir="." executable="python"
+          output="${src.generated}/com/rabbitmq/client/impl/AMQImpl.java"
+          errorproperty="amqp.generate.error2"
+          resultproperty="amqp.generate.result2">
       <arg line="codegen.py"/>
       <arg line="body"/>
       <arg line="${AMQP_SPEC_JSON_PATH}"/>
     </exec>
+  	<fail message="Generation of AMQPImpl.java failed with message:${line.separator}${amqp.generate.error2}">
+        <condition>
+            <not>
+                <equals arg1="${amqp.generate.result2}" arg2="0" />
+            </not>
+        </condition>
+  	</fail>
   </target>
 
   <target name="build" depends="amqp-generate">

src/com/rabbitmq/client/Channel.java

@@ -70,9 +70,18 @@ public interface Channel extends ShutdownNotifier{
     Connection getConnection();
 
     /**
-     * Close this channel with the given code and message
+     * Close this channel with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code
+     * and message 'OK'.
+     * 
+     * @throws java.io.IOException if an error is encountered
+     */
+    void close() throws IOException;
+    
+    /**
+     * Close this channel.
+     * 
      * @param closeCode the close code (See under "Reply Codes" in the AMQP specification)
-     * @param closeMessage a message indicating the reason for closing the channel
+     * @param closeMessage a message indicating the reason for closing the connection
      * @throws java.io.IOException if an error is encountered
      */
     void close(int closeCode, String closeMessage) throws IOException;
@@ -189,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.
@@ -241,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
@@ -309,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/Connection.java

@@ -121,44 +121,103 @@ public interface Connection extends ShutdownNotifier { // rename to AMQPConnecti
     Channel createChannel(int channelNumber) throws IOException;
 
     /**
-     * Close this connection and all its channels.
+     * Close this connection and all its channels
+     * with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code
+     * and message 'OK'.
      *
-     * This method will wait infinitely for all the close operations to
-     * complete.
+     * Waits for all the close operations to complete.
      *
      * @throws IOException if an I/O problem is encountered
      */
     void close() throws IOException;
+    
+    /**
+     * Close this connection and all its channels.
+     *
+     * Waits for all the close operations to complete.
+     *
+     * @param closeCode the close code (See under "Reply Codes" in the AMQP specification)
+     * @param closeMessage a message indicating the reason for closing the connection
+     * @throws IOException if an I/O problem is encountered
+     */
+    void close(int closeCode, String closeMessage) 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
+     * with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code
+     * and message 'OK'.
+     * 
+     * This method behaves in a similar way as {@link #close()}, with the only difference
+     * that it waits with a provided timeout for all the close operations to
+     * complete. When timeout is reached the socket is forced to close.
+     * 
+     * @param timeout timeout (in milliseconds) for completing all the close-related
      * operations, use -1 for infinity
      * @throws IOException if an I/O problem is encountered
      */
     void close(int timeout) throws IOException;
+    
+    /**
+     * Close this connection and all its channels.
+     *
+     * Waits with the given timeout for all the close operations to complete.
+     * When timeout is reached the socket is forced to close.
+     * 
+     * @param closeCode the close code (See under "Reply Codes" in the AMQP specification)
+     * @param closeMessage a message indicating the reason for closing the connection
+     * @param timeout timeout (in milliseconds) for completing all the close-related
+     * operations, use -1 for infinity
+     * @throws IOException if an I/O problem is encountered
+     */
+    void close(int closeCode, String closeMessage, int timeout) throws IOException;
 
     /**
-     * Abort this connection and all its channels.
+     * Abort this connection and all its channels
+     * with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code
+     * and message 'OK'.
      *
-     * This method will force the connection to close. It will silently discard
-     * any exceptions enountered in close operations
+     * Forces the connection to close.
+     * Any encountered exceptions in the close operations are silently discarded.
      */
     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.
+     * Forces the connection to close and waits for all the close operations to complete.
+     * Any encountered exceptions in the close operations are silently discarded.
+     * 
+     * @param closeCode the close code (See under "Reply Codes" in the AMQP specification)
+     * @param closeMessage a message indicating the reason for closing the connection
+     */
+    void abort(int closeCode, String closeMessage);
+    
+    /**
+     * Abort this connection and all its channels
+     * with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code
+     * and message 'OK'.
+     *
+     * This method behaves in a similar way as {@link #abort()}, with the only difference
+     * that it waits with a provided timeout for all the close operations to
+     * complete. When timeout is reached the socket is forced to close.
      *
-     * @param timeout timeout (in miliseconds) for completing all the close-related
+     * @param timeout timeout (in milliseconds) for completing all the close-related
      * operations, use -1 for infinity
      */
     void abort(int timeout);
+
+    /**
+     * Abort this connection and all its channels.
+     *
+     * Forces the connection to close and waits with the given timeout
+     * for all the close operations to complete. When timeout is reached
+     * the socket is forced to close.
+     * Any encountered exceptions in the close operations are silently discarded.
+     *
+     * @param closeCode the close code (See under "Reply Codes" in the AMQP specification)
+     * @param closeMessage a message indicating the reason for closing the connection
+     * @param timeout timeout (in milliseconds) for completing all the close-related
+     * operations, use -1 for infinity
+     */
+    void abort(int closeCode, String closeMessage, int timeout);
 }

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

@@ -219,17 +219,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) {