Remove some QueueingConsumer usages in tests

References #213

Author: acogoluegnes

src/test/java/com/rabbitmq/client/test/ClientTests.java

@@ -33,7 +33,7 @@
     Bug20004Test.class,
     CloseInMainLoop.class,
     ChannelNumberAllocationTests.class,
-    QueueingConsumerShutdownTests.class,
+    QueueingConsumerTests.class,
     MultiThreadedChannel.class,
     IntAllocatorTests.class,
     AMQBuilderApiTest.class,

src/test/java/com/rabbitmq/client/test/QueueingConsumerShutdownTests.java renamed to src/test/java/com/rabbitmq/client/test/QueueingConsumerTests.java

@@ -16,9 +16,13 @@
 package com.rabbitmq.client.test;
 
 import static org.junit.Assert.*;
-import org.junit.Test;
 
+import com.rabbitmq.client.ConsumerCancelledException;
+import org.junit.Test;
 
+import java.io.IOException;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -27,7 +31,7 @@
 import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.ShutdownSignalException;
 
-public class QueueingConsumerShutdownTests extends BrokerTestCase{
+public class QueueingConsumerTests extends BrokerTestCase{
   static final String QUEUE = "some-queue";
   static final int THREADS = 5;
 
@@ -64,4 +68,36 @@ public class QueueingConsumerShutdownTests extends BrokerTestCase{
     assertEquals(0, count.get());
   }
 
+  @Test public void consumerCancellationInterruptsQueuingConsumerWait()
+      throws IOException, InterruptedException {
+    String queue = "cancel_notification_queue_for_queueing_consumer";
+    final BlockingQueue<Boolean> result = new ArrayBlockingQueue<Boolean>(1);
+    channel.queueDeclare(queue, false, true, false, null);
+    final QueueingConsumer consumer = new QueueingConsumer(channel);
+    Runnable receiver = new Runnable() {
+
+      public void run() {
+        try {
+          try {
+            consumer.nextDelivery();
+          } catch (ConsumerCancelledException e) {
+            result.put(true);
+            return;
+          } catch (ShutdownSignalException e) {
+          } catch (InterruptedException e) {
+          }
+          result.put(false);
+        } catch (InterruptedException e) {
+          fail();
+        }
+      }
+    };
+    Thread t = new Thread(receiver);
+    t.start();
+    channel.basicConsume(queue, consumer);
+    channel.queueDelete(queue);
+    assertTrue(result.take());
+    t.join();
+  }
+
 }

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

@@ -15,24 +15,21 @@
 
 package com.rabbitmq.client.test.functional;
 
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.test.BrokerTestCase;
+import org.junit.Test;
 
 import java.io.IOException;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Test;
-
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Consumer;
-import com.rabbitmq.client.ConsumerCancelledException;
-import com.rabbitmq.client.DefaultConsumer;
-import com.rabbitmq.client.QueueingConsumer;
-import com.rabbitmq.client.ShutdownSignalException;
-import com.rabbitmq.client.test.BrokerTestCase;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 public class ConsumerCancelNotification extends BrokerTestCase {
 
@@ -43,7 +40,7 @@ public class ConsumerCancelNotification extends BrokerTestCase {
         final BlockingQueue<Boolean> result = new ArrayBlockingQueue<Boolean>(1);
 
         channel.queueDeclare(queue, false, true, false, null);
-        Consumer consumer = new QueueingConsumer(channel) {
+        Consumer consumer = new DefaultConsumer(channel) {
             @Override
             public void handleCancel(String consumerTag) throws IOException {
                 try {
@@ -58,38 +55,6 @@ public void handleCancel(String consumerTag) throws IOException {
         assertTrue(result.take());
     }
 
-    @Test public void consumerCancellationInterruptsQueuingConsumerWait()
-            throws IOException, InterruptedException {
-        final BlockingQueue<Boolean> result = new ArrayBlockingQueue<Boolean>(1);
-        channel.queueDeclare(queue, false, true, false, null);
-        final QueueingConsumer consumer = new QueueingConsumer(channel);
-        Runnable receiver = new Runnable() {
-
-            public void run() {
-                try {
-                    try {
-                        consumer.nextDelivery();
-                    } catch (ConsumerCancelledException e) {
-                        result.put(true);
-                        return;
-                    } catch (ShutdownSignalException e) {
-                    } catch (InterruptedException e) {
-                    }
-                    result.put(false);
-                } catch (InterruptedException e) {
-                    fail();
-                }
-            }
-        };
-        Thread t = new Thread(receiver);
-        t.start();
-        channel.basicConsume(queue, consumer);
-        channel.queueDelete(queue);
-        assertTrue(result.take());
-        t.join();
-    }
-
-
     class AlteringConsumer extends DefaultConsumer {
         private final String altQueue;
         private final CountDownLatch latch;

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

@@ -15,21 +15,20 @@
 
 package com.rabbitmq.client.test.functional;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import com.rabbitmq.client.*;
+import com.rabbitmq.client.test.BrokerTestCase;
+import org.junit.Test;
 
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 
-import org.junit.Test;
-
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.MessageProperties;
-import com.rabbitmq.client.QueueingConsumer;
-import com.rabbitmq.client.test.BrokerTestCase;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
 
 public class ConsumerPriorities extends BrokerTestCase {
     @Test public void validation() throws IOException {
@@ -43,7 +42,7 @@ private void assertFailValidation(Map<String, Object> args) throws IOException {
         Channel ch = connection.createChannel();
         String queue = ch.queueDeclare().getQueue();
         try {
-            ch.basicConsume(queue, true, args, new QueueingConsumer(ch));
+            ch.basicConsume(queue, true, args, new DefaultConsumer(ch));
             fail("Validation should fail for " + args);
         } catch (IOException ioe) {
             checkShutdownSignal(AMQP.PRECONDITION_FAILED, ioe);
@@ -54,9 +53,9 @@ private void assertFailValidation(Map<String, Object> args) throws IOException {
 
     @Test public void consumerPriorities() throws Exception {
         String queue = channel.queueDeclare().getQueue();
-        QueueingConsumer highConsumer = new QueueingConsumer(channel);
-        QueueingConsumer medConsumer = new QueueingConsumer(channel);
-        QueueingConsumer lowConsumer = new QueueingConsumer(channel);
+        QueueMessageConsumer highConsumer = new QueueMessageConsumer(channel);
+        QueueMessageConsumer medConsumer = new QueueMessageConsumer(channel);
+        QueueMessageConsumer lowConsumer = new QueueMessageConsumer(channel);
         String high = channel.basicConsume(queue, true, args(1), highConsumer);
         String med = channel.basicConsume(queue, true, medConsumer);
         channel.basicConsume(queue, true, args(-1), lowConsumer);
@@ -78,17 +77,41 @@ private Map<String, Object> args(Object o) {
         return map;
     }
 
-    private void assertContents(QueueingConsumer qc, int count, String msg) throws InterruptedException {
+    private void assertContents(QueueMessageConsumer c, int count, String msg) throws InterruptedException {
         for (int i = 0; i < count; i++) {
-            QueueingConsumer.Delivery d = qc.nextDelivery();
-            assertEquals(msg, new String(d.getBody()));
+            byte[] body = c.nextDelivery(100);
+            assertEquals(msg, new String(body));
         }
-        assertEquals(null, qc.nextDelivery(0));
+        assertEquals(null, c.nextDelivery());
     }
 
     private void publish(String queue, int count, String msg) throws IOException {
         for (int i = 0; i < count; i++) {
             channel.basicPublish("", queue, MessageProperties.MINIMAL_BASIC, msg.getBytes());
         }
     }
+
+    class QueueMessageConsumer extends DefaultConsumer {
+
+        BlockingQueue<byte[]> messages = new LinkedBlockingQueue<byte[]>();
+
+        public QueueMessageConsumer(Channel channel) {
+            super(channel);
+        }
+
+        @Override
+        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+            messages.add(body);
+        }
+
+        byte[] nextDelivery() {
+            return messages.poll();
+        }
+
+        byte[] nextDelivery(long timeoutInMs) throws InterruptedException {
+            return messages.poll(timeoutInMs, TimeUnit.MILLISECONDS);
+        }
+
+    }
+
 }

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

@@ -23,9 +23,7 @@
 
 import java.io.IOException;
 import java.util.*;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 import static org.junit.Assert.*;
 
@@ -173,13 +171,13 @@ protected void releaseResources() throws IOException {
         channel.queueBind(DLQ, DLX, "test");
 
         //measure round-trip latency
-        QueueingConsumer c = new QueueingConsumer(channel);
+        QueueMessageConsumer c = new QueueMessageConsumer(channel);
         String cTag = channel.basicConsume(TEST_QUEUE_NAME, true, c);
         long start = System.currentTimeMillis();
         publish(null, "test");
-        Delivery d = c.nextDelivery(TTL);
+        byte[] body = c.nextDelivery(TTL);
         long stop = System.currentTimeMillis();
-        assertNotNull(d);
+        assertNotNull(body);
         channel.basicCancel(cTag);
         long latency = stop-start;
 
@@ -565,14 +563,14 @@ private void sleep(long millis) {
 
     /* check that each message arrives within epsilon of the
        publication time + TTL + latency */
-    private void checkPromptArrival(QueueingConsumer c,
+    private void checkPromptArrival(QueueMessageConsumer c,
                                     int count, long latency) throws Exception {
         long epsilon = TTL / 10;
         for (int i = 0; i < count; i++) {
-            Delivery d = c.nextDelivery(TTL + TTL + latency + epsilon);
-            assertNotNull("message #" + i + " did not expire", d);
+            byte[] body = c.nextDelivery(TTL + TTL + latency + epsilon);
+            assertNotNull("message #" + i + " did not expire", body);
             long now = System.currentTimeMillis();
-            long publishTime = Long.valueOf(new String(d.getBody()));
+            long publishTime = Long.valueOf(new String(body));
             long targetTime = publishTime + TTL + latency;
             assertTrue("expiry outside bounds (+/- " + epsilon + "): " +
                        (now - targetTime),
@@ -699,4 +697,27 @@ public void process(GetResponse getResponse) {
     private static String randomQueueName() {
         return DeadLetterExchange.class.getSimpleName() + "-" + UUID.randomUUID().toString();
     }
+
+    class QueueMessageConsumer extends DefaultConsumer {
+
+        BlockingQueue<byte[]> messages = new LinkedBlockingQueue<byte[]>();
+
+        public QueueMessageConsumer(Channel channel) {
+            super(channel);
+        }
+
+        @Override
+        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+            messages.add(body);
+        }
+
+        byte[] nextDelivery() {
+            return messages.poll();
+        }
+
+        byte[] nextDelivery(long timeoutInMs) throws InterruptedException {
+            return messages.poll(timeoutInMs, TimeUnit.MILLISECONDS);
+        }
+
+    }
 }