Add tests without Consumer cancel

Author: Steve Powell

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

@@ -92,7 +92,7 @@ private void publishAndGet(int count, boolean doAck)
 
     /**
      * Test we don't requeue acknowledged messages (using get)
-     * @throws Exception test
+     * @throws Exception untested
      */
     public void testNormal() throws Exception
     {
@@ -101,7 +101,7 @@ public void testNormal() throws Exception
 
     /**
      * Test we requeue unacknowledged messages (using get)
-     * @throws Exception test
+     * @throws Exception untested
      */
     public void testRequeueing() throws Exception
     {
@@ -110,7 +110,7 @@ public void testRequeueing() throws Exception
 
     /**
      * Test we requeue unacknowledged message (using consumer)
-     * @throws Exception test
+     * @throws Exception untested
      */
     public void testRequeueingConsumer() throws Exception
     {
@@ -155,7 +155,7 @@ private void publishLotsAndGet()
 
     /**
      * Test close while consuming many messages successfully requeues unacknowledged messages
-     * @throws Exception test
+     * @throws Exception untested
      */
     public void testRequeueInFlight() throws Exception
     {
@@ -166,29 +166,51 @@ public void testRequeueInFlight() throws Exception
 
     /**
      * Test close while consuming partially not acked with cancel successfully requeues unacknowledged messages
-     * @throws Exception test
+     * @throws Exception untested
      */
     public void testRequeueInFlightConsumerNoAck() throws Exception
     {
         for (int i = 0; i < 5; i++) {
-            publishLotsAndConsumeSome(false);
+            publishLotsAndConsumeSome(false, true);
         }
     }
 
     /**
      * Test close while consuming partially acked with cancel successfully requeues unacknowledged messages
-     * @throws Exception test
+     * @throws Exception untested
      */
     public void testRequeueInFlightConsumerAck() throws Exception
     {
         for (int i = 0; i < 5; i++) {
-            publishLotsAndConsumeSome(true);
+            publishLotsAndConsumeSome(true, true);
+        }
+    }
+
+    /**
+     * Test close while consuming partially not acked without cancel successfully requeues unacknowledged messages
+     * @throws Exception untested
+     */
+    public void testRequeueInFlightConsumerNoAckNoCancel() throws Exception
+    {
+        for (int i = 0; i < 5; i++) {
+            publishLotsAndConsumeSome(false, false);
+        }
+    }
+
+    /**
+     * Test close while consuming partially acked without cancel successfully requeues unacknowledged messages
+     * @throws Exception untested
+     */
+    public void testRequeueInFlightConsumerAckNoCancel() throws Exception
+    {
+        for (int i = 0; i < 5; i++) {
+            publishLotsAndConsumeSome(true, false);
         }
     }
 
     private static final int MESSAGES_TO_CONSUME = 20;
 
-    private void publishLotsAndConsumeSome(boolean ack)
+    private void publishLotsAndConsumeSome(boolean ack, boolean cancelBeforeFinish)
         throws IOException, InterruptedException, ShutdownSignalException
     {
         openConnection();
@@ -201,7 +223,7 @@ private void publishLotsAndConsumeSome(boolean ack)
         }
 
         CountDownLatch latch = new CountDownLatch(1);
-        PartialConsumer c = new PartialConsumer(channel, MESSAGES_TO_CONSUME, ack, latch);
+        PartialConsumer c = new PartialConsumer(channel, MESSAGES_TO_CONSUME, ack, latch, cancelBeforeFinish);
         channel.basicConsume(Q, c);
         latch.await();  // wait for consumer
 
@@ -228,13 +250,15 @@ private class PartialConsumer extends DefaultConsumer {
         private Channel channel;
         private CountDownLatch latch;
         private volatile boolean acknowledge;
+        private final boolean cancelBeforeFinish;
 
-        public PartialConsumer(Channel channel, int count, boolean acknowledge, CountDownLatch latch) {
+        public PartialConsumer(Channel channel, int count, boolean acknowledge, CountDownLatch latch, boolean cancelBeforeFinish) {
             super(channel);
             this.count = count;
             this.channel = channel;
             this.latch = latch;
             this.acknowledge = acknowledge;
+            this.cancelBeforeFinish = cancelBeforeFinish;
         }
 
         @Override
@@ -247,7 +271,8 @@ public void handleDelivery(String consumerTag,
             if (this.acknowledge)
                 this.channel.basicAck(envelope.getDeliveryTag(), false);
             if (--this.count == 0) {
-                this.channel.basicCancel(this.getConsumerTag());
+                if (this.cancelBeforeFinish)
+                    this.channel.basicCancel(this.getConsumerTag());
                 this.acknowledge = false; // don't acknowledge any more
                 this.latch.countDown();
             }