Merge branch '4.1.x-stable' into 4.2.x-stable

Author: acogoluegnes

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

@@ -15,9 +15,11 @@
 
 package com.rabbitmq.client.test.functional;
 
-import com.rabbitmq.client.*;
 import com.rabbitmq.client.test.BrokerTestCase;
 import org.junit.Test;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -27,10 +29,17 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import java.util.concurrent.CountDownLatch;
+
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.Envelope;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.MessageProperties;
 
 public class ConsumerPriorities extends BrokerTestCase {
+
     @Test public void validation() throws IOException {
         assertFailValidation(args("banana"));
         assertFailValidation(args(new HashMap<Object, Object>()));
@@ -50,6 +59,8 @@ private void assertFailValidation(Map<String, Object> args) throws IOException {
     }
 
     private static final int COUNT = 10;
+    private static final long DELIVERY_TIMEOUT_MS = 100;
+    private static final long CANCEL_OK_TIMEOUT_MS = 10 * 1000;
 
     @Test public void consumerPriorities() throws Exception {
         String queue = channel.queueDeclare().getQueue();
@@ -61,13 +72,20 @@ private void assertFailValidation(Map<String, Object> args) throws IOException {
         channel.basicConsume(queue, true, args(-1), lowConsumer);
 
         publish(queue, COUNT, "high");
+        assertContents(highConsumer, COUNT, "high");
         channel.basicCancel(high);
+        assertTrue(
+            "High priority consumer should have been cancelled",
+            highConsumer.cancelLatch.await(CANCEL_OK_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+        );
         publish(queue, COUNT, "med");
+        assertContents(medConsumer, COUNT, "med");
         channel.basicCancel(med);
+        assertTrue(
+            "Medium priority consumer should have been cancelled",
+            medConsumer.cancelLatch.await(CANCEL_OK_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+        );
         publish(queue, COUNT, "low");
-
-        assertContents(highConsumer, COUNT, "high");
-        assertContents(medConsumer, COUNT, "med");
         assertContents(lowConsumer, COUNT, "low");
     }
 
@@ -77,12 +95,12 @@ private Map<String, Object> args(Object o) {
         return map;
     }
 
-    private void assertContents(QueueMessageConsumer c, int count, String msg) throws InterruptedException {
+    private void assertContents(QueueMessageConsumer qc, int count, String msg) throws InterruptedException {
         for (int i = 0; i < count; i++) {
-            byte[] body = c.nextDelivery(100);
+            byte[] body = qc.nextDelivery(DELIVERY_TIMEOUT_MS);
             assertEquals(msg, new String(body));
         }
-        assertEquals(null, c.nextDelivery());
+        assertEquals(null, qc.nextDelivery(DELIVERY_TIMEOUT_MS));
     }
 
     private void publish(String queue, int count, String msg) throws IOException {
@@ -91,10 +109,12 @@ private void publish(String queue, int count, String msg) throws IOException {
         }
     }
 
-    class QueueMessageConsumer extends DefaultConsumer {
+    private static class QueueMessageConsumer extends DefaultConsumer {
 
         BlockingQueue<byte[]> messages = new LinkedBlockingQueue<byte[]>();
 
+        CountDownLatch cancelLatch = new CountDownLatch(1);
+
         public QueueMessageConsumer(Channel channel) {
             super(channel);
         }
@@ -104,14 +124,14 @@ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProp
             messages.add(body);
         }
 
-        byte[] nextDelivery() {
-            return messages.poll();
+        @Override
+        public void handleCancelOk(String consumerTag) {
+            cancelLatch.countDown();
         }
 
         byte[] nextDelivery(long timeoutInMs) throws InterruptedException {
             return messages.poll(timeoutInMs, TimeUnit.MILLISECONDS);
         }
 
     }
-
 }