Clean up auto-deleted queues from recovery cache when the last (known) consumer is cancelled

This addresses a memory leak in the most common case: declare an auto-deleted queue,
add a consumer, do some work, cancel.

Of course, our cache is oblivious to other consumers that may be used, including
those on other connections, but we only invalidate our own connection-scoped cache,
not delete the queue.

Author: michaelklishin

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

@@ -367,7 +367,8 @@ public String basicConsume(String queue, boolean autoAck, String consumerTag, bo
     }
 
     public void basicCancel(String consumerTag) throws IOException {
-        this.deleteRecordedConsumer(consumerTag);
+        RecordedConsumer c = this.deleteRecordedConsumer(consumerTag);
+        this.maybeDeleteRecordedAutoDeleteQueue(c.getQueue());
         delegate.basicCancel(consumerTag);
     }
 
@@ -578,7 +579,11 @@ private void recordConsumer(String result,
         this.connection.recordConsumer(result, consumer);
     }
 
-    private void deleteRecordedConsumer(String consumerTag) {
-        this.connection.deleteRecordedConsumer(consumerTag);
+    private RecordedConsumer deleteRecordedConsumer(String consumerTag) {
+        return this.connection.deleteRecordedConsumer(consumerTag);
+    }
+
+    private void maybeDeleteRecordedAutoDeleteQueue(String queue) {
+        this.connection.maybeDeleteRecordedAutoDeleteQueue(queue);
     }
 }

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

@@ -19,6 +19,7 @@
 import java.net.ConnectException;
 import java.net.InetAddress;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -639,7 +640,35 @@ void recordConsumer(String result, RecordedConsumer consumer) {
         this.consumers.put(result, consumer);
     }
 
-    void deleteRecordedConsumer(String consumerTag) {
-        this.consumers.remove(consumerTag);
+    RecordedConsumer deleteRecordedConsumer(String consumerTag) {
+        return this.consumers.remove(consumerTag);
+    }
+
+    void maybeDeleteRecordedAutoDeleteQueue(String queue) {
+        synchronized (this.recordedQueues) {
+            synchronized (this.consumers) {
+                if(!hasMoreConsumersOnQueue(this.consumers.values(), 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); }
+                }
+            }
+        }
+    }
+
+    boolean hasMoreConsumersOnQueue(Collection<RecordedConsumer> consumers, String queue) {
+        boolean result = false;
+        for (RecordedConsumer c : consumers) {
+            if(queue.equals(c.getQueue())) {
+                result = true;
+                break;
+            }
+        }
+        return result;
+    }
+
+    public Map<String, RecordedQueue> getRecordedQueues() {
+        return recordedQueues;
     }
 }

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

@@ -32,6 +32,8 @@ public boolean isServerNamed() {
         return this.serverNamed;
     }
 
+    public boolean isAutoDelete() { return this.autoDelete; }
+
     public void recover() throws IOException {
         this.name = this.channel.queueDeclare(this.getNameToUseForRecovery(),
                                                      this.durable,

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

@@ -206,6 +206,21 @@ 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());
+        for(int i = 0; i < 20000; i++) {
+            String q = UUID.randomUUID().toString();
+            ch.queueDeclareNoWait(q, false, false, true, null);
+            QueueingConsumer dummy = new QueueingConsumer(ch);
+            String tag = ch.basicConsume(q, true, dummy);
+            ch.basicCancel(tag);
+        }
+        assertEquals(0, ((AutorecoveringConnection)connection).getRecordedQueues().size());
+        ch.close();
+    }
+
     public void testServerNamedQueueRecovery() throws IOException, InterruptedException {
         String q = channel.queueDeclare("", false, false, false, null).getQueue();
         String x = "amq.fanout";