Merging bug 22867 onto amqp0_9_1

Author: Matthew Sackman

src/com/rabbitmq/client/Channel.java

@@ -214,11 +214,12 @@ void basicPublish(String exchange, String routingKey, boolean mandatory, boolean
      * @param exchange the name of the exchange
      * @param type the exchange type
      * @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
+     * @param autoDelete true if the server should delete the exchange when it is no longer in use
      * @param arguments other properties (construction arguments) for the exchange
      * @return a declaration-confirm method to indicate the exchange was successfully declared
      * @throws java.io.IOException if an error is encountered
      */
-    Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable,
+    Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,
                                        Map<String, Object> arguments) throws IOException;
 
     /**
@@ -287,7 +288,7 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      * exclusively owned by another connection.
      */
     Queue.DeclareOk queueDeclarePassive(String queue) throws IOException;
-    
+
     /**
      * Delete a queue, without regard for whether it is in use or has messages on it
      * @see com.rabbitmq.client.AMQP.Queue.Delete

src/com/rabbitmq/client/impl/ChannelN.java

@@ -421,13 +421,13 @@ public void basicPublish(String exchange, String routingKey,
 
     /** Public API - {@inheritDoc} */
     public Exchange.DeclareOk exchangeDeclare(String exchange, String type,
-                                              boolean durable,
+                                              boolean durable, boolean autoDelete,
                                               Map<String, Object> arguments)
         throws IOException
     {
         return (Exchange.DeclareOk)
             exnWrappingRpc(new Exchange.Declare(TICKET, exchange, type,
-                                                false, durable, false,
+                                                false, durable, autoDelete,
                                                 false, false, arguments)).getMethod();
     }
 
@@ -436,14 +436,14 @@ public Exchange.DeclareOk exchangeDeclare(String exchange, String type,
                                               boolean durable)
         throws IOException
     {
-        return exchangeDeclare(exchange, type, durable, null);
+        return exchangeDeclare(exchange, type, durable, false, null);
     }
 
     /** Public API - {@inheritDoc} */
     public Exchange.DeclareOk exchangeDeclare(String exchange, String type)
         throws IOException
     {
-        return exchangeDeclare(exchange, type, false, null);
+        return exchangeDeclare(exchange, type, false, false, null);
     }
 
     /** Public API - {@inheritDoc} */

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

