Merge bug24725

Author: Simon MacMullen

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

@@ -54,9 +54,6 @@ protected void setUp() throws IOException {
         channel.queueDeclare("confirm-test-2", true, true, false, null);
         channel.basicConsume("confirm-test-2", true,
                              new DefaultConsumer(channel));
-        Map<String, Object> argMap =
-            Collections.singletonMap(TTL_ARG, (Object)1);
-        channel.queueDeclare("confirm-ttl", true, true, false, argMap);
         channel.queueBind("confirm-test", "amq.direct",
                           "confirm-multiple-queues");
         channel.queueBind("confirm-test-2", "amq.direct",
@@ -140,9 +137,16 @@ public void testBasicReject()
     public void testQueueTTL()
         throws IOException, InterruptedException
     {
-        publishN("", "confirm-ttl", true, false, false);
+        for (int ttl : new int[]{ 1, 0 }) {
+            Map<String, Object> argMap =
+                Collections.singletonMap(TTL_ARG, (Object)ttl);
+            channel.queueDeclare("confirm-ttl", true, true, false, argMap);
 
-        channel.waitForConfirmsOrDie();
+            publishN("", "confirm-ttl", true, false, false);
+            channel.waitForConfirmsOrDie();
+
+            channel.queueDelete("confirm-ttl");
+        }
     }
 
     public void testBasicRejectRequeue()

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

@@ -84,14 +84,11 @@ public void testDeclareQueueWithRoutingKeyButNoDeadLetterExchange()
     }
 
     public void testDeadLetterQueueTTLExpiredMessages() throws Exception {
-        Map<String, Object> args = new HashMap<String, Object>();
-        args.put("x-message-ttl", 1000);
+        ttlTest(1000);
+    }
 
-        deadLetterTest(new Runnable() {
-                public void run() {
-                    sleep(2000);
-                }
-            }, args, "expired");
+    public void testDeadLetterQueueZeroTTLExpiredMessages() throws Exception {
+        ttlTest(0);
     }
 
     public void testDeadLetterDeletedDLX() throws Exception {
@@ -309,6 +306,14 @@ public void process(GetResponse getResponse) {
             });
     }
 
+    private void ttlTest(final long ttl) throws Exception {
+        Map<String, Object> args = new HashMap<String, Object>();
+        args.put("x-message-ttl", ttl);
+        deadLetterTest(new Runnable() {
+                public void run() { sleep(ttl + 1000); }
+            }, args, "expired");
+    }
+
     private void sleep(long millis) {
         try {
             Thread.sleep(millis);

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

@@ -20,6 +20,7 @@
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.GetResponse;
 import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.QueueingConsumer.Delivery;
 import com.rabbitmq.client.test.BrokerTestCase;
 
 import java.io.IOException;
@@ -60,19 +61,18 @@ public void testCreateQueueTTLTypes() throws IOException {
         }
     }
 
-    public void testCreateQueueWithInvalidTTL() throws Exception {
+    public void testTTLAllowZero() throws Exception {
         try {
-            declareQueue(TTL_INVALID_QUEUE_NAME, "foobar");
-            fail("Should not be able to declare a queue with a non-long value for x-message-ttl");
+            declareQueue(0);
         } catch (IOException e) {
-            checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
+            fail("Should be able to declare a queue with zero for x-message-ttl");
         }
     }
 
-    public void testTTLMustBeGtZero() throws Exception {
+    public void testCreateQueueWithInvalidTTL() throws Exception {
         try {
-            declareQueue(TTL_INVALID_QUEUE_NAME, 0);
-            fail("Should not be able to declare a queue with zero for x-message-ttl");
+            declareQueue(TTL_INVALID_QUEUE_NAME, "foobar");
+            fail("Should not be able to declare a queue with a non-long value for x-message-ttl");
         } catch (IOException e) {
             checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
         }
@@ -130,7 +130,6 @@ public void testPublishAndGetWithExpiry() throws Exception {
 
         assertEquals(MSG[1], get());
         assertEquals(MSG[2], get());
-
     }
 
     /*
@@ -194,6 +193,25 @@ public void testExpiryWithRequeueAfterConsume() throws Exception {
         assertNull("Requeued message not expired", get());
     }
 
+    public void testZeroTTLDelivery() throws Exception {
+        declareAndBindQueue(0);
+
+        // when there is no consumer, message should expire
+        publish(MSG[0]);
+        assertNull(get());
+
+        // when there is a consumer, message should be delivered
+        QueueingConsumer c = new QueueingConsumer(channel);
+        channel.basicConsume(TTL_QUEUE_NAME, c);
+        publish(MSG[0]);
+        Delivery d = c.nextDelivery(100);
+        assertNotNull(d);
+
+        // requeued messages should expire
+        channel.basicReject(d.getEnvelope().getDeliveryTag(), true);
+        assertNull(c.nextDelivery(100));
+    }
+
     private String get() throws IOException {
         GetResponse response = basicGet(TTL_QUEUE_NAME);
         return response == null ? null : new String(response.getBody());