Simple tests. The QosTests class is surprisingly unhelpful here, almost all of it is to do with how qos needs to be shared, survive consumers starting and stopping, etc.

Author: Simon MacMullen

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

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

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

@@ -0,0 +1,65 @@
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.QueueingConsumer.Delivery;
+import com.rabbitmq.client.test.BrokerTestCase;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+
+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();
+    }
+
+    public void testPrefetch() throws IOException {
+        QueueingConsumer c = new QueueingConsumer(channel);
+        publish(q, 5);
+        consume(c, false);
+        Queue<Delivery> dels = drain(c, 2);
+        for (Delivery del : dels) {
+            ack(del, false);
+        }
+        Delivery last = drain(c, 2).getLast();
+        ack(last, true);
+        drain(c, 1);
+        publish(q, 5);
+        drain(c, 1);
+    }
+
+    public void testAutoAckIgnoresPrefetch() throws IOException {
+        QueueingConsumer c = new QueueingConsumer(channel);
+        publish(q, 10);
+        consume(c, true);
+        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, boolean autoAck) throws IOException {
+        channel.basicConsume(q, autoAck, "", false, false, args(2), c);
+    }
+
+    private void ack(Delivery del, boolean multi) throws IOException {
+        channel.basicAck(del.getEnvelope().getDeliveryTag(), multi);
+    }
+
+    private Map<String, Object> args(int prefetch) {
+        Map<String, Object> a = new HashMap<String, Object>();
+        if (prefetch != 0) {
+            a.put("x-prefetch", prefetch);
+        }
+        return a;
+    }
+}

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

@@ -22,6 +22,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Deque;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
@@ -65,10 +66,10 @@ public void fill(int n)
      * receive n messages - check that we receive no fewer and cannot
      * receive more
      **/
-    public Queue<Delivery> drain(QueueingConsumer c, int n)
+    public static Deque<Delivery> drain(QueueingConsumer c, int n)
         throws IOException
     {
-        Queue<Delivery> res = new LinkedList<Delivery>();
+        Deque<Delivery> res = new LinkedList<Delivery>();
         try {
             long start = System.currentTimeMillis();
             for (int i = 0; i < n; i++) {