Delete recorded consumers when autorecovering model is disposed

Fixes #1302

Author: lukebakken

projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs

@@ -612,13 +612,23 @@ public void RecordBinding(RecordedBinding rb)
             }
         }
 
-        public void RecordConsumer(string name, RecordedConsumer c)
+        public void RecordConsumer(string consumerTag, RecordedConsumer recordedConsumer)
         {
+            if (string.IsNullOrEmpty(consumerTag))
+            {
+                throw new ArgumentNullException(nameof(consumerTag));
+            }
+
+            if (recordedConsumer is null)
+            {
+                throw new ArgumentNullException(nameof(recordedConsumer));
+            }
+
             lock (_recordedEntitiesLock)
             {
-                if (!_recordedConsumers.ContainsKey(name))
+                if (!_recordedConsumers.ContainsKey(consumerTag))
                 {
-                    _recordedConsumers.Add(name, c);
+                    _recordedConsumers.Add(consumerTag, recordedConsumer);
                 }
             }
         }
@@ -651,6 +661,14 @@ public override string ToString()
 
         public void UnregisterModel(AutorecoveringModel model)
         {
+            lock (_recordedEntitiesLock)
+            {
+                foreach (string ct in model.ConsumerTags)
+                {
+                    DeleteRecordedConsumer(ct);
+                }
+            }
+
             lock (_models)
             {
                 _models.Remove(model);

projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs

@@ -43,6 +43,7 @@ internal sealed class AutorecoveringModel : IFullModel, IRecoverable
         private readonly object _eventLock = new object();
         private AutorecoveringConnection _connection;
         private RecoveryAwareModel _delegate;
+        private List<string> _recordedConsumerTags = new List<string>();
 
         private EventHandler<BasicAckEventArgs> _recordedBasicAckEventHandlers;
         private EventHandler<BasicNackEventArgs> _recordedBasicNackEventHandlers;
@@ -91,6 +92,8 @@ public TimeSpan ContinuationTimeout
             }
         }
 
+        public IEnumerable<string> ConsumerTags => _recordedConsumerTags;
+
         public AutorecoveringModel(AutorecoveringConnection conn, RecoveryAwareModel _delegate)
         {
             _connection = conn;
@@ -476,7 +479,7 @@ public void Close(ShutdownEventArgs reason, bool abort)
 
             try
             {
-                _delegate.Close(reason, abort).GetAwaiter().GetResult();;
+                _delegate.Close(reason, abort).GetAwaiter().GetResult(); ;
             }
             finally
             {
@@ -510,6 +513,8 @@ private void Dispose(bool disposing)
             {
                 Abort();
 
+                _recordedConsumerTags.Clear();
+                _recordedConsumerTags = null;
                 _connection = null;
                 _delegate = null;
                 _recordedBasicAckEventHandlers = null;
@@ -1184,16 +1189,16 @@ public string BasicConsume(
                 throw new ObjectDisposedException(GetType().FullName);
             }
 
-            string result = _delegate.BasicConsume(queue, autoAck, consumerTag, noLocal,
+            string resultConsumerTag = _delegate.BasicConsume(queue, autoAck, consumerTag, noLocal,
                 exclusive, arguments, consumer);
             RecordedConsumer rc = new RecordedConsumer(this, queue).
-                WithConsumerTag(result).
                 WithConsumer(consumer).
                 WithExclusive(exclusive).
                 WithAutoAck(autoAck).
                 WithArguments(arguments);
-            _connection.RecordConsumer(result, rc);
-            return result;
+            _connection.RecordConsumer(resultConsumerTag, rc);
+            _recordedConsumerTags.Add(resultConsumerTag);
+            return resultConsumerTag;
         }
 
         public BasicGetResult BasicGet(string queue,