@@ -63,7 +63,7 @@ public void testCloseOKNormallyReceived() throws Exception{
   public void testCloseWithFaultyConsumer() throws Exception{
     SpecialConnection connection = new SpecialConnection();
     Channel channel = connection.createChannel();
-    channel.exchangeDeclare("x", "direct", false, null);
+    channel.exchangeDeclare("x", "direct");
     channel.queueDeclare("q", false, false, false, null);
     channel.queueBind("q", "x", "k");
 

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

@@ -109,7 +109,7 @@ public void handleBasicReturn(int replyCode,
     protected void setupRouting(String name, String ae) throws IOException {
         Map<String, Object> args = new HashMap<String, Object>();
         if (ae != null) args.put("alternate-exchange", ae);
-        channel.exchangeDeclare(name, "direct", false, args);
+        channel.exchangeDeclare(name, "direct", false, false, args);
         channel.queueBind(name, name, name);
     }
 

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

@@ -147,6 +147,40 @@ public void testExchangeIfUnused() throws IOException {
         fail("Exchange delete should have failed");
     }
 
+    /**
+     * This tests whether the server checks that an auto_delete
+     * exchange actually deletes the bindings attached to it when it
+     * is deleted.
+     *
+     * To test this, you declare and auto_delete exchange and bind an
+     * auto_delete queue to it.
+     *
+     * Start a consumer on this queue, send a message, let it get
+     * consumed and then cancel the consumer
+     *
+     * The unsubscribe should cause the queue to auto_delete, which in
+     * turn should cause the exchange to auto_delete.
+     *
+     * Then re-declare the queue again and try to rebind it to the same exhange.
+     *
+     * Because the exchange has been auto-deleted, the bind operation
+     * should fail.
+     */
+    public void testExchangeAutoDelete() throws IOException {
+        doAutoDelete(false, 1);
+    }
+
+    /**
+     * Runs something similar to testExchangeAutoDelete, but adds
+     * different queues with the same binding to the same exchange.
+     *
+     * The difference should be that the original exchange should not
+     * get auto-deleted
+     */
+    public void testExchangeAutoDeleteManyBindings() throws IOException {
+        doAutoDelete(false, 10);
+    }
+
     /**
      *
      */

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

@@ -120,6 +120,57 @@ protected void deleteExchangeAndQueue(Binding binding) throws IOException {
     channel.exchangeDelete(binding.x);
   }
 
+  protected void doAutoDelete(boolean durable, int queues) throws IOException {
+    String[] queueNames = null;
+    Binding binding = Binding.randomBinding();
+    channel.exchangeDeclare(binding.x, "direct", durable, true, null);
+    channel.queueDeclare(binding.q, durable, false, true, null);
+    channel.queueBind(binding.q, binding.x, binding.k);
+    if (queues > 1) {
+      int j = queues - 1;
+      queueNames = new String[j];
+      for (int i = 0; i < j; i++) {
+        queueNames[i] = randomString();
+        channel.queueDeclare(queueNames[i], durable, false, false, null);
+        channel.queueBind(queueNames[i], binding.x, binding.k);
+        channel.basicConsume(queueNames[i], true, new QueueingConsumer(channel));
+      }
+    }
+    subscribeSendUnsubscribe(binding);
+    if (durable) {
+      restart();
+    }
+    if (queues > 1) {
+      for (String s : queueNames) {
+        channel.basicConsume(s, true, new QueueingConsumer(channel));
+        Binding tmp = new Binding(s, binding.x, binding.k);
+        sendUnroutable(tmp);
+      }
+    }
+    channel.queueDeclare(binding.q, durable, true, true, null);
+    // if (queues == 1): Because the exchange does not exist, this
+    // bind should fail
+    try {
+      channel.queueBind(binding.q, binding.x, binding.k);
+      sendRoutable(binding);
+    }
+    catch (Exception e) {
+      // do nothing, this is the correct behaviour
+      channel = null;
+      return;
+    }
+    if (queues == 1) {
+      deleteExchangeAndQueue(binding);
+      fail("Queue bind should have failed");
+    }
+    // Do some cleanup
+    if (queues > 1) {
+      for (String q : queueNames) {
+        channel.queueDelete(q);
+      }
+    }
+  }
+
 
   protected void restart() throws IOException {
   }
@@ -172,4 +223,26 @@ protected Binding(String q, String x, String k) {
   }
 
 
+  // A couple of tests that are common to the subclasses (which differ on
+  // whether the broker is restarted)
+
+  /**
+   *
+   * The same thing as testExchangeAutoDelete, but with durable
+   * queues.
+   *
+   * Main difference is restarting the broker to make sure that the
+   * durable queues are blasted away.
+   */
+  public void testExchangeAutoDeleteDurable() throws IOException {
+    doAutoDelete(true, 1);
+  }
+
+  /**
+   * The same thing as testExchangeAutoDeleteManyBindings, but with
+   * durable queues.
+   */
+  public void testExchangeAutoDeleteDurableManyBindings() throws IOException {
+    doAutoDelete(true, 10);
+  }
 }

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

@@ -50,44 +50,49 @@ public void releaseResources() throws IOException {
     }
 
     public static void verifyEquivalent(Channel channel, String name,
-            String type, boolean durable,
+            String type, boolean durable, boolean autoDelete,
             Map<String, Object> args) throws IOException {
         channel.exchangeDeclarePassive(name);
-        channel.exchangeDeclare(name, type, durable, args);
+        channel.exchangeDeclare(name, type, durable, autoDelete, args);
     }
 
     // Note: this will close the channel
     public static void verifyNotEquivalent(Channel channel, String name,
-            String type, boolean durable,
+            String type, boolean durable, boolean autoDelete,
             Map<String, Object> args) throws IOException {
         channel.exchangeDeclarePassive(name);
         try {
-            channel.exchangeDeclare(name, type, durable, args);
+            channel.exchangeDeclare(name, type, durable, autoDelete, args);
             fail("Exchange was supposed to be not equivalent");
         } catch (IOException ioe) {
             return;
         }
     }
 
     public void testExchangeNoArgsEquivalence() throws IOException {
-        channel.exchangeDeclare(NAME, TYPE, false, null);
-        verifyEquivalent(channel, NAME, TYPE, false, null);
+        channel.exchangeDeclare(NAME, TYPE, false, false, null);
+        verifyEquivalent(channel, NAME, TYPE, false, false, null);
     }
 
     public void testExchangeNonsenseArgsEquivalent() throws IOException {
-        channel.exchangeDeclare(NAME, TYPE, false, null);
+        channel.exchangeDeclare(NAME, TYPE, false, false, null);
         Map<String, Object> args = new HashMap<String, Object>();
         args.put("nonsensical-argument-surely-not-in-use", "foo");
-        verifyEquivalent(channel, NAME, TYPE, false, args);
+        verifyEquivalent(channel, NAME, TYPE, false, false, args);
     }
 
     public void testExchangeDurableNotEquivalent() throws IOException {
-        channel.exchangeDeclare(NAME, TYPE, false, null);
-        verifyNotEquivalent(channel, NAME, TYPE, true, null);
+        channel.exchangeDeclare(NAME, TYPE, false, false, null);
+        verifyNotEquivalent(channel, NAME, TYPE, true, false, null);
     }
 
     public void testExchangeTypeNotEquivalent() throws IOException {
-        channel.exchangeDeclare(NAME, "direct", false, null);
-        verifyNotEquivalent(channel, NAME, "fanout", false, null);
+        channel.exchangeDeclare(NAME, "direct", false, false, null);
+        verifyNotEquivalent(channel, NAME, "fanout", false, false, null);
+    }
+
+    public void testExchangeAutoDeleteNotEquivalent() throws IOException {
+        channel.exchangeDeclare(NAME, "direct", false, false, null);
+        verifyNotEquivalent(channel, NAME, "direct", false, true, null);
     }
 }

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

@@ -48,6 +48,7 @@ public static TestSuite suite() {
         suite.addTestSuite(PersistentTransactions.class);
         suite.addTestSuite(RequeueOnConnectionClose.class);
         suite.addTestSuite(RequeueOnChannelClose.class);
+        suite.addTestSuite(DurableOnTransient.class);        
         suite.addTestSuite(NoRequeueOnCancel.class);
         suite.addTestSuite(Bug20004Test.class);
         suite.addTestSuite(ExchangeDeleteIfUnused.class);

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

@@ -203,7 +203,7 @@ public void testFairness()
         //behind" - a notion of fairness somewhat short of perfect but
         //probably good enough.
         for (String q : queues) {
-            AMQP.Queue.DeclareOk ok = channel.queueDeclare(q, false, false, false, null);
+            AMQP.Queue.DeclareOk ok = channel.queueDeclarePassive(q);
             assertTrue(ok.getMessageCount() < messageCount);
         }
 
@@ -480,7 +480,7 @@ protected String declareBindConsume(Channel ch,
                                         boolean noAck)
         throws IOException
     {
-        AMQP.Queue.DeclareOk ok = ch.queueDeclare("", false, false, false, null);
+        AMQP.Queue.DeclareOk ok = ch.queueDeclare();
         String queue = ok.getQueue();
         ch.queueBind(queue, "amq.fanout", "");
         ch.basicConsume(queue, noAck, c);

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

@@ -84,7 +84,7 @@ public void testTypes() throws IOException {
 
         //sending as part of method arguments - we are relying on
         //exchange.declare ignoring the arguments table.
-        channel.exchangeDeclare("x", "direct", false, table);
+        channel.exchangeDeclare("x", "direct", false, false, table);
         channel.exchangeDelete("x");
     }