merge default into bug19250

Author: tonyg

build.xml

@@ -1,5 +1,5 @@
 <?xml version="1.0"?>
-<project default="build">
+<project name="RabbitMQ Java client" default="build">
 
   <property file="build.properties"/>
   <property file="config.properties"/>
@@ -27,20 +27,54 @@
   </path>
 
   <property name="AMQP_SPEC_JSON_PATH" value="${codegen.dir}/amqp-${spec.version}.json"/>
-
-  <target name="amqp-generate" description="generate AMQP.java and AMQImpl.java from AMQP spec">
+  
+  <target name="amqp-generate-check" description="check if codegen needs to be run">
+    <uptodate property="amqp.generate.notRequired">
+      <srcfiles file="codegen.py"/>
+      <srcfiles dir="${codegen.dir}">
+        <include name="*" />
+      </srcfiles>
+      <compositemapper>
+        <mapper type="merge" to="${basedir}/${src.generated}/com/rabbitmq/client/impl/AMQImpl.java" />
+        <mapper type="merge" to="${basedir}/${src.generated}/com/rabbitmq/client/AMQP.java" />
+      </compositemapper>
+    </uptodate>
+  </target>
+  
+  <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">
@@ -236,7 +270,6 @@
 
   <target name="jar" depends="build">
     <mkdir dir="${lib.out}"/>
-	<echo message="Javac out = ${javac.out}"/>
     <antcall target="doJarWithTags">
        <param name="jar.name" value="rabbitmq-client"/>
        <param name="base" value="${javac.out}"/>
@@ -267,7 +300,6 @@
   </target>
 
     <target name="doJarWithTags">
-	<echo message="Javac out = ${basedir}"/>
         <jar destfile="${lib.out}/${jar.name}.jar"
              basedir="${base}">
             <manifest>

scripts/runperftest.sh

@@ -2,7 +2,7 @@
 
 run() {
     echo "=== running with '$2'"
-    sh ./runjava.sh com.rabbitmq.examples.MulticastMain -h $1 -z 10 -i 20 $2
+    sh `dirname $0`/runjava.sh com.rabbitmq.examples.MulticastMain -h $1 -z 10 -i 20 $2
     sleep 2
 }
 

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;

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

@@ -155,13 +155,6 @@ public synchronized void enqueueRpc(RpcContinuation k)
         _activeRpc = k;
     }
 
-    public synchronized void transmitAndEnqueue(Method m, RpcContinuation k)
-        throws IOException
-    {
-        enqueueRpc(k);
-        transmit(m);
-    }
-
     public synchronized RpcContinuation nextOutstandingRpc()
     {
         RpcContinuation result = _activeRpc;
@@ -200,20 +193,14 @@ public synchronized void rpc(Method m, RpcContinuation k)
         throws IOException
     {
         ensureIsOpen();
-        transmitAndEnqueue(m, k);
+        quiescingRpc(m, k);
     }
-
-    /**
-     * Just like rpc(Method), but for use during quiescing/close/shutdown.
-     * Not for regular use. Doesn't do the ensureIsOpen() check.
-     */
-    public AMQCommand quiescingRpc(Method m,
-                                   int timeoutMillisec)
-        throws IOException, ShutdownSignalException, TimeoutException
+    
+    public synchronized void quiescingRpc(Method m, RpcContinuation k)
+        throws IOException
     {
-        SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation();
-        transmitAndEnqueue(m, k);
-        return k.getReply(timeoutMillisec);
+        enqueueRpc(k);
+        quiescingTransmit(m);
     }
 
     /**
@@ -228,26 +215,56 @@ 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
+     * @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);
         }
+    }
+    
+    public void notifyOutstandingRpc(ShutdownSignalException signal) {
         RpcContinuation k = nextOutstandingRpc();
         if (k != null) {
             k.handleShutdownSignal(signal);
         }
     }
 
-    public void transmit(Method m) throws IOException {
+    public synchronized void transmit(Method m) throws IOException {
+        ensureIsOpen();
+        quiescingTransmit(m);
+    }
+
+    public synchronized void transmit(AMQCommand c) throws IOException {
+        ensureIsOpen();
+        quiescingTransmit(c);
+    }
+
+    public synchronized void quiescingTransmit(Method m) throws IOException {
         new AMQCommand(m).transmit(this);
     }
 
+    public synchronized void quiescingTransmit(AMQCommand c) throws IOException {
+        c.transmit(this);
+    }
+
     public AMQConnection getAMQConnection() {
         return _connection;
     }