Merge pull request #1323 from rabbitmq/rabbitmq-dotnet-client-1302-main

Port #1317 to main

Author: lukebakken

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

@@ -45,6 +45,7 @@ internal sealed class AutorecoveringChannel : IChannel, IRecoverable
         private AutorecoveringConnection _connection;
         private RecoveryAwareChannel _innerChannel;
         private bool _disposed;
+        private List<string> _recordedConsumerTags = new List<string>();
 
         private ushort _prefetchCountConsumer;
         private ushort _prefetchCountGlobal;
@@ -122,6 +123,8 @@ public event EventHandler<EventArgs> Recovery
             remove { InnerChannel.Recovery -= value; }
         }
 
+        public IEnumerable<string> ConsumerTags => _recordedConsumerTags;
+
         public int ChannelNumber => InnerChannel.ChannelNumber;
 
         public ShutdownEventArgs CloseReason => InnerChannel.CloseReason;
@@ -207,6 +210,8 @@ public void Dispose()
 
             this.Abort();
 
+            _recordedConsumerTags.Clear();
+            _recordedConsumerTags = null;
             _connection = null;
             _innerChannel = null;
             _disposed = true;
@@ -238,9 +243,12 @@ public string BasicConsume(
             IDictionary<string, object> arguments,
             IBasicConsumer consumer)
         {
-            string result = InnerChannel.BasicConsume(queue, autoAck, consumerTag, noLocal, exclusive, arguments, consumer);
-            _connection.RecordConsumer(new RecordedConsumer(this, consumer, queue, autoAck, result, exclusive, arguments));
-            return result;
+            string resultConsumerTag = InnerChannel.BasicConsume(queue, autoAck, consumerTag, noLocal, exclusive, arguments, consumer);
+            var rc = new RecordedConsumer(channel: this, consumer: consumer, consumerTag: resultConsumerTag,
+                queue: queue, autoAck: autoAck, exclusive: exclusive, arguments: arguments);
+            _connection.RecordConsumer(rc);
+            _recordedConsumerTags.Add(resultConsumerTag);
+            return resultConsumerTag;
         }
 
         public BasicGetResult BasicGet(string queue, bool autoAck)

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

@@ -248,6 +248,14 @@ private void RecordChannel(AutorecoveringChannel m)
 
         internal void DeleteRecordedChannel(AutorecoveringChannel channel)
         {
+            lock (_recordedEntitiesLock)
+            {
+                foreach (string ct in channel.ConsumerTags)
+                {
+                    DeleteRecordedConsumer(ct);
+                }
+            }
+
             lock (_channels)
             {
                 _channels.Remove(channel);

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

@@ -29,40 +29,69 @@
 //  Copyright (c) 2007-2020 VMware, Inc.  All rights reserved.
 //---------------------------------------------------------------------------
 
+using System;
 using System.Collections.Generic;
 
 namespace RabbitMQ.Client.Impl
 {
 #nullable enable
     internal readonly struct RecordedConsumer : IRecordedConsumer
     {
-        public AutorecoveringChannel Channel { get; }
-        public IBasicConsumer Consumer { get; }
-        public string Queue { get; }
-        public bool AutoAck { get; }
-        public string ConsumerTag { get; }
-        public bool Exclusive { get; }
-        public IDictionary<string, object>? Arguments { get; }
+        private readonly AutorecoveringChannel _channel;
+        private readonly IBasicConsumer _consumer;
+        private readonly string _queue;
+        private readonly bool _autoAck;
+        private readonly string _consumerTag;
+        private readonly bool _exclusive;
+        private readonly IDictionary<string, object>? _arguments;
 
-        public RecordedConsumer(AutorecoveringChannel channel, IBasicConsumer consumer, string queue, bool autoAck, string consumerTag, bool exclusive, IDictionary<string, object>? arguments)
+        public RecordedConsumer(AutorecoveringChannel channel, IBasicConsumer consumer, string consumerTag, string queue, bool autoAck, bool exclusive, IDictionary<string, object>? arguments)
         {
-            Channel = channel;
-            Consumer = consumer;
-            Queue = queue;
-            AutoAck = autoAck;
-            ConsumerTag = consumerTag;
-            Exclusive = exclusive;
-            Arguments = arguments;
+            if (channel == null)
+            {
+                throw new ArgumentNullException(nameof(channel));
+            }
+            _channel = channel;
+
+            if (consumer == null)
+            {
+                throw new ArgumentNullException(nameof(consumer));
+            }
+            _consumer = consumer;
+
+            if (string.IsNullOrEmpty(queue))
+            {
+                throw new ArgumentNullException(nameof(queue));
+            }
+            _queue = queue;
+
+            if (string.IsNullOrEmpty(consumerTag))
+            {
+                throw new ArgumentNullException(nameof(consumerTag));
+            }
+            _consumerTag = consumerTag;
+
+            _autoAck = autoAck;
+            _exclusive = exclusive;
+            _arguments = arguments;
         }
 
+        public AutorecoveringChannel Channel => _channel;
+        public IBasicConsumer Consumer => _consumer;
+        public string Queue => _queue;
+        public bool AutoAck => _autoAck;
+        public string ConsumerTag => _consumerTag;
+        public bool Exclusive => _exclusive;
+        public IDictionary<string, object>? Arguments => _arguments;
+
         public static RecordedConsumer WithNewConsumerTag(string newTag, in RecordedConsumer old)
         {
-            return new RecordedConsumer(old.Channel, old.Consumer, old.Queue, old.AutoAck, newTag, old.Exclusive, old.Arguments);
+            return new RecordedConsumer(old.Channel, old.Consumer, newTag, old.Queue, old.AutoAck, old.Exclusive, old.Arguments);
         }
 
         public static RecordedConsumer WithNewQueueNameTag(string newQueueName, in RecordedConsumer old)
         {
-            return new RecordedConsumer(old.Channel, old.Consumer, newQueueName, old.AutoAck, old.ConsumerTag, old.Exclusive, old.Arguments);
+            return new RecordedConsumer(old.Channel, old.Consumer, old.ConsumerTag, newQueueName, old.AutoAck, old.Exclusive, old.Arguments);
         }
 
         public string Recover(IChannel channel)