Merged default into amqp_0_9_1

Author: Emile Joubert

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

@@ -67,6 +67,40 @@ public void testQueuePurge() throws IOException {
         deleteExchangeAndQueue(binding);
     }
 
+    /**
+     * See bug 21854:
+     * "When Queue.Purge is called, sent-but-unacknowledged messages are no
+     * longer purged, even if the channel they were sent down is not
+     * (Tx-)transacted."
+     */
+    public void testUnackedPurge() throws IOException {
+        Binding binding = setupExchangeBindings(false);
+        channel.basicPublish(binding.x, binding.k, null, payload);
+
+        GetResponse response = channel.basicGet(binding.q, false);
+        assertFalse(response.getEnvelope().isRedeliver());
+        assertNotNull("The response SHOULD NOT BE null", response);
+
+        // If we purge the queue the unacked message should still be there on
+        // recover.
+        channel.queuePurge(binding.q);
+        response = channel.basicGet(binding.q, true);
+        assertNull("The response SHOULD BE null", response);
+
+        channel.basicRecoverAsync(true);
+        response = channel.basicGet(binding.q, false);
+        assertTrue(response.getEnvelope().isRedeliver());
+        assertNotNull("The response SHOULD NOT BE null", response);
+
+        // If we recover then purge the message should go away
+        channel.basicRecoverAsync(true);
+        channel.queuePurge(binding.q);
+        response = channel.basicGet(binding.q, true);
+        assertNull("The response SHOULD BE null", response);
+
+        deleteExchangeAndQueue(binding);
+    }
+
     /**
      * This tests whether when you delete an exchange, that any
      * bindings attached to it are deleted as well.

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

@@ -68,14 +68,11 @@ protected void releaseResources() throws IOException {
         channel.exchangeDelete(X);
     }
 
-    public void testBind()
+    public void testBindDurableToTransient()
         throws IOException
     {
-        try {
-            channel.queueBind(Q, X, "");
-            fail("Expected exception from queueBind");
-        } catch (IOException ee) {
-            // Pass!
-        }
+        channel.queueBind(Q, X, "");
+        basicPublish();
+        assertNotNull(basicGet());
     }
 }

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

@@ -35,9 +35,7 @@
 import java.util.HashMap;
 import java.io.IOException;
 
-import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.QueueingConsumer;
 
 import com.rabbitmq.client.test.BrokerTestCase;
 
@@ -50,26 +48,27 @@ public class ExchangeDeclare extends BrokerTestCase {
     public void releaseResources() throws IOException {
         channel.exchangeDelete(NAME);
     }
-    
-    public static void verifyEquivalent(Channel channel, String name, String type, boolean durable, Map<String, Object> args)
-        throws IOException {
+
+    public static void verifyEquivalent(Channel channel, String name,
+            String type, boolean durable,
+            Map<String, Object> args) throws IOException {
         channel.exchangeDeclarePassive(name);
         channel.exchangeDeclare(name, type, durable, args);
     }
 
     // Note: this will close the channel
-    public static void verifyNotEquivalent(Channel channel, String name, String type, boolean durable, Map<String, Object> args)
-        throws IOException {
+    public static void verifyNotEquivalent(Channel channel, String name,
+            String type, boolean durable,
+            Map<String, Object> args) throws IOException {
         channel.exchangeDeclarePassive(name);
         try {
             channel.exchangeDeclare(name, type, durable, args);
             fail("Exchange was supposed to be not equivalent");
-        }
-        catch (IOException ioe) {
+        } catch (IOException ioe) {
             return;
         }
     }
-    
+
     public void testExchangeNoArgsEquivalence() throws IOException {
         channel.exchangeDeclare(NAME, TYPE, false, null);
         verifyEquivalent(channel, NAME, TYPE, false, null);
@@ -91,4 +90,4 @@ public void testExchangeTypeNotEquivalent() throws IOException {
         channel.exchangeDeclare(NAME, "direct", false, null);
         verifyNotEquivalent(channel, NAME, "fanout", false, null);
     }
-}
+}

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

@@ -0,0 +1,72 @@
+//   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 Developers of the Original Code are LShift Ltd,
+//   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+//   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+//   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+//   Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+//   Ltd. Portions created by Cohesive Financial Technologies LLC are
+//   Copyright (C) 2007-2010 Cohesive Financial Technologies
+//   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+//   (C) 2007-2010 Rabbit Technologies Ltd.
+//
+//   All Rights Reserved.
+//
+//   Contributor(s): ______________________________________.
+//
+
+package com.rabbitmq.client.test.functional;
+
+import java.io.IOException;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.test.BrokerTestCase;
+
+/* Declare an exchange, bind a queue to it, then try to delete it,
+ * setting if-unused to true.  This should throw an exception. */
+public class ExchangeDeleteIfUnused extends BrokerTestCase {
+    private final static String EXCHANGE_NAME = "xchg1";
+    private final static String ROUTING_KEY = "something";
+
+    protected void createResources()
+        throws IOException
+    {
+        super.createResources();
+        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
+        String queueName = channel.queueDeclare().getQueue();
+        channel.queueBind(queueName, EXCHANGE_NAME, ROUTING_KEY);
+    }
+
+    protected void releaseResources()
+        throws IOException
+    {
+        channel.exchangeDelete(EXCHANGE_NAME);
+        super.releaseResources();
+    }
+
+    /* Attempt to Exchange.Delete(ifUnused = true) a used exchange.
+     * Should throw an exception. */
+    public void testExchangeDelete() {
+        try {
+            channel.exchangeDelete(EXCHANGE_NAME, true);
+            fail("Exception expected if exchange in use");
+        } catch (IOException e) {
+            checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
+        }
+    }
+}

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

