merge bug24180 into default

Author: rade

test/src/com/rabbitmq/client/test/BrokerTestCase.java

@@ -31,6 +31,7 @@
 import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.impl.ShutdownNotifierComponent;
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.tools.Host;
 
 public class BrokerTestCase extends TestCase
 {
@@ -81,6 +82,14 @@ protected void releaseResources()
         throws IOException
     {}
 
+    protected void restart()
+        throws IOException
+    {
+        tearDown();
+        Host.executeCommand("cd ../rabbitmq-test; make restart-app");
+        setUp();
+    }
+
     public void openConnection()
         throws IOException
     {

test/src/com/rabbitmq/client/test/ConfirmBase.java

@@ -0,0 +1,70 @@
+//  The contents of this file are subject to the Mozilla Public License
+//  Version 1.1 (the "License"); you may not use this file except in
+//  compliance with the License. You may obtain a copy of the License
+//  at http://www.mozilla.org/MPL/
+//
+//  Software distributed under the License is distributed on an "AS IS"
+//  basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+//  the License for the specific language governing rights and
+//  limitations under the License.
+//
+//  The Original Code is RabbitMQ.
+//
+//  The Initial Developer of the Original Code is VMware, Inc.
+//  Copyright (c) 2007-2011 VMware, Inc.  All rights reserved.
+//
+
+package com.rabbitmq.client.test;
+
+import com.rabbitmq.client.ConfirmListener;
+import com.rabbitmq.client.MessageProperties;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ConfirmBase extends BrokerTestCase {
+
+    protected SortedSet<Long> unconfirmedSet;
+
+    @Override
+    protected void setUp() throws IOException {
+        super.setUp();
+        unconfirmedSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
+        channel.setConfirmListener(new ConfirmListener() {
+                public void handleAck(long seqNo, boolean multiple) {
+                    if (!unconfirmedSet.contains(seqNo)) {
+                        fail("got duplicate ack: " + seqNo);
+                    }
+                    if (multiple) {
+                        unconfirmedSet.headSet(seqNo + 1).clear();
+                    } else {
+                        unconfirmedSet.remove(seqNo);
+                    }
+                }
+
+                public void handleNack(long seqNo, boolean multiple) {
+                    fail("got a nack");
+                }
+            });
+        channel.confirmSelect();
+    }
+
+    protected void waitAcks() throws InterruptedException {
+        while (unconfirmedSet.size() > 0)
+            Thread.sleep(10);
+    }
+
+    protected void publish(String exchangeName, String queueName,
+                           boolean persistent, boolean mandatory,
+                           boolean immediate)
+        throws IOException {
+        unconfirmedSet.add(channel.getNextPublishSeqNo());
+        channel.basicPublish(exchangeName, queueName, mandatory, immediate,
+                             persistent ? MessageProperties.PERSISTENT_BASIC
+                                        : MessageProperties.BASIC,
+                             "nop".getBytes());
+    }
+
+}

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

@@ -108,7 +108,7 @@ protected void doAutoDelete(boolean durable, int queues) throws IOException {
     }
   }
 
-
+  @Override
   protected void restart() throws IOException {
   }
 

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

@@ -19,38 +19,24 @@
 
 import com.rabbitmq.client.test.BrokerTestCase;
 import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.ConfirmListener;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.GetResponse;
-import com.rabbitmq.client.MessageProperties;
+import com.rabbitmq.client.test.ConfirmBase;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.Map;
-import java.util.SortedSet;
-import java.util.TreeSet;
 
-public class Confirm extends BrokerTestCase
+public class Confirm extends ConfirmBase
 {
-    final static int NUM_MESSAGES = 1000;
+    private final static int NUM_MESSAGES = 1000;
+
     private static final String TTL_ARG = "x-message-ttl";
-    private SortedSet<Long> unconfirmedSet;
 
     @Override
     protected void setUp() throws IOException {
         super.setUp();
-        unconfirmedSet =
-            Collections.synchronizedSortedSet(new TreeSet<Long>());
-        channel.setConfirmListener(new ConfirmListener() {
-                public void handleAck(long seqNo, boolean multiple) {
-                    Confirm.this.handleAck(seqNo, multiple);
-                }
-
-                public void handleNack(long seqNo, boolean multiple) {
-                    fail("got a nack");
-                }
-            });
         channel.confirmSelect();
         channel.queueDeclare("confirm-test", true, true, false, null);
         channel.basicConsume("confirm-test", true,
@@ -222,8 +208,7 @@ public void testConfirmSelect()
         ch.confirmSelect();
     }
 
-    /* Publish NUM_MESSAGES persistent messages and wait for
-     * confirmations. */
+    /* Publish NUM_MESSAGES messages and wait for confirmations. */
     public void confirmTest(String exchange, String queueName,
                             boolean persistent, boolean mandatory,
                             boolean immediate)
@@ -240,34 +225,10 @@ private void publishN(String exchangeName, String queueName,
         throws IOException
     {
         for (long i = 0; i < NUM_MESSAGES; i++) {
-            unconfirmedSet.add(channel.getNextPublishSeqNo());
             publish(exchangeName, queueName, persistent, mandatory, immediate);
         }
     }
 
-
-    private void publish(String exchangeName, String queueName,
-                         boolean persistent, boolean mandatory,
-                         boolean immediate)
-        throws IOException
-    {
-        channel.basicPublish(exchangeName, queueName, mandatory, immediate,
-                             persistent ? MessageProperties.PERSISTENT_BASIC
-                                        : MessageProperties.BASIC,
-                             "nop".getBytes());
-    }
-
-    private void handleAck(long msgSeqNo, boolean multiple) {
-        if (!unconfirmedSet.contains(msgSeqNo)) {
-            fail("got duplicate ack: " + msgSeqNo);
-        }
-        if (multiple) {
-            unconfirmedSet.headSet(msgSeqNo + 1).clear();
-        } else {
-            unconfirmedSet.remove(msgSeqNo);
-        }
-    }
-
     private void basicRejectCommon(boolean requeue)
         throws IOException
     {
@@ -280,9 +241,4 @@ private void basicRejectCommon(boolean requeue)
             channel.basicReject(dtag, requeue);
         }
     }
-
-    private void waitAcks() throws InterruptedException {
-        while (unconfirmedSet.size() > 0)
-            Thread.sleep(10);
-    }
 }

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

@@ -355,6 +355,21 @@ public void testFlow() throws IOException
         drain(c, 1);
     }
 
+    public void testLimitAndFlow() throws IOException
+    {
+        channel.basicQos(1);
+        QueueingConsumer c = new QueueingConsumer(channel);
+        declareBindConsume(c);
+        channel.flow(false);
+        fill(3);
+        drain(c, 0);
+        channel.flow(true);
+        ack(drain(c, 1), false);
+        drain(c, 1);
+        channel.basicQos(0);
+        drain(c, 1);
+    }
+
     public void testNoConsumers() throws Exception {
         String q = declareBind(channel);
         fill(1);

test/src/com/rabbitmq/client/test/server/RestartBase.java renamed to test/src/com/rabbitmq/client/test/server/MessageRecovery.java

@@ -12,25 +12,25 @@
 //
 //  The Initial Developer of the Original Code is VMware, Inc.
 //  Copyright (c) 2007-2011 VMware, Inc.  All rights reserved.
-//
-
 
 package com.rabbitmq.client.test.server;
 
-import com.rabbitmq.client.test.BrokerTestCase;
+import com.rabbitmq.client.test.ConfirmBase;
 
 import java.io.IOException;
 
-import com.rabbitmq.tools.Host;
-
-public class RestartBase extends BrokerTestCase
+public class MessageRecovery extends ConfirmBase
 {
-    protected void restart()
-        throws IOException
-    {
-        tearDown();
-        Host.executeCommand("cd ../rabbitmq-test; make restart-app");
-        setUp();
+
+    private final static String Q = "recovery-test";
+
+    public void test() throws IOException, InterruptedException {
+        channel.queueDeclare(Q, true, false, false, null);
+        publish("", Q, true, false, false);
+        waitAcks();
+        restart();
+        assertDelivered(Q, 1);
+        channel.queueDelete(Q);
     }
 
 }