Invalidate local exchange cache entries for auto-delete exchanges on queueUnbind

Author: michaelklishin

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

@@ -321,6 +321,7 @@ public AMQP.Queue.UnbindOk queueUnbind(String queue, String exchange, String rou
 
     public AMQP.Queue.UnbindOk queueUnbind(String queue, String exchange, String routingKey, Map<String, Object> arguments) throws IOException {
         deleteRecordedQueueBinding(queue, exchange, routingKey, arguments);
+        this.maybeDeleteRecordedAutoDeleteExchange(exchange);
         return delegate.queueUnbind(queue, exchange, routingKey, arguments);
     }
 
@@ -586,4 +587,8 @@ private RecordedConsumer deleteRecordedConsumer(String consumerTag) {
     private void maybeDeleteRecordedAutoDeleteQueue(String queue) {
         this.connection.maybeDeleteRecordedAutoDeleteQueue(queue);
     }
+
+    private void maybeDeleteRecordedAutoDeleteExchange(String exchange) {
+        this.connection.maybeDeleteRecordedAutoDeleteExchange(exchange);
+    }
 }

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

@@ -59,10 +59,10 @@ public class AutorecoveringConnection implements Connection, Recoverable, Networ
     // Records topology changes
     private final Map<String, RecordedQueue> recordedQueues = new ConcurrentHashMap<String, RecordedQueue>();
     private final List<RecordedBinding> recordedBindings = new ArrayList<RecordedBinding>();
-    private Map<String, RecordedExchange> recordedExchanges = new ConcurrentHashMap<String, RecordedExchange>();
+    private final Map<String, RecordedExchange> recordedExchanges = new ConcurrentHashMap<String, RecordedExchange>();
     private final Map<String, RecordedConsumer> consumers = new ConcurrentHashMap<String, RecordedConsumer>();
-    private List<ConsumerRecoveryListener> consumerRecoveryListeners = new ArrayList<ConsumerRecoveryListener>();
-    private List<QueueRecoveryListener> queueRecoveryListeners = new ArrayList<QueueRecoveryListener>();
+    private final List<ConsumerRecoveryListener> consumerRecoveryListeners = new ArrayList<ConsumerRecoveryListener>();
+    private final List<QueueRecoveryListener> queueRecoveryListeners = new ArrayList<QueueRecoveryListener>();
 
     public AutorecoveringConnection(ConnectionParams params, FrameHandlerFactory f, Address[] addrs) {
         this.cf = new RecoveryAwareAMQConnectionFactory(params, f, addrs);
@@ -651,12 +651,36 @@ void maybeDeleteRecordedAutoDeleteQueue(String queue) {
                     RecordedQueue q = this.recordedQueues.get(queue);
                     // last consumer on this connection is gone, remove recorded queue
                     // if it is auto-deleted. See bug 26364.
-                    if(q.isAutoDelete()) { this.recordedQueues.remove(queue); }
+                    if((q != null) && q.isAutoDelete()) { this.recordedQueues.remove(queue); }
                 }
             }
         }
     }
 
+    void maybeDeleteRecordedAutoDeleteExchange(String exchange) {
+        synchronized (this.recordedExchanges) {
+            synchronized (this.consumers) {
+                if(!hasMoreDestinationsBoundToExchange(this.recordedBindings, exchange)) {
+                    RecordedExchange x = this.recordedExchanges.get(exchange);
+                    // last binding where this exchange is the source is gone, remove recorded exchange
+                    // if it is auto-deleted. See bug 26364.
+                    if((x != null) && x.isAutoDelete()) { this.recordedExchanges.remove(exchange); }
+                }
+            }
+        }
+    }
+
+    boolean hasMoreDestinationsBoundToExchange(List<RecordedBinding> bindings, String exchange) {
+        boolean result = false;
+        for (RecordedBinding b : bindings) {
+            if(exchange.equals(b.getSource())) {
+                result = true;
+                break;
+            }
+        }
+        return result;
+    }
+
     boolean hasMoreConsumersOnQueue(Collection<RecordedConsumer> consumers, String queue) {
         boolean result = false;
         for (RecordedConsumer c : consumers) {
@@ -671,4 +695,8 @@ boolean hasMoreConsumersOnQueue(Collection<RecordedConsumer> consumers, String q
     public Map<String, RecordedQueue> getRecordedQueues() {
         return recordedQueues;
     }
+
+    public Map<String, RecordedExchange> getRecordedExchanges() {
+        return recordedExchanges;
+    }
 }

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

@@ -39,4 +39,8 @@ public RecordedExchange arguments(Map<String, Object> value) {
         this.arguments = value;
         return this;
     }
+
+    public boolean isAutoDelete() {
+        return autoDelete;
+    }
 }

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

@@ -210,7 +210,7 @@ protected void testClientNamedQueueRecoveryWith(String q, boolean noWait) throws
     public void testDeclarationOfManyAutoDeleteQueuesWithTransientConsumer() throws IOException {
         Channel ch = connection.createChannel();
         assertEquals(0, ((AutorecoveringConnection)connection).getRecordedQueues().size());
-        for(int i = 0; i < 20000; i++) {
+        for(int i = 0; i < 5000; i++) {
             String q = UUID.randomUUID().toString();
             ch.queueDeclareNoWait(q, false, false, true, null);
             QueueingConsumer dummy = new QueueingConsumer(ch);
@@ -221,6 +221,23 @@ public void testDeclarationOfManyAutoDeleteQueuesWithTransientConsumer() throws
         ch.close();
     }
 
+    // bug 26364
+    public void testDeclarationofManyAutoDeleteExchangesWithTransientQueues() 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();
+            final String rk = "doesn't matter";
+            ch.queueBind(q, x, rk);
+            ch.queueUnbind(q, x, rk);
+            ch.queueDelete(q);
+        }
+        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";