Remove QueueingConsumer usages in some tests

References #213

Author: acogoluegnes

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

@@ -18,12 +18,15 @@
 
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
+import com.rabbitmq.client.*;
 import org.junit.Test;
 
-import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.test.BrokerTestCase;
 
 public class NoRequeueOnCancel extends BrokerTestCase
@@ -43,11 +46,16 @@ protected void releaseResources() throws IOException {
     {
         channel.basicPublish("", Q, null, "1".getBytes());
 
-        QueueingConsumer c;
-
-        c = new QueueingConsumer(channel);
+        final CountDownLatch latch = new CountDownLatch(1);
+        Consumer c = new DefaultConsumer(channel) {
+            @Override
+            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+                latch.countDown();
+            }
+        };
         String consumerTag = channel.basicConsume(Q, false, c);
-        c.nextDelivery();
+        assertTrue(latch.await(5, TimeUnit.SECONDS));
+
         channel.basicCancel(consumerTag);
 
         assertNull(channel.basicGet(Q, true));

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

@@ -16,18 +16,18 @@
 
 package com.rabbitmq.client.test.functional;
 
-import static org.junit.Assert.fail;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.test.BrokerTestCase;
+import org.junit.Test;
 
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeoutException;
 
-import org.junit.Test;
-
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.QueueingConsumer;
-import com.rabbitmq.client.test.BrokerTestCase;
+import static org.junit.Assert.fail;
 
 /**
  * Test queue auto-delete and exclusive semantics.
@@ -125,7 +125,7 @@ void verifyNotEquivalent(boolean durable, boolean exclusive,
         channel.queueDeclare(name, false, false, true, null);
         // now it's there
         verifyQueue(name, false, false, true, null);
-        QueueingConsumer consumer = new QueueingConsumer(channel);
+        Consumer consumer = new DefaultConsumer(channel);
         String consumerTag = channel.basicConsume(name, consumer);
         channel.basicCancel(consumerTag);
         // now it's not .. we hope
@@ -143,7 +143,7 @@ void verifyNotEquivalent(boolean durable, boolean exclusive,
         channel.queueDeclare(name, false, true, false, null);
         // now it's there
         verifyQueue(name, false, true, false, null);
-        QueueingConsumer consumer = new QueueingConsumer(channel);
+        Consumer consumer = new DefaultConsumer(channel);
         String consumerTag = channel.basicConsume(name, consumer);
         channel.basicCancel(consumerTag);
         // and still there, because exclusive no longer implies autodelete

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

@@ -23,12 +23,13 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
+import com.rabbitmq.client.*;
 import org.junit.Test;
 
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.test.BrokerTestCase;
 
 public class Recover extends BrokerTestCase {
@@ -67,12 +68,20 @@ void verifyRedeliverOnRecover(RecoverCallback call)
 
     void verifyNoRedeliveryWithAutoAck(RecoverCallback call)
         throws IOException, InterruptedException {
-        QueueingConsumer consumer = new QueueingConsumer(channel);
+        final CountDownLatch latch = new CountDownLatch(1);
+        final AtomicReference<byte[]> bodyReference = new AtomicReference<byte[]>();
+        Consumer consumer = new DefaultConsumer(channel) {
+            @Override
+            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+                bodyReference.set(body);
+                latch.countDown();
+            }
+        };
         channel.basicConsume(queue, true, consumer); // auto ack.
         channel.basicPublish("", queue, new AMQP.BasicProperties.Builder().build(), body);
-        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
+        assertTrue(latch.await(5, TimeUnit.SECONDS));
         assertTrue("consumed message body not as sent",
-                   Arrays.equals(body, delivery.getBody()));
+                   Arrays.equals(body, bodyReference.get()));
         call.recover(channel);
         assertNull("should be no message available", channel.basicGet(queue, true));
     }