Add test for recorded binding clean-up after exchange deletion

(cherry picked from commit b451798)

Author: acogoluegnes

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

@@ -1163,25 +1163,19 @@ boolean hasMoreConsumersOnQueue(Collection<RecordedConsumer> consumers, String q
     }
 
     Set<RecordedBinding> removeBindingsWithDestination(String s) {
-        final Set<RecordedBinding> result = new LinkedHashSet<>();
-        synchronized (this.recordedBindings) {
-            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;
+        return this.removeBindingsWithCondition(b -> b.getSource().equals(s));
     }
 
     Set<RecordedBinding> removeBindingsWithSource(String s) {
+        return this.removeBindingsWithCondition(b -> b.getSource().equals(s));
+    }
+
+    private Set<RecordedBinding> removeBindingsWithCondition(Predicate<RecordedBinding> condition) {
         final Set<RecordedBinding> result = new LinkedHashSet<>();
         synchronized (this.recordedBindings) {
             for (Iterator<RecordedBinding> it = this.recordedBindings.iterator(); it.hasNext(); ) {
                 RecordedBinding b = it.next();
-                if (b.getSource().equals(s)) {
+                if (condition.test(b)) {
                     it.remove();
                     result.add(b);
                 }

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

@@ -865,6 +865,21 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
         }
     }
 
+    @Test public void thatBindingFromDeletedExchangeIsDeleted() throws IOException, InterruptedException {
+        String q = generateQueueName();
+        channel.queueDeclare(q, false, false, false, null);
+        try {
+            String x = generateExchangeName();
+            channel.exchangeDeclare(x, "fanout");
+            channel.queueBind(q, x, "");
+            assertRecordedBinding(connection, 1);
+            channel.exchangeDelete(x);
+            assertRecordedBinding(connection, 0);
+        } finally {
+            channel.queueDelete(q);
+        }
+    }
+
     private void assertConsumerCount(int exp, String q) throws IOException {
         assertThat(channel.queueDeclarePassive(q).getConsumerCount()).isEqualTo(exp);
     }
@@ -1017,4 +1032,8 @@ private static void assertRecordedQueues(Connection conn, int size) {
     private static void assertRecordedExchanges(Connection conn, int size) {
         assertThat(((AutorecoveringConnection)conn).getRecordedExchanges()).hasSize(size);
     }
+
+    private static void assertRecordedBinding(Connection conn, int size) {
+        assertThat(((AutorecoveringConnection)conn).getRecordedBindings()).hasSize(size);
+    }
 }