Merge in default

Author: Simon MacMullen

src/com/rabbitmq/client/Channel.java

@@ -215,11 +215,24 @@ public interface Channel extends ShutdownNotifier {
      * @param prefetchCount maximum number of messages that the server
      * will deliver, 0 if unlimited
      * @param global true if the settings should be applied to the
-     * entire connection rather than just the current channel
+     * entire channel rather than each consumer
      * @throws java.io.IOException if an error is encountered
      */
     void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;
 
+    /**
+     * Request a specific prefetchCount "quality of service" settings
+     * for this channel.
+     *
+     * @see #basicQos(int, int, boolean)
+     * @param prefetchCount maximum number of messages that the server
+     * will deliver, 0 if unlimited
+     * @param global true if the settings should be applied to the
+     * entire channel rather than each consumer
+     * @throws java.io.IOException if an error is encountered
+     */
+    void basicQos(int prefetchCount, boolean global) throws IOException;
+
     /**
      * Request a specific prefetchCount "quality of service" settings
      * for this channel.
@@ -615,6 +628,23 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      */
     String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;
 
+    /**
+     * Start a non-nolocal, non-exclusive consumer, with
+     * a server-generated consumerTag and specified arguments.
+     * @param queue the name of the queue
+     * @param autoAck true if the server should consider messages
+     * acknowledged once delivered; false if the server should expect
+     * explicit acknowledgements
+     * @param arguments a set of arguments for the consume
+     * @param callback an interface to the consumer object
+     * @return the consumerTag generated by the server
+     * @throws java.io.IOException if an error is encountered
+     * @see com.rabbitmq.client.AMQP.Basic.Consume
+     * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
+     * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
+     */
+    String basicConsume(String queue, boolean autoAck, Map<String, Object> arguments, Consumer callback) throws IOException;
+
     /**
      * Start a non-nolocal, non-exclusive consumer.
      * @param queue the name of the queue

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

@@ -599,6 +599,13 @@ public void basicQos(int prefetchSize, int prefetchCount, boolean global)
 	exnWrappingRpc(new Basic.Qos(prefetchSize, prefetchCount, global));
     }
 
+    /** Public API - {@inheritDoc} */
+    public void basicQos(int prefetchCount, boolean global)
+            throws IOException
+    {
+        basicQos(0, prefetchCount, global);
+    }
+
     /** Public API - {@inheritDoc} */
     public void basicQos(int prefetchCount)
 	throws IOException
@@ -939,6 +946,14 @@ public String basicConsume(String queue, boolean autoAck, Consumer callback)
         return basicConsume(queue, autoAck, "", callback);
     }
 
+    /** Public API - {@inheritDoc} */
+    public String basicConsume(String queue, boolean autoAck, Map<String, Object> arguments,
+                               Consumer callback)
+        throws IOException
+    {
+        return basicConsume(queue, autoAck, "", false, false, arguments, callback);
+    }
+
     /** Public API - {@inheritDoc} */
     public String basicConsume(String queue, boolean autoAck, String consumerTag,
                                Consumer callback)

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

