Improve queue size limit tests

Author: Emile Joubert

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

@@ -18,6 +18,7 @@
 package com.rabbitmq.client.test.functional;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.GetResponse;
 import com.rabbitmq.client.MessageProperties;
 import com.rabbitmq.client.test.BrokerTestCase;
 import java.io.IOException;
@@ -30,47 +31,45 @@
 public class QueueSizeLimit extends BrokerTestCase {
 
     private final int MAXLENGTH = 5;
-    private final String q = "queue-maxdepth";
+    private final int MAXLENGTH1 = MAXLENGTH + 1;
+    private final String q = "queue-maxlength";
 
     @Override
     protected void setUp() throws IOException {
         super.setUp();
         channel.confirmSelect();
     }
 
-    private void setupDlx() throws IOException {
-        channel.exchangeDeclare("DLX", "fanout");
+    AMQP.BasicProperties setupDlx(boolean persistent) throws IOException {
         channel.queueDeclare("DLQ", false, true, false, null);
-        channel.queueBind("DLQ", "DLX", "test");
-    }
-
-    private void tearDownDlx() throws IOException {
-        channel.exchangeDelete("DLX");
-        channel.queueDelete("DLQ");
+        channel.queueBind("DLQ", "amq.fanout", "");
+        declareQueue(persistent, true);
+        AMQP.BasicProperties props = null;
+        if (persistent) {
+            props = MessageProperties.MINIMAL_PERSISTENT_BASIC;
+        }
+        return props;
     }
 
     public void testQueueSize() throws IOException, InterruptedException {
-        declareQueue();
-        fillUp();
-        syncPublish(null, "msg" + MAXLENGTH + 1);
-        assertEquals(MAXLENGTH, declareQueue());
-        assertHead("msg2", q);
+        declareQueue(false, false);
+        fill(false, false);
+        syncPublish(null, "msg" + MAXLENGTH1);
+        assertHead(MAXLENGTH, "msg2", q);
     }
 
     public void testQueueUnacked() throws IOException, InterruptedException {
-        declareQueue();
-        fillUnacked();
-        syncPublish(null, "msg" + MAXLENGTH + 1);
-        assertEquals(1, declareQueue());
-        assertHead("msg" + MAXLENGTH + 1, q);
+        declareQueue(false, false);
+        fill(false, true);
+        syncPublish(null, "msg" + MAXLENGTH1);
+        assertHead(1, "msg" + MAXLENGTH1, q);
     }
 
     public void testPersistent() throws IOException, InterruptedException {
-        declareQueue(true);
-        fillUnacked(true);
-        syncPublish(MessageProperties.MINIMAL_PERSISTENT_BASIC, "msg" + MAXLENGTH + 1);
-        assertEquals(1, declareQueue(true));
-        assertHead("msg" + MAXLENGTH + 1, q);
+        declareQueue(true, false);
+        fill(true, true);
+        syncPublish(MessageProperties.MINIMAL_PERSISTENT_BASIC, "msg" + MAXLENGTH1);
+        assertHead(1, "msg" + MAXLENGTH1, q);
     }
 
     public void testDlxHeadTransient() throws IOException, InterruptedException {
@@ -90,83 +89,54 @@ public void testDlxTailDurable() throws IOException, InterruptedException {
     }
 
     public void dlxHead(boolean persistent) throws IOException, InterruptedException {
-        declareQueue(persistent, true);
-        setupDlx();
-        AMQP.BasicProperties props = null;
-        if (persistent)
-            props = MessageProperties.MINIMAL_PERSISTENT_BASIC;
-        fillUp(persistent);
-        syncPublish(props, "msg" + MAXLENGTH + 1);
-        assertEquals(MAXLENGTH, declareQueue(persistent));
-        assertHead("msg1", "DLQ");
-        assertNull(channel.basicGet("DLQ", true));
-        tearDownDlx();
-    }
-
-    public void dlxTail(boolean persistent) throws IOException, InterruptedException {
-        declareQueue(persistent, true);
-        setupDlx();
-        AMQP.BasicProperties props = null;
-        if (persistent)
-            props = MessageProperties.MINIMAL_PERSISTENT_BASIC;
-        fillUnacked(persistent);
-        syncPublish(props, "msg" + MAXLENGTH + 1);
-        assertNull(null, channel.basicGet("DLQ", true));
-        assertHead("msg" + MAXLENGTH + 1, q);
-        assertNull(channel.basicGet(q, true));
-        tearDownDlx();
-    }
-
-    private int declareQueue() throws IOException {
-        return declareQueue(false, false);
-    }
-
-    private int declareQueue(boolean durable) throws IOException {
-        return declareQueue(durable, false);
-    }
-
-    private void fillUp() throws IOException, InterruptedException {
-        fill(false, false);
-    }
-
-    private void fillUp(boolean persistent) throws IOException, InterruptedException {
+        AMQP.BasicProperties props = setupDlx(persistent);
         fill(persistent, false);
+        syncPublish(props, "msg" + MAXLENGTH1);
+        assertEquals(MAXLENGTH, declareQueue(persistent, true));
+        assertHead(1, "msg1", "DLQ");
     }
 
-    private void fillUnacked() throws IOException, InterruptedException {
-        fill(false, true);
-    }
-
-    private void fillUnacked(boolean persistent) throws IOException, InterruptedException {
+    public void dlxTail(boolean persistent) throws IOException, InterruptedException {
+        AMQP.BasicProperties props = setupDlx(persistent);
         fill(persistent, true);
+        syncPublish(props, "msg" + MAXLENGTH1);
+        assertNull(null, channel.basicGet("DLQ", true));
+        assertHead(1, "msg" + MAXLENGTH1, q);
     }
 
     private void fill(boolean persistent, boolean unAcked) throws IOException, InterruptedException {
         for (int i=1; i <= MAXLENGTH; i++){
             syncPublish(null, "msg" + i);
             if (unAcked) {
-                channel.basicGet(q, false);
-                assertEquals(0, declareQueue(persistent));
-            } else {
-                assertEquals(i, declareQueue(persistent));
+                assertNotNull(channel.basicGet(q, false));
             }
         }
+        if (unAcked) {
+            assertEquals(0, declareQueue(persistent, false));
+        } else {
+            assertEquals(MAXLENGTH, declareQueue(persistent, false));
+        }
     }
 
     private void syncPublish(AMQP.BasicProperties props, String payload) throws IOException, InterruptedException {
         channel.basicPublish("", q, props, payload.getBytes());
         channel.waitForConfirmsOrDie();
     }
 
-    private int declareQueue(boolean durable, boolean dlx) throws IOException {
+    private int declareQueue(boolean persistent, boolean dlx) throws IOException {
         Map<String, Object> args = new HashMap<String, Object>();
         args.put("x-max-length", MAXLENGTH);
-        if (dlx) args.put("x-dead-letter-exchange", "DLX");
-        AMQP.Queue.DeclareOk ok = channel.queueDeclare(q, durable, true, true, args);
+        if (dlx) {
+            args.put("x-dead-letter-exchange", "amq.fanout");
+        }
+        AMQP.Queue.DeclareOk ok = channel.queueDeclare(q, persistent, true, true, args);
         return ok.getMessageCount();
     }
 
-    private void assertHead(String expected, String queueName) throws IOException {
-        assertEquals(expected, new String(channel.basicGet(queueName, true).getBody()));
+    private void assertHead(int expectedLength, String expectedPayload, String queueName) throws IOException {
+        GetResponse head = channel.basicGet(queueName, true);
+        assertNotNull(head);
+        assertEquals(expectedPayload, new String(head.getBody()));
+        assertEquals(expectedLength, head.getMessageCount() + 1);
     }
 }