Update tests

Author: Simon MacMullen

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

@@ -1,7 +1,5 @@
 package com.rabbitmq.client.test.functional;
 
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.GetResponse;
 import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.QueueingConsumer.Delivery;
@@ -10,8 +8,6 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Deque;
-import java.util.HashMap;
-import java.util.Map;
 
 import static com.rabbitmq.client.test.functional.QosTests.drain;
 
@@ -111,31 +107,15 @@ public void testPrefetchZeroMeansInfinity() throws IOException {
         drain(c, 10);
     }
 
-    public void testPrefetchValidation() throws IOException {
-        validationFail(-1);
-        validationFail(new HashMap<String, Object>());
-        validationFail("banana");
-    }
-
-    private void validationFail(Object badThing) throws IOException {
-        Channel ch = connection.createChannel();
-        QueueingConsumer c = new QueueingConsumer(ch);
-
-        try {
-            ch.basicConsume(q, false, args(badThing), c);
-        } catch (IOException e) {
-            checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
-        }
-    }
-
     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.basicConsume(q, autoAck, args(prefetch), c);
+        channel.basicQos(prefetch);
+        channel.basicConsume(q, autoAck, c);
     }
 
     private void ack(Delivery del, boolean multi) throws IOException {
@@ -149,10 +129,4 @@ private void ack(GetResponse get, boolean multi) throws IOException {
     private void nack(Delivery del, boolean multi, boolean requeue) throws IOException {
         channel.basicNack(del.getEnvelope().getDeliveryTag(), multi, requeue);
     }
-
-    private Map<String, Object> args(Object prefetch) {
-        Map<String, Object> a = new HashMap<String, Object>();
-        a.put("x-prefetch", prefetch);
-        return a;
-    }
 }

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

@@ -86,17 +86,6 @@ public static Deque<Delivery> drain(QueueingConsumer c, int n)
         return res;
     }
 
-    public void testMessageLimitGlobalFails()
-        throws IOException
-    {
-        try {
-            channel.basicQos(0, 1, true);
-            fail("basic.qos{global=false} should not be supported");
-        } catch (IOException ioe) {
-            checkShutdownSignal(AMQP.NOT_IMPLEMENTED, ioe);
-        }
-    }
-
     public void testMessageLimitPrefetchSizeFails()
         throws IOException
     {
@@ -121,15 +110,15 @@ public void testNoAckNoAlterLimit()
     {
         QueueingConsumer c = new QueueingConsumer(channel);
         declareBindConsume(channel, c, true);
-        channel.basicQos(1);
+        channel.basicQos(1, true);
         fill(2);
         drain(c, 2);
     }
 
     public void testNoAckObeysLimit()
         throws IOException
     {
-        channel.basicQos(1);
+        channel.basicQos(1, true);
         QueueingConsumer c1 = new QueueingConsumer(channel);
         declareBindConsume(channel, c1, false);
         fill(1);
@@ -202,7 +191,7 @@ public void testSingleChannelAndQueueFairness()
         //channel & queue, and a prefetch limit set, that all
         //consumers get a fair share of the messages.
 
-        channel.basicQos(1);
+        channel.basicQos(1, true);
         String q = channel.queueDeclare().getQueue();
         channel.queueBind(q, "amq.fanout", "");
 
@@ -247,7 +236,7 @@ public void testSingleChannelAndQueueFairness()
     public void testConsumerLifecycle()
         throws IOException
     {
-        channel.basicQos(1);
+        channel.basicQos(1, true);
         QueueingConsumer c = new QueueingConsumer(channel);
         String queue = "qosTest";
         channel.queueDeclare(queue, false, false, false, null);
@@ -270,7 +259,7 @@ public void testSetLimitAfterConsume()
     {
         QueueingConsumer c = new QueueingConsumer(channel);
         declareBindConsume(c);
-        channel.basicQos(1);
+        channel.basicQos(1, true);
         fill(3);
         //We actually only guarantee that the limit takes effect
         //*eventually*, so this can in fact fail. It's pretty unlikely
@@ -285,7 +274,7 @@ public void testLimitIncrease()
     {
         QueueingConsumer c = new QueueingConsumer(channel);
         configure(c, 1, 3);
-        channel.basicQos(2);
+        channel.basicQos(2, true);
         drain(c, 1);
     }
 
@@ -294,7 +283,7 @@ public void testLimitDecrease()
     {
         QueueingConsumer c = new QueueingConsumer(channel);
         Queue<Delivery> d = configure(c, 2, 4);
-        channel.basicQos(1);
+        channel.basicQos(1, true);
         drain(c, 0);
         ack(d, true);
         drain(c, 1);
@@ -305,7 +294,7 @@ public void testLimitedToUnlimited()
     {
         QueueingConsumer c = new QueueingConsumer(channel);
         configure(c, 1, 3);
-        channel.basicQos(0);
+        channel.basicQos(0, true);
         drain(c, 2);
     }
 
@@ -320,8 +309,8 @@ public void testLimitingMultipleChannels()
         String q2 = declareBindConsume(ch2, c2, false);
         ch1.basicConsume(q2, false, c1);
         ch2.basicConsume(q1, false, c2);
-        ch1.basicQos(1);
-        ch2.basicQos(1);
+        ch1.basicQos(1, true);
+        ch2.basicQos(1, true);
         fill(5);
         Queue<Delivery> d1 = drain(c1, 1);
         Queue<Delivery> d2 = drain(c2, 1);
@@ -340,13 +329,13 @@ public void testLimitInheritsUnackedCount()
         declareBindConsume(c);
         fill(1);
         drain(c, 1);
-        channel.basicQos(2);
+        channel.basicQos(2, true);
         fill(2);
         drain(c, 1);
     }
 
     public void testRecoverReducesLimit() throws Exception {
-        channel.basicQos(2);
+        channel.basicQos(2, true);
         QueueingConsumer c = new QueueingConsumer(channel);
         declareBindConsume(c);
         fill(3);
@@ -427,7 +416,7 @@ protected List<String> configure(QueueingConsumer c,
                                      int messages)
         throws IOException
     {
-        channel.basicQos(limit);
+        channel.basicQos(limit, true);
 
         //declare/bind/consume-from queues
         List <String> queues = new ArrayList<String>();
@@ -446,7 +435,7 @@ protected Queue<Delivery> configure(QueueingConsumer c,
                                         int messages)
         throws IOException
     {
-        channel.basicQos(limit);
+        channel.basicQos(limit, true);
         declareBindConsume(c);
         fill(messages);
         return drain(c, limit);