@@ -23,7 +23,7 @@ private void assertFailValidation(Map<String, Object> args) throws IOException {
         Channel ch = connection.createChannel();
         String queue = ch.queueDeclare().getQueue();
         try {
-            ch.basicConsume(queue, true, "", false, false, args, new QueueingConsumer(ch));
+            ch.basicConsume(queue, true, args, new QueueingConsumer(ch));
             fail("Validation should fail for " + args);
         } catch (IOException ioe) {
             checkShutdownSignal(AMQP.PRECONDITION_FAILED, ioe);
@@ -37,14 +37,14 @@ public void testConsumerPriorities() throws Exception {
         QueueingConsumer highConsumer = new QueueingConsumer(channel);
         QueueingConsumer medConsumer = new QueueingConsumer(channel);
         QueueingConsumer lowConsumer = new QueueingConsumer(channel);
-        channel.basicConsume(queue, true, "high", false, false, args(1), highConsumer);
-        channel.basicConsume(queue, true, "med", false, false, null, medConsumer);
-        channel.basicConsume(queue, true, "low", false, false, args(-1), lowConsumer);
+        String high = channel.basicConsume(queue, true, args(1), highConsumer);
+        String med = channel.basicConsume(queue, true, medConsumer);
+        channel.basicConsume(queue, true, args(-1), lowConsumer);
 
         publish(queue, COUNT, "high");
-        channel.basicCancel("high");
+        channel.basicCancel(high);
         publish(queue, COUNT, "med");
-        channel.basicCancel("med");
+        channel.basicCancel(med);
         publish(queue, COUNT, "low");
 
         assertContents(highConsumer, COUNT, "high");

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

@@ -76,5 +76,6 @@ public static void add(TestSuite suite) {
         suite.addTestSuite(ConsumerPriorities.class);
         suite.addTestSuite(Policies.class);
         suite.addTestSuite(ConnectionRecovery.class);
+        suite.addTestSuite(PerConsumerPrefetch.class);
     }
 }

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

@@ -0,0 +1,132 @@
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.GetResponse;
+import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.QueueingConsumer.Delivery;
+import com.rabbitmq.client.test.BrokerTestCase;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Deque;
+
+import static com.rabbitmq.client.test.functional.QosTests.drain;
+
+public class PerConsumerPrefetch extends BrokerTestCase {
+    private String q;
+
+    @Override
+    protected void createResources() throws IOException {
+        q = channel.queueDeclare().getQueue();
+    }
+
+    private interface Closure {
+        public void makeMore(Deque<Delivery> deliveries) throws IOException;
+    }
+
+    public void testSingleAck() throws IOException {
+        testPrefetch(new Closure() {
+            public void makeMore(Deque<Delivery> deliveries) throws IOException {
+                for (Delivery del : deliveries) {
+                    ack(del, false);
+                }
+            }
+        });
+    }
+
+    public void testMultiAck() throws IOException {
+        testPrefetch(new Closure() {
+            public void makeMore(Deque<Delivery> deliveries) throws IOException {
+                ack(deliveries.getLast(), true);
+            }
+        });
+    }
+
+    public void testSingleNack() throws IOException {
+        for (final boolean requeue: Arrays.asList(false, true)) {
+            testPrefetch(new Closure() {
+                public void makeMore(Deque<Delivery> deliveries) throws IOException {
+                    for (Delivery del : deliveries) {
+                        nack(del, false, requeue);
+                    }
+                }
+            });
+        }
+    }
+
+    public void testMultiNack() throws IOException {
+        for (final boolean requeue: Arrays.asList(false, true)) {
+            testPrefetch(new Closure() {
+                public void makeMore(Deque<Delivery> deliveries) throws IOException {
+                    nack(deliveries.getLast(), true, requeue);
+                }
+            });
+        }
+    }
+
+    public void testRecover() throws IOException {
+        testPrefetch(new Closure() {
+            public void makeMore(Deque<Delivery> deliveries) throws IOException {
+                channel.basicRecover();
+            }
+        });
+    }
+
+    private void testPrefetch(Closure closure) throws IOException {
+        QueueingConsumer c = new QueueingConsumer(channel);
+        publish(q, 15);
+        consume(c, 5, false);
+        Deque<Delivery> deliveries = drain(c, 5);
+
+        ack(channel.basicGet(q, false), false);
+        drain(c, 0);
+
+        closure.makeMore(deliveries);
+        drain(c, 5);
+    }
+
+    public void testPrefetchOnEmpty() throws IOException {
+        QueueingConsumer c = new QueueingConsumer(channel);
+        publish(q, 5);
+        consume(c, 10, false);
+        drain(c, 5);
+        publish(q, 10);
+        drain(c, 5);
+    }
+
+    public void testAutoAckIgnoresPrefetch() throws IOException {
+        QueueingConsumer c = new QueueingConsumer(channel);
+        publish(q, 10);
+        consume(c, 1, true);
+        drain(c, 10);
+    }
+
+    public void testPrefetchZeroMeansInfinity() throws IOException {
+        QueueingConsumer c = new QueueingConsumer(channel);
+        publish(q, 10);
+        consume(c, 0, false);
+        drain(c, 10);
+    }
+
+    private void publish(String q, int n) throws IOException {
+        for (int i = 0; i < n; i++) {
+            channel.basicPublish("", q, null, "".getBytes());
+        }
+    }
+
+    private void consume(QueueingConsumer c, int prefetch, boolean autoAck) throws IOException {
+        channel.basicQos(prefetch);
+        channel.basicConsume(q, autoAck, c);
+    }
+
+    private void ack(Delivery del, boolean multi) throws IOException {
+        channel.basicAck(del.getEnvelope().getDeliveryTag(), multi);
+    }
+
+    private void ack(GetResponse get, boolean multi) throws IOException {
+        channel.basicAck(get.getEnvelope().getDeliveryTag(), multi);
+    }
+
+    private void nack(Delivery del, boolean multi, boolean requeue) throws IOException {
+        channel.basicNack(del.getEnvelope().getDeliveryTag(), multi, requeue);
+    }
+}