merge default

Author: Tim Watson

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

@@ -231,7 +231,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 +246,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

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 {

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

@@ -239,32 +239,51 @@ public void setDefaultConsumer(Consumer consumer) {
 
     /**
      * Sends a ShutdownSignal to all active consumers.
+     * Idempotent.
      * @param signal an exception signalling channel shutdown
      */
-    private CountDownLatch broadcastShutdownSignal(ShutdownSignalException signal) {
+    private void broadcastShutdownSignal(ShutdownSignalException signal) {
         Map<String, Consumer> snapshotConsumers;
         synchronized (_consumers) {
             snapshotConsumers = new HashMap<String, Consumer>(_consumers);
         }
-        return this.dispatcher.handleShutdownSignal(snapshotConsumers, signal);
+        this.finishedShutdownFlag = this.dispatcher.handleShutdownSignal(snapshotConsumers, signal);
     }
 
     /**
-     * Protected API - overridden to quiesce consumer work and broadcast the signal
-     * to all consumers after calling the superclass's method.
+     * Start to shutdown -- defer rest of processing until ready
      */
-    @Override public void processShutdownSignal(ShutdownSignalException signal,
+    private void startProcessShutdownSignal(ShutdownSignalException signal,
                                                 boolean ignoreClosed,
                                                 boolean notifyRpc)
+    {   super.processShutdownSignal(signal, ignoreClosed, notifyRpc);
+    }
+
+    /**
+     * Finish shutdown processing -- idempotent
+     */
+    private void finishProcessShutdownSignal()
     {
         this.dispatcher.quiesce();
-        super.processShutdownSignal(signal, ignoreClosed, notifyRpc);
-        this.finishedShutdownFlag = broadcastShutdownSignal(signal);
+        broadcastShutdownSignal(getCloseReason());
+
         synchronized (unconfirmedSet) {
             unconfirmedSet.notifyAll();
         }
     }
 
+    /**
+     * Protected API - overridden to quiesce consumer work and broadcast the signal
+     * to all consumers after calling the superclass's method.
+     */
+    @Override public void processShutdownSignal(ShutdownSignalException signal,
+                                                boolean ignoreClosed,
+                                                boolean notifyRpc)
+    {
+        startProcessShutdownSignal(signal, ignoreClosed, notifyRpc);
+        finishProcessShutdownSignal();
+    }
+
     CountDownLatch getShutdownLatch() {
         return this.finishedShutdownFlag;
     }
@@ -526,13 +545,18 @@ public void close(int closeCode,
             signal.initCause(cause);
         }
 
-        BlockingRpcContinuation<AMQCommand> k = new SimpleBlockingRpcContinuation();
+        BlockingRpcContinuation<AMQCommand> k = new BlockingRpcContinuation<AMQCommand>(){
+            @Override
+            public AMQCommand transformReply(AMQCommand command) {
+                ChannelN.this.finishProcessShutdownSignal();
+                return command;
+            }};
         boolean notify = false;
         try {
             // Synchronize the block below to avoid race conditions in case
             // connnection wants to send Connection-CloseOK
             synchronized (_channelMutex) {
-                processShutdownSignal(signal, !initiatedByApplication, true);
+                startProcessShutdownSignal(signal, !initiatedByApplication, true);
                 quiescingRpc(reason, k);
             }
 
@@ -581,7 +605,16 @@ public void basicPublish(String exchange, String routingKey,
                              BasicProperties props, byte[] body)
         throws IOException
     {
-        basicPublish(exchange, routingKey, false, false, props, body);
+        basicPublish(exchange, routingKey, false, props, body);
+    }
+
+    /** Public API - {@inheritDoc} */
+    public void basicPublish(String exchange, String routingKey,
+                             boolean mandatory,
+                             BasicProperties props, byte[] body)
+        throws IOException
+    {
+        basicPublish(exchange, routingKey, mandatory, false, props, body);
     }
 
     /** Public API - {@inheritDoc} */

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

@@ -30,7 +30,7 @@
  * Dispatches notifications to a {@link Consumer} on an internally-managed executor service and work
  * pool.
  * <p/>
- * Each {@link Channel} has a single {@link ConsumerDispatcher}, but the executor service and work
+ * Each {@link Channel} has a single <code>ConsumerDispatcher</code>, but the executor service and work
  * pool may be shared with other channels, typically those on the same {@link AMQConnection}.
  */
 final class ConsumerDispatcher {

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

@@ -164,18 +164,16 @@ private static <W> int drainTo(LinkedList<W> deList, Collection<W> c, int maxEle
      * @param item the work item to add to the client queue
      * @return <code><b>true</b></code> if and only if the client is marked <i>ready</i>
      * &mdash; <i>as a result of this work item</i>
-     * @throws IllegalArgumentException if key not registered.
      */
     public boolean addWorkItem(K key, W item) {
         synchronized (this.monitor) {
             Queue<W> queue = this.pool.get(key);
-            if (queue == null) {
-                throw new IllegalArgumentException("Client " + key + " not registered");
-            }
-            queue.offer(item);
-            if (isDormant(key)) {
-                dormantToReady(key);
-                return true;
+            if (queue != null) {
+                queue.offer(item);
+                if (isDormant(key)) {
+                    dormantToReady(key);
+                    return true;
+                }
             }
             return false;
         }

src/com/rabbitmq/tools/jsonrpc/JsonRpcClient.java

@@ -31,15 +31,13 @@
 import com.rabbitmq.tools.json.JSONWriter;
 
 /**
-	  <a
-	  href="http://json-rpc.org/wd/JSON-RPC-1-1-WD-20060807.html">JSON-RPC</a>
-	  is a lightweight RPC mechanism using <a
-	  href="http://www.json.org/">JSON</a> as a data language for
-	  request and reply messages. It is rapidly becoming a
-	  standard in web development, where it is used to make RPC
-	  requests over HTTP. RabbitMQ provides an AMQP transport
-	  binding for JSON-RPC in the form of
-	  the <code>JsonRpcClient</code> class.
+	  <a href="http://json-rpc.org">JSON-RPC</a> is a lightweight
+	  RPC mechanism using <a href="http://www.json.org/">JSON</a>
+	  as a data language for request and reply messages. It is
+	  rapidly becoming a standard in web development, where it is
+	  used to make RPC requests over HTTP. RabbitMQ provides an
+	  AMQP transport binding for JSON-RPC in the form of the
+	  <code>JsonRpcClient</code> class.
 
 	  JSON-RPC services are self-describing - each service is able
 	  to list its supported procedures, and each procedure

src/com/rabbitmq/utility/IntAllocator.java

@@ -28,20 +28,22 @@
  * <p/><b>Implementation notes:</b>
  * <br/>This was originally an ordered chain of non-overlapping Intervals,
  * together with a fixed size array cache for freed integers.
- * <br/>{@link #reserve()} was expensive in this scheme, whereas in the
- * present implementation it is O(1), as is {@link #free()}.
- * <br/>Although {@link #allocate()} is slightly slower than O(1) and in the
- * worst case could be O(N), the use of the {@link #lastIndex} field
+ * <br/>{@link #reserve(int)} was expensive in this scheme, whereas in the
+ * present implementation it is O(1), as is {@link #free(int)}.
+ * <p>Although {@link #allocate()} is slightly slower than O(1) and in the
+ * worst case could be O(N), the use of a "<code>lastIndex</code>" field
  * for starting the next scan for free integers means this is negligible.
- * <br/>The data representation overhead is O(N) where N is the size of the
+ * </p>
+ * <p>The data representation overhead is O(N) where N is the size of the
  * allocation range. One <code>long</code> is used for every 64 integers in the
  * range.
- * <br/>Very little Object creation and destruction occurs in use.
+ * </p>
+ * <p>Very little Object creation and destruction occurs in use.</p>
  */
 public class IntAllocator {
 
     private final int loRange; // the integer bit 0 represents
-    private final int hiRange; // the integer(+1) the highest bit represents
+    private final int hiRange; // one more than the integer the highest bit represents
     private final int numberOfBits; // relevant in freeSet
     private int lastIndex = 0; // for searching for FREE integers
     /** A bit is SET in freeSet if the corresponding integer is FREE

test/src/com/rabbitmq/client/impl/WorkPoolTests.java

@@ -13,19 +13,15 @@ public class WorkPoolTests extends TestCase {
     private WorkPool<String, Object> pool = new WorkPool<String, Object>();
 
     /**
-     * Test that an unknown key is rejected.
+     * Test unknown key tolerated silently
+     * @throws Exception untested
      */
-    public void testUnkownKey() {
-        try {
-            this.pool.addWorkItem("test", new Object());
-            fail("Expected IllegalArgumentException");
-        } catch (IllegalArgumentException e) {
-            // expected
-        }
+    public void testUnknownKey() throws Exception{
+        assertFalse(this.pool.addWorkItem("test", new Object()));
     }
 
     /**
-     * Test basic add work and remove work.
+     * Test add work and remove work
      * @throws Exception untested
      */
     public void testBasicInOut() throws Exception {

test/src/com/rabbitmq/client/test/functional/AlternateExchange.java

@@ -139,36 +139,35 @@ protected void checkGet(boolean[] expected) throws IOException {
      *
      * @param key the routing key of the message to be sent
      * @param mandatory whether the message should be marked as 'mandatory'
-     * @param immediate whether the message should be marked as 'immediate'
      * @param expected indicates which queues we expect the message to
      *        get routed to
      * @param ret whether a 'basic.return' is expected
      *
      * @see #checkGet(boolean[])
      */
-    protected void check(String key, boolean mandatory, boolean immediate,
-                         boolean[] expected, boolean ret)
+    protected void check(String key, boolean mandatory, boolean[] expected,
+                         boolean ret)
         throws IOException {
 
         gotReturn.set(false);
-        channel.basicPublish("x", key, mandatory, immediate, null,
+        channel.basicPublish("x", key, mandatory, false, null,
                              "ae-test".getBytes());
         checkGet(expected);
         assertEquals(ret, gotReturn.get());
     }
 
     protected void check(String key, boolean[] expected, boolean ret)
         throws IOException {
-        check(key, false, false, expected, ret);
+        check(key, false, expected, ret);
     }
 
-    protected void check(String key, boolean mandatory, boolean immediate,
-                         boolean ret) throws IOException {
-        check(key, mandatory, immediate, expected(key), ret);
+    protected void check(String key, boolean mandatory, boolean ret)
+        throws IOException {
+        check(key, mandatory, expected(key), ret);
     }
 
     protected void check(String key, boolean ret) throws IOException {
-        check(key, false, false, ret);
+        check(key, false, ret);
     }
 
     /**
@@ -198,9 +197,7 @@ public void testAe() throws IOException {
             //ordinary
             check(k, false);
             //mandatory
-            check(k, true, false, k.equals("z"));
-            //immediate
-            check(k, false, true, unrouted, true);
+            check(k, true, k.equals("z"));
         }
 
         cleanup();