Invalidate recorded auto-delete exchanges after queue.delete and exchange.delete

 When the destination end of a binding is deleted,
 so is the binding.

Author: michaelklishin

src/com/rabbitmq/client/impl/recovery/AutorecoveringChannel.java

@@ -14,6 +14,7 @@
 import com.rabbitmq.client.ReturnListener;
 import com.rabbitmq.client.ShutdownListener;
 import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.impl.AMQImpl;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -221,11 +222,11 @@ public AMQP.Exchange.DeleteOk exchangeDelete(String exchange) throws IOException
     }
 
     public AMQP.Exchange.BindOk exchangeBind(String destination, String source, String routingKey) throws IOException {
-        return delegate.exchangeBind(destination, source, routingKey);
+        return exchangeBind(destination, source, routingKey, null);
     }
 
     public AMQP.Exchange.BindOk exchangeBind(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException {
-        AMQP.Exchange.BindOk ok = delegate.exchangeBind(destination, source, routingKey, arguments);
+        final AMQP.Exchange.BindOk ok = delegate.exchangeBind(destination, source, routingKey, arguments);
         recordExchangeBinding(destination, source, routingKey, arguments);
         return ok;
     }
@@ -241,6 +242,7 @@ public AMQP.Exchange.UnbindOk exchangeUnbind(String destination, String source,
 
     public AMQP.Exchange.UnbindOk exchangeUnbind(String destination, String source, String routingKey, Map<String, Object> arguments) throws IOException {
         deleteRecordedExchangeBinding(destination, source, routingKey, arguments);
+        this.maybeDeleteRecordedAutoDeleteExchange(source);
         return delegate.exchangeUnbind(destination, source, routingKey, arguments);
     }
 

src/com/rabbitmq/client/impl/recovery/AutorecoveringConnection.java

@@ -21,8 +21,11 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
@@ -626,6 +629,10 @@ void recordQueue(String queue, RecordedQueue meta) {
 
     void deleteRecordedQueue(String queue) {
         this.recordedQueues.remove(queue);
+        Set<RecordedBinding> xs = this.removeBindingsWithDestination(queue);
+        for (RecordedBinding b : xs) {
+            this.maybeDeleteRecordedAutoDeleteExchange(b.getSource());
+        }
     }
 
     void recordExchange(String exchange, RecordedExchange x) {
@@ -634,6 +641,10 @@ void recordExchange(String exchange, RecordedExchange x) {
 
     void deleteRecordedExchange(String exchange) {
         this.recordedExchanges.remove(exchange);
+        Set<RecordedBinding> xs = this.removeBindingsWithDestination(exchange);
+        for (RecordedBinding b : xs) {
+            this.maybeDeleteRecordedAutoDeleteExchange(b.getSource());
+        }
     }
 
     void recordConsumer(String result, RecordedConsumer consumer) {
@@ -692,6 +703,18 @@ boolean hasMoreConsumersOnQueue(Collection<RecordedConsumer> consumers, String q
         return result;
     }
 
+    Set<RecordedBinding> removeBindingsWithDestination(String s) {
+        Set<RecordedBinding> result = new HashSet<RecordedBinding>();
+        for (Iterator<RecordedBinding> it = this.recordedBindings.iterator(); it.hasNext(); ) {
+            RecordedBinding b = it.next();
+            if(b.getDestination().equals(s)) {
+                it.remove();
+                result.add(b);
+            }
+        }
+        return result;
+    }
+
     public Map<String, RecordedQueue> getRecordedQueues() {
         return recordedQueues;
     }

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

@@ -206,7 +206,6 @@ protected void testClientNamedQueueRecoveryWith(String q, boolean noWait) throws
         ch.queueDelete(q);
     }
 
-    // bug 26364
     public void testDeclarationOfManyAutoDeleteQueuesWithTransientConsumer() throws IOException {
         Channel ch = connection.createChannel();
         assertEquals(0, ((AutorecoveringConnection)connection).getRecordedQueues().size());
@@ -221,8 +220,7 @@ public void testDeclarationOfManyAutoDeleteQueuesWithTransientConsumer() throws
         ch.close();
     }
 
-    // bug 26364
-    public void testDeclarationofManyAutoDeleteExchangesWithTransientQueues() throws IOException {
+    public void testDeclarationofManyAutoDeleteExchangesWithTransientQueuesThatAreUnbound() throws IOException {
         Channel ch = connection.createChannel();
         assertEquals(0, ((AutorecoveringConnection)connection).getRecordedExchanges().size());
         for(int i = 0; i < 5000; i++) {
@@ -238,6 +236,52 @@ public void testDeclarationofManyAutoDeleteExchangesWithTransientQueues() throws
         ch.close();
     }
 
+    public void testDeclarationofManyAutoDeleteExchangesWithTransientQueuesThatAreDeleted() throws IOException {
+        Channel ch = connection.createChannel();
+        assertEquals(0, ((AutorecoveringConnection)connection).getRecordedExchanges().size());
+        for(int i = 0; i < 5000; i++) {
+            String x = UUID.randomUUID().toString();
+            ch.exchangeDeclare(x, "fanout", false, true, null);
+            String q = ch.queueDeclare().getQueue();
+            ch.queueBind(q, x, "doesn't matter");
+            ch.queueDelete(q);
+        }
+        assertEquals(0, ((AutorecoveringConnection)connection).getRecordedExchanges().size());
+        ch.close();
+    }
+
+    public void testDeclarationofManyAutoDeleteExchangesWithTransientExchangesThatAreUnbound() throws IOException {
+        Channel ch = connection.createChannel();
+        assertEquals(0, ((AutorecoveringConnection)connection).getRecordedExchanges().size());
+        for(int i = 0; i < 5000; i++) {
+            String src = UUID.randomUUID().toString();
+            String dest = UUID.randomUUID().toString();
+            ch.exchangeDeclare(src, "fanout", false, true, null);
+            ch.exchangeDeclare(dest, "fanout", false, true, null);
+            final String rk = "doesn't matter";
+            ch.exchangeBind(dest, src, rk);
+            ch.exchangeUnbind(dest, src, rk);
+            ch.exchangeDelete(dest);
+        }
+        assertEquals(0, ((AutorecoveringConnection)connection).getRecordedExchanges().size());
+        ch.close();
+    }
+
+    public void testDeclarationofManyAutoDeleteExchangesWithTransientExchangesThatAreDeleted() throws IOException {
+        Channel ch = connection.createChannel();
+        assertEquals(0, ((AutorecoveringConnection)connection).getRecordedExchanges().size());
+        for(int i = 0; i < 100; i++) {
+            String src = "src-" + UUID.randomUUID().toString();
+            String dest = "dest-" + UUID.randomUUID().toString();
+            ch.exchangeDeclare(src, "fanout", false, true, null);
+            ch.exchangeDeclare(dest, "fanout", false, true, null);
+            ch.exchangeBind(dest, src, "doesn't matter");
+            ch.exchangeDelete(dest);
+        }
+        assertEquals(0, ((AutorecoveringConnection)connection).getRecordedExchanges().size());
+        ch.close();
+    }
+
     public void testServerNamedQueueRecovery() throws IOException, InterruptedException {
         String q = channel.queueDeclare("", false, false, false, null).getQueue();
         String x = "amq.fanout";