Merge bug25056 (very late!)

Author: Simon MacMullen

build.xml

@@ -308,7 +308,7 @@
   	</fail>
   </target>
 
-  <target name="test-suite-run" depends="test-client, test-ssl, test-server, test-functional, test-main-silent"/>
+  <target name="test-suite-run" depends="test-client, test-ssl, test-server, test-functional, test-functional-and-server-with-ha, test-main-silent"/>
 
   <target name="test-client" depends="test-build" description="Run the client test suites.">
     <junit printSummary="withOutAndErr"
@@ -369,6 +369,19 @@
     </junit>
   </target>
 
+  <!-- TODO: merge test-server, test-functional and this into one, once umbrellas have been merged -->
+  <target name="test-functional-and-server-with-ha" depends="detect-umbrella, test-build" if="UMBRELLA_AVAILABLE">
+    <junit printSummary="withOutAndErr"
+        haltOnFailure="${haltOnFailureJunit}"
+        failureproperty="test.failure"
+        fork="yes">
+      <classpath refid="test.classpath"/>
+      <formatter type="plain"/>
+      <formatter type="xml"/>
+      <test todir="${build.out}" name="com.rabbitmq.client.test.server.HATests"/>
+    </junit>
+  </target>
+
   <target name="test-single" depends="test-build">
     <junit printSummary="withOutAndErr"
         haltOnFailure="${haltOnFailureJunit}"
@@ -491,4 +504,3 @@
     </bundlor:bundlor>
   </target>
 </project>
-

pom.xml

@@ -29,36 +29,8 @@
 
     <developers>
       <developer>
-        <id>matthias.radestock</id>
-        <name>Matthias Radestock</name>
-        <roles>
-          <role>Developer</role>
-        </roles>
-      </developer>
-     <developer>
-        <id>tonyg</id>
-        <name>Tony Garnock-Jones</name>
-        <roles>
-          <role>Developer</role>
-        </roles>
-      </developer>
-      <developer>
-        <id>matthew.sackman</id>
-        <name>Matthias Sackman</name>
-        <roles>
-          <role>Developer</role>
-        </roles>
-      </developer>
-      <developer>
-        <id>david.maciver</id>
-        <name>David MacIver</name>
-        <roles>
-          <role>Developer</role>
-        </roles>
-      </developer>
-      <developer>
-        <id>paul.jones</id>
-        <name>Paul Jones</name>
+        <id>rabbitmq.team</id>
+        <name>The RabbitMQ Team</name>
         <roles>
           <role>Developer</role>
         </roles>

src/com/rabbitmq/client/Channel.java

@@ -45,9 +45,17 @@
  * </ul>
  * <p>
  *
- * While a Channel can be used by multiple threads, it's important to ensure
- * that only one thread executes a command at once. Concurrent execution of
- * commands will likely cause an UnexpectedFrameError to be thrown.
+ * <p>
+ * {@link Channel} instances are safe for use by multiple
+ * threads. Requests into a {@link Channel} are serialized, with only one
+ * thread running commands at a time.
+ * As such, applications may prefer using a {@link Channel} per thread
+ * instead of sharing the same <code>Channel</code> across multiple threads.
+ *
+ * An <b>important caveat</b> to this is that confirms are <b>not</b> handled
+ * properly when a {@link Channel} is shared between multiple threads. In that
+ * scenario, it is therefore important to ensure that the {@link Channel}
+ * instance is <b>not</b> accessed concurrently by multiple threads.
  *
  */
 