@@ -50,11 +50,14 @@ public static TestSuite suite() {
         suite.addTestSuite(RequeueOnChannelClose.class);
         suite.addTestSuite(NoRequeueOnCancel.class);
         suite.addTestSuite(Bug20004Test.class);
+        suite.addTestSuite(ExchangeDeleteIfUnused.class);
         suite.addTestSuite(QosTests.class);
         suite.addTestSuite(AlternateExchange.class);
         suite.addTestSuite(ExchangeDeclare.class);
         suite.addTestSuite(QueueLifecycle.class);
         suite.addTestSuite(QueueExclusivity.class);
+        suite.addTestSuite(InvalidAcks.class);
+        suite.addTestSuite(InvalidAcksTx.class);
         return suite;
     }
 }

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

@@ -0,0 +1,9 @@
+package com.rabbitmq.client.test.functional;
+
+import java.io.IOException;
+
+public class InvalidAcks extends InvalidAcksBase {
+    protected void select() throws IOException {}
+    protected void commit() throws IOException {}
+}
+

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

@@ -0,0 +1,51 @@
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.GetResponse;
+import com.rabbitmq.client.test.BrokerTestCase;
+
+import java.io.IOException;
+
+/**
+ * See bug 21846:
+ * Basic.Ack is now required to signal a channel error immediately upon
+ * detecting an invalid deliveryTag, even if the channel is (Tx-)transacted.
+ *
+ * Specifically, a client MUST not acknowledge the same message more than once.
+ */
+public abstract class InvalidAcksBase extends BrokerTestCase {
+    protected abstract void select() throws IOException;
+    protected abstract void commit() throws IOException;
+
+    public void testDoubleAck()
+        throws IOException
+    {
+        select();
+        String q = channel.queueDeclare().getQueue();
+        basicPublishVolatile(q);
+        commit();
+
+        long tag = channel.basicGet(q, false).getEnvelope().getDeliveryTag();
+        channel.basicAck(tag, false);
+        channel.basicAck(tag, false);
+
+        expectChannelError(AMQP.NOT_FOUND);
+    }
+
+    private void expectChannelError(int error) {
+        try {
+            channel.queueDeclare();
+            fail("Expected channel error " + error);
+        } catch (IOException e) {
+            checkShutdownSignal(error, e);
+        }
+    }
+
+    public void testCrazyAck()
+        throws IOException
+    {
+        select();
+        channel.basicAck(123456, false);
+        expectChannelError(AMQP.NOT_FOUND);
+    }
+}

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

@@ -0,0 +1,13 @@
+package com.rabbitmq.client.test.functional;
+
+import java.io.IOException;
+
+public class InvalidAcksTx extends InvalidAcksBase {
+    protected void select() throws IOException {
+        channel.txSelect();
+    }
+
+    protected void commit() throws IOException {
+        channel.txCommit();
+    }
+}

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

@@ -31,6 +31,7 @@
 
 package com.rabbitmq.client.test.functional;
 
+import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.test.BrokerTestCase;
 import java.io.IOException;
 
@@ -227,22 +228,6 @@ public void testRollbackAcksAndReAck()
         closeChannel();
     }
 
-    /*
-      it is legal to ack the same message twice
-    */
-    public void testDuplicateAck()
-        throws IOException
-    {
-        openChannel();
-        basicPublish();
-        txSelect();
-        basicGet();
-        basicAck();
-        basicAck();
-        txCommit();
-        closeChannel();
-    }
-
     /*
       it is illegal to ack with an unknown delivery tag
     */
@@ -319,4 +304,27 @@ public void testAckAll()
         closeChannel();
     }
 
+    public void testNonTransactedCommit()
+        throws IOException
+    {
+        openChannel();
+        try {
+            txCommit();
+            fail("Expected channel error");
+        } catch (IOException e) {
+            checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
+        }
+    }
+
+    public void testNonTransactedRollback()
+        throws IOException
+    {
+        openChannel();
+        try {
+            txRollback();
+            fail("Expected channel error");
+        } catch (IOException e) {
+            checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
+        }
+    }
 }

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

@@ -46,13 +46,15 @@ public class ExchangeEquivalence extends BrokerTestCase {
 
     public void testAlternateExchangeEquivalence() throws IOException {
         channel.exchangeDeclare("alternate", "direct", false, args);
-        ExchangeDeclare.verifyEquivalent(channel, "alternate", "direct", false, args);
+        ExchangeDeclare.verifyEquivalent(channel, "alternate", "direct", false,
+                args);
     }
 
     public void testAlternateExchangeNonEquivalence() throws IOException {
         channel.exchangeDeclare("alternate", "direct", false, args);
         Map<String, Object> altargs = new HashMap<String, Object>();
         altargs.put("alternate-exchange", "somewhere");
-        ExchangeDeclare.verifyNotEquivalent(channel, "alternate", "direct", false, altargs);
+        ExchangeDeclare.verifyNotEquivalent(channel, "alternate", "direct",
+                false, altargs);
     }
-}
+}