Remove QueueingConsumer usages in some tests

References #213

Author: acogoluegnes

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

@@ -378,7 +378,7 @@ public void queueRecovered(String oldName, String newName) {
         for(int i = 0; i < 5000; i++) {
             String q = UUID.randomUUID().toString();
             ch.queueDeclare(q, false, false, true, null);
-            QueueingConsumer dummy = new QueueingConsumer(ch);
+            DefaultConsumer dummy = new DefaultConsumer(ch);
             String tag = ch.basicConsume(q, true, dummy);
             ch.basicCancel(tag);
         }

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

@@ -17,7 +17,6 @@
 
 import com.rabbitmq.client.*;
 import com.rabbitmq.client.AMQP.BasicProperties;
-import com.rabbitmq.client.QueueingConsumer.Delivery;
 import com.rabbitmq.client.test.BrokerTestCase;
 import org.junit.Test;
 

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

@@ -22,43 +22,41 @@
 import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 
+import com.rabbitmq.client.*;
 import org.junit.Test;
 
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.GetResponse;
-import com.rabbitmq.client.MessageProperties;
-import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.test.BrokerTestCase;
 
 public class DirectReplyTo extends BrokerTestCase {
     private static final String QUEUE = "amq.rabbitmq.reply-to";
 
     @Test public void roundTrip() throws IOException, InterruptedException {
-        QueueingConsumer c = new QueueingConsumer(channel);
+        QueueMessageConsumer c = new QueueMessageConsumer(channel);
         String replyTo = rpcFirstHalf(c);
         declare(connection, replyTo, true);
         channel.confirmSelect();
         basicPublishVolatile("response".getBytes(), "", replyTo, MessageProperties.BASIC);
         channel.waitForConfirms();
 
-        QueueingConsumer.Delivery del = c.nextDelivery();
-        assertEquals("response", new String(del.getBody()));
+        byte[] body = c.nextDelivery(10000);
+        assertEquals("response", new String(body));
     }
 
     @Test public void hack() throws IOException, InterruptedException {
-        QueueingConsumer c = new QueueingConsumer(channel);
+        QueueMessageConsumer c = new QueueMessageConsumer(channel);
         String replyTo = rpcFirstHalf(c);
         // 5 chars should overwrite part of the key but not the pid; aiming to prove
         // we can't publish using just the pid
         replyTo = replyTo.substring(0, replyTo.length() - 5) + "xxxxx";
         declare(connection, replyTo, false);
         basicPublishVolatile("response".getBytes(), "", replyTo, MessageProperties.BASIC);
 
-        QueueingConsumer.Delivery del = c.nextDelivery(500);
-        assertNull(del);
+        byte[] body = c.nextDelivery(500);
+        assertNull(body);
     }
 
     private void declare(Connection connection, String q, boolean expectedExists) throws IOException {
@@ -75,7 +73,7 @@ private void declare(Connection connection, String q, boolean expectedExists) th
     }
 
     @Test public void consumeFail() throws IOException, InterruptedException {
-        QueueingConsumer c = new QueueingConsumer(channel);
+        DefaultConsumer c = new DefaultConsumer(channel);
         Channel ch = connection.createChannel();
         try {
             ch.basicConsume(QUEUE, false, c);
@@ -95,7 +93,7 @@ private void declare(Connection connection, String q, boolean expectedExists) th
     }
 
     @Test public void consumeSuccess() throws IOException, InterruptedException {
-        QueueingConsumer c = new QueueingConsumer(channel);
+        DefaultConsumer c = new DefaultConsumer(channel);
         String ctag = channel.basicConsume(QUEUE, true, c);
         channel.basicCancel(ctag);
 
@@ -104,7 +102,7 @@ private void declare(Connection connection, String q, boolean expectedExists) th
         assertNotSame(ctag, ctag2);
     }
 
-    private String rpcFirstHalf(QueueingConsumer c) throws IOException {
+    private String rpcFirstHalf(Consumer c) throws IOException {
         channel.basicConsume(QUEUE, true, c);
         String serverQueue = channel.queueDeclare().getQueue();
         basicPublishVolatile("request".getBytes(), "", serverQueue, props());
@@ -116,4 +114,27 @@ private String rpcFirstHalf(QueueingConsumer c) throws IOException {
     private AMQP.BasicProperties props() {
         return MessageProperties.BASIC.builder().replyTo(QUEUE).build();
     }
+
+    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);
+        }
+
+    }
 }