Merge bug25260

Author: Simon MacMullen

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

@@ -1,6 +1,7 @@
 package com.rabbitmq.client.test.functional;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.GetResponse;
 import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.QueueingConsumer.Delivery;
@@ -16,15 +17,15 @@
 import java.util.concurrent.Callable;
 
 public class DeadLetterExchange extends BrokerTestCase {
-    private static final String DLX = "dead.letter.exchange";
-    private static final String DLX_ARG = "x-dead-letter-exchange";
-    private static final String DLX_RK_ARG = "x-dead-letter-routing-key";
-    private static final String TEST_QUEUE_NAME = "test.queue.dead.letter";
-    private static final String DLQ = "queue.dlq";
-    private static final String DLQ2 = "queue.dlq2";
-    private static final int MSG_COUNT = 10;
-    private static final int MSG_COUNT_MANY = 1000;
-    private static final int TTL = 1000;
+    public static final String DLX = "dead.letter.exchange";
+    public static final String DLX_ARG = "x-dead-letter-exchange";
+    public static final String DLX_RK_ARG = "x-dead-letter-routing-key";
+    public static final String TEST_QUEUE_NAME = "test.queue.dead.letter";
+    public static final String DLQ = "queue.dlq";
+    public static final String DLQ2 = "queue.dlq2";
+    public static final int MSG_COUNT = 10;
+    public static final int MSG_COUNT_MANY = 1000;
+    public static final int TTL = 1000;
 
     @Override
     protected void createResources() throws IOException {
@@ -348,19 +349,23 @@ private void deadLetterTest(Callable<?> deathTrigger,
 
         deathTrigger.call();
 
-        consumeN(DLQ, MSG_COUNT, new WithResponse() {
-                @SuppressWarnings("unchecked")
-                public void process(GetResponse getResponse) {
-                    Map<String, Object> headers = getResponse.getProps().getHeaders();
-                    assertNotNull(headers);
-                    ArrayList<Object> death = (ArrayList<Object>)headers.get("x-death");
-                    assertNotNull(death);
-                    assertEquals(1, death.size());
-                    assertDeathReason(death, 0, TEST_QUEUE_NAME, reason,
-                                      "amq.direct",
-                                      Arrays.asList(new String[]{"test"}));
-                }
-            });
+        consume(channel, reason);
+    }
+
+    public static void consume(final Channel channel, final String reason) throws IOException {
+        consumeN(channel, DLQ, MSG_COUNT, new WithResponse() {
+            @SuppressWarnings("unchecked")
+            public void process(GetResponse getResponse) {
+                Map<String, Object> headers = getResponse.getProps().getHeaders();
+                assertNotNull(headers);
+                ArrayList<Object> death = (ArrayList<Object>) headers.get("x-death");
+                assertNotNull(death);
+                assertEquals(1, death.size());
+                assertDeathReason(death, 0, TEST_QUEUE_NAME, reason,
+                        "amq.direct",
+                        Arrays.asList(new String[]{"test"}));
+            }
+        });
     }
 
     private void ttlTest(final long ttl) throws Exception {
@@ -455,6 +460,12 @@ private void waitUntil(long when) throws Exception {
 
     private void consumeN(String queue, int n, WithResponse withResponse)
         throws IOException
+    {
+        consumeN(channel, queue, n, withResponse);
+    }
+
+    private static void consumeN(Channel channel, String queue, int n, WithResponse withResponse)
+        throws IOException
     {
         for(int x = 0; x < n; x++) {
             GetResponse getResponse =
@@ -469,7 +480,7 @@ private void consumeN(String queue, int n, WithResponse withResponse)
     }
 
     @SuppressWarnings("unchecked")
-    private void assertDeathReason(List<Object> death, int num,
+    private static void assertDeathReason(List<Object> death, int num,
                                    String queue, String reason,
                                    String exchange, List<String> routingKeys)
     {
@@ -489,7 +500,7 @@ private void assertDeathReason(List<Object> death, int num,
     }
 
     @SuppressWarnings("unchecked")
-    private void assertDeathReason(List<Object> death, int num,
+    private static void assertDeathReason(List<Object> death, int num,
                                    String queue, String reason) {
         Map<String, Object> deathHeader =
             (Map<String, Object>)death.get(num);

test/src/com/rabbitmq/client/test/server/DeadLetterExchangeDurable.java

@@ -0,0 +1,54 @@
+package com.rabbitmq.client.test.server;
+
+import com.rabbitmq.client.MessageProperties;
+import com.rabbitmq.client.test.BrokerTestCase;
+import com.rabbitmq.client.test.functional.DeadLetterExchange;
+import com.rabbitmq.tools.Host;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+public class DeadLetterExchangeDurable extends BrokerTestCase {
+    @Override
+    protected void createResources() throws IOException {
+        Map<String, Object> args = new HashMap<String, Object>();
+        args.put("x-message-ttl", 5000);
+        args.put("x-dead-letter-exchange", DeadLetterExchange.DLX);
+
+        channel.exchangeDeclare(DeadLetterExchange.DLX, "direct", true);
+        channel.queueDeclare(DeadLetterExchange.DLQ, true, false, false, null);
+        channel.queueDeclare(DeadLetterExchange.TEST_QUEUE_NAME, true, false, false, args);
+        channel.queueBind(DeadLetterExchange.TEST_QUEUE_NAME, "amq.direct", "test");
+        channel.queueBind(DeadLetterExchange.DLQ, DeadLetterExchange.DLX, "test");
+    }
+
+    @Override
+    protected void releaseResources() throws IOException {
+        channel.exchangeDelete(DeadLetterExchange.DLX);
+        channel.queueDelete(DeadLetterExchange.DLQ);
+        channel.queueDelete(DeadLetterExchange.TEST_QUEUE_NAME);
+    }
+
+    public void testDeadLetterQueueTTLExpiredWhileDown() throws Exception {
+        for(int x = 0; x < DeadLetterExchange.MSG_COUNT; x++) {
+            channel.basicPublish("amq.direct", "test", MessageProperties.MINIMAL_PERSISTENT_BASIC, "test message".getBytes());
+        }
+
+        closeConnection();
+        Host.executeCommand("cd ../rabbitmq-test; make stop-app");
+        Thread.sleep(5000);
+        Host.executeCommand("cd ../rabbitmq-test; make start-app");
+        openConnection();
+        openChannel();
+
+        //This has the effect of waiting for the queue to complete the
+        //dead lettering. Some raciness remains though since the
+        //dead-letter publication is async so the 'consume' below may
+        //reach the dlq before all dead-lettered messages have arrived
+        //there.
+        assertNull(basicGet(DeadLetterExchange.TEST_QUEUE_NAME));
+
+        DeadLetterExchange.consume(channel, "expired");
+    }
+}

test/src/com/rabbitmq/client/test/server/ServerTests.java

@@ -30,6 +30,7 @@ public static TestSuite suite() {
     public static void add(TestSuite suite) {
         suite.addTestSuite(Permissions.class);
         suite.addTestSuite(DurableBindingLifecycle.class);
+        suite.addTestSuite(DeadLetterExchangeDurable.class);
         suite.addTestSuite(EffectVisibilityCrossNodeTest.class);
         suite.addTestSuite(ExclusiveQueueDurability.class);
         suite.addTestSuite(AbsentQueue.class);