Merge pull request #1317 from rabbitmq/rabbitmq-dotnet-client-1302

Delete recorded consumers when autorecovering model is disposed

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).
+            RecordedConsumer rc = new RecordedConsumer(this, queue, resultConsumerTag).
                 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,

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

@@ -29,16 +29,41 @@
 //  Copyright (c) 2007-2020 VMware, Inc.  All rights reserved.
 //---------------------------------------------------------------------------
 
+using System;
 using System.Collections.Generic;
 
 namespace RabbitMQ.Client.Impl
 {
     internal class RecordedConsumer : IRecordedConsumer
     {
-        public RecordedConsumer(AutorecoveringModel model, string queue)
+        public RecordedConsumer(AutorecoveringModel model, string queue, string consumerTag)
         {
-            Model = model;
-            Queue = queue;
+            if (model == null)
+            {
+                throw new ArgumentNullException(nameof(model));
+            }
+            else
+            {
+                Model = model ?? throw new ArgumentNullException(nameof(model));
+            }
+
+            if (string.IsNullOrEmpty(queue))
+            {
+                throw new ArgumentNullException(nameof(consumerTag));
+            }
+            else
+            {
+                Queue = queue;
+            }
+
+            if (string.IsNullOrEmpty(consumerTag))
+            {
+                throw new ArgumentNullException(nameof(consumerTag));
+            }
+            else
+            {
+                ConsumerTag = consumerTag;
+            }
         }
 
         public AutorecoveringModel Model { get; }
@@ -72,13 +97,20 @@ public RecordedConsumer WithAutoAck(bool value)
 
         public RecordedConsumer WithConsumer(IBasicConsumer value)
         {
-            Consumer = value;
+            Consumer = value ?? throw new System.ArgumentNullException(nameof(value));
             return this;
         }
 
         public RecordedConsumer WithConsumerTag(string value)
         {
-            ConsumerTag = value;
+            if (string.IsNullOrEmpty(value))
+            {
+                throw new System.ArgumentNullException(nameof(value));
+            }
+            else
+            {
+                ConsumerTag = value;
+            }
             return this;
         }
 
@@ -90,7 +122,7 @@ public RecordedConsumer WithExclusive(bool value)
 
         public RecordedConsumer WithQueue(string value)
         {
-            Queue = value;
+            Queue = value ?? throw new System.ArgumentNullException(nameof(value));
             return this;
         }
     }