@@ -231,7 +239,7 @@ public interface Channel extends ShutdownNotifier {
     void basicQos(int prefetchCount) throws IOException;
 
     /**
-     * Publish a message with both "mandatory" and "immediate" flags set to false
+     * Publish a message
      * @see com.rabbitmq.client.AMQP.Basic.Publish
      * @param exchange the exchange to publish the message to
      * @param routingKey the routing key
@@ -246,8 +254,22 @@ public interface Channel extends ShutdownNotifier {
      * @see com.rabbitmq.client.AMQP.Basic.Publish
      * @param exchange the exchange to publish the message to
      * @param routingKey the routing key
-     * @param mandatory true if we are requesting a mandatory publish
-     * @param immediate true if we are requesting an immediate publish
+     * @param mandatory true if the 'mandatory' flag is to be set
+     * @param props other properties for the message - routing headers etc
+     * @param body the message body
+     * @throws java.io.IOException if an error is encountered
+     */
+    void basicPublish(String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body)
+            throws IOException;
+
+    /**
+     * Publish a message
+     * @see com.rabbitmq.client.AMQP.Basic.Publish
+     * @param exchange the exchange to publish the message to
+     * @param routingKey the routing key
+     * @param mandatory true if the 'mandatory' flag is to be set
+     * @param immediate true if the 'immediate' flag is to be
+     * set. Note that the RabbitMQ server does not support this flag.
      * @param props other properties for the message - routing headers etc
      * @param body the message body
      * @throws java.io.IOException if an error is encountered
@@ -347,9 +369,9 @@ Exchange.DeclareOk exchangeDeclare(String exchange,
      * Bind an exchange to an exchange, with no extra arguments.
      * @see com.rabbitmq.client.AMQP.Exchange.Bind
      * @see com.rabbitmq.client.AMQP.Exchange.BindOk
-     * @param destination: the name of the exchange to which messages flow across the binding
-     * @param source: the name of the exchange from which messages flow across the binding
-     * @param routingKey: the routine key to use for the binding
+     * @param destination the name of the exchange to which messages flow across the binding
+     * @param source the name of the exchange from which messages flow across the binding
+     * @param routingKey the routine key to use for the binding
      * @return a binding-confirm method if the binding was successfully created
      * @throws java.io.IOException if an error is encountered
      */
@@ -359,10 +381,10 @@ Exchange.DeclareOk exchangeDeclare(String exchange,
      * Bind an exchange to an exchange.
      * @see com.rabbitmq.client.AMQP.Exchange.Bind
      * @see com.rabbitmq.client.AMQP.Exchange.BindOk
-     * @param destination: the name of the exchange to which messages flow across the binding
-     * @param source: the name of the exchange from which messages flow across the binding
-     * @param routingKey: the routine key to use for the binding
-     * @param arguments: other properties (binding parameters)
+     * @param destination the name of the exchange to which messages flow across the binding
+     * @param source the name of the exchange from which messages flow across the binding
+     * @param routingKey the routine key to use for the binding
+     * @param arguments other properties (binding parameters)
      * @return a binding-confirm method if the binding was successfully created
      * @throws java.io.IOException if an error is encountered
      */
@@ -372,9 +394,9 @@ Exchange.DeclareOk exchangeDeclare(String exchange,
      * Unbind an exchange from an exchange, with no extra arguments.
      * @see com.rabbitmq.client.AMQP.Exchange.Bind
      * @see com.rabbitmq.client.AMQP.Exchange.BindOk
-     * @param destination: the name of the exchange to which messages flow across the binding
-     * @param source: the name of the exchange from which messages flow across the binding
-     * @param routingKey: the routine key to use for the binding
+     * @param destination the name of the exchange to which messages flow across the binding
+     * @param source the name of the exchange from which messages flow across the binding
+     * @param routingKey the routine key to use for the binding
      * @return a binding-confirm method if the binding was successfully created
      * @throws java.io.IOException if an error is encountered
      */
@@ -384,10 +406,10 @@ Exchange.DeclareOk exchangeDeclare(String exchange,
      * Unbind an exchange from an exchange.
      * @see com.rabbitmq.client.AMQP.Exchange.Bind
      * @see com.rabbitmq.client.AMQP.Exchange.BindOk
-     * @param destination: the name of the exchange to which messages flow across the binding
-     * @param source: the name of the exchange from which messages flow across the binding
-     * @param routingKey: the routine key to use for the binding
-     * @param arguments: other properties (binding parameters)
+     * @param destination the name of the exchange to which messages flow across the binding
+     * @param source the name of the exchange from which messages flow across the binding
+     * @param routingKey the routine key to use for the binding
+     * @param arguments other properties (binding parameters)
      * @return a binding-confirm method if the binding was successfully created
      * @throws java.io.IOException if an error is encountered
      */

src/com/rabbitmq/client/ConnectionFactory.java

@@ -350,7 +350,7 @@ public void setClientProperties(Map<String, Object> clientProperties) {
 
     /**
      * Gets the sasl config to use when authenticating
-     * @return
+     * @return the sasl config
      * @see com.rabbitmq.client.SaslConfig
      */
     public SaslConfig getSaslConfig() {

src/com/rabbitmq/client/SaslMechanism.java

@@ -25,7 +25,7 @@
 public interface SaslMechanism {
     /**
      * The name of this mechanism (e.g. PLAIN)
-     * @return
+     * @return the name
      */
     String getName();
 

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

@@ -32,15 +32,15 @@
  */
 public class AMQCommand implements Command {
 
-    /** EMPTY_CONTENT_BODY_FRAME_SIZE = 8 = 1 + 2 + 4 + 1
+    /** EMPTY_FRAME_SIZE = 8 = 1 + 2 + 4 + 1
      * <ul><li>1 byte of frame type</li>
      * <li>2 bytes of channel number</li>
      * <li>4 bytes of frame payload length</li>
      * <li>1 byte of payload trailer FRAME_END byte</li></ul>
-     * See {@link #checkEmptyContentBodyFrameSize}, an assertion
-     * checked at startup.
+     * See {@link #checkEmptyFrameSize}, an assertion checked at
+     * startup.
      */
-    private static final int EMPTY_CONTENT_BODY_FRAME_SIZE = 8;
+    public static final int EMPTY_FRAME_SIZE = 8;
 
     /** The assembler for this command - synchronised on - contains all the state */
     private final CommandAssembler assembler;
@@ -108,7 +108,7 @@ public void transmit(AMQChannel channel) throws IOException {
 
                 int frameMax = connection.getFrameMax();
                 int bodyPayloadMax = (frameMax == 0) ? body.length : frameMax
-                        - EMPTY_CONTENT_BODY_FRAME_SIZE;
+                        - EMPTY_FRAME_SIZE;
 
                 for (int offset = 0; offset < body.length; offset += bodyPayloadMax) {
                     int remaining = body.length - offset;
@@ -157,26 +157,26 @@ private static StringBuilder contentBodyStringBuilder(byte[] body, boolean suppr
 
     /** Called to check internal code assumptions. */
     public static void checkPreconditions() {
-        checkEmptyContentBodyFrameSize();
+        checkEmptyFrameSize();
     }
 
     /**
-     * Since we're using a pre-computed value for
-     * EMPTY_CONTENT_BODY_FRAME_SIZE we check this is
-     * actually correct when run against the framing code in Frame.
+     * Since we're using a pre-computed value for EMPTY_FRAME_SIZE we
+     * check this is actually correct when run against the framing
+     * code in Frame.
      */
-    private static void checkEmptyContentBodyFrameSize() {
+    private static void checkEmptyFrameSize() {
         Frame f = new Frame(AMQP.FRAME_BODY, 0, new byte[0]);
         ByteArrayOutputStream s = new ByteArrayOutputStream();
         try {
             f.writeTo(new DataOutputStream(s));
         } catch (IOException ioe) {
-            throw new AssertionError("IOException while checking EMPTY_CONTENT_BODY_FRAME_SIZE");
+            throw new AssertionError("IOException while checking EMPTY_FRAME_SIZE");
         }
         int actualLength = s.toByteArray().length;
-        if (EMPTY_CONTENT_BODY_FRAME_SIZE != actualLength) {
-            throw new AssertionError("Internal error: expected EMPTY_CONTENT_BODY_FRAME_SIZE("
-                    + EMPTY_CONTENT_BODY_FRAME_SIZE
+        if (EMPTY_FRAME_SIZE != actualLength) {
+            throw new AssertionError("Internal error: expected EMPTY_FRAME_SIZE("
+                    + EMPTY_FRAME_SIZE
                     + ") is not equal to computed value: " + actualLength);
         }
     }

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

@@ -41,6 +41,7 @@
 import com.rabbitmq.client.SaslConfig;
 import com.rabbitmq.client.SaslMechanism;
 import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.impl.AMQChannel.BlockingRpcContinuation;
 import com.rabbitmq.utility.BlockingCell;
 import com.rabbitmq.utility.Utility;
 
@@ -663,6 +664,16 @@ public ShutdownSignalException shutdown(Object reason,
                          boolean initiatedByApplication,
                          Throwable cause,
                          boolean notifyRpc)
+    {
+        ShutdownSignalException sse = startShutdown(reason, initiatedByApplication, cause, notifyRpc);
+        finishShutdown(sse);
+        return sse;
+    }
+
+    private ShutdownSignalException startShutdown(Object reason,
+                         boolean initiatedByApplication,
+                         Throwable cause,
+                         boolean notifyRpc)
     {
         ShutdownSignalException sse = new ShutdownSignalException(true,initiatedByApplication,
                                                                   reason, this);
@@ -677,10 +688,12 @@ public ShutdownSignalException shutdown(Object reason,
 
         _channel0.processShutdownSignal(sse, !initiatedByApplication, notifyRpc);
 
+        return sse;
+    }
+
+    private void finishShutdown(ShutdownSignalException sse) {
         ChannelManager cm = _channelManager;
         if (cm != null) cm.handleSignal(sse);
-
-        return sse;
     }
 
     /** Public API - {@inheritDoc} */
@@ -775,10 +788,15 @@ public void close(int closeCode,
                     .replyText(closeMessage)
                 .build();
 
-            shutdown(reason, initiatedByApplication, cause, true);
+            final ShutdownSignalException sse = startShutdown(reason, initiatedByApplication, cause, true);
             if(sync){
-              AMQChannel.SimpleBlockingRpcContinuation k =
-                  new AMQChannel.SimpleBlockingRpcContinuation();
+                BlockingRpcContinuation<AMQCommand> k = new BlockingRpcContinuation<AMQCommand>(){
+                    @Override
+                    public AMQCommand transformReply(AMQCommand command) {
+                        AMQConnection.this.finishShutdown(sse);
+                        return command;
+                    }};
+
               _channel0.quiescingRpc(reason, k);
               k.getReply(timeout);
             } else {