fix: properly cancel token in AsyncEventingBasicConsumer on channel close

Author: Joakim Andersson

projects/RabbitMQ.Client/ConsumerDispatching/AsyncConsumerDispatcher.cs

@@ -33,8 +33,10 @@ protected override async Task ProcessChannelAsync()
                                             work.DeliveryTag, work.BasicProperties!, work.Body.Size))
                                         {
                                             await work.Consumer.HandleBasicDeliverAsync(
-                                                work.ConsumerTag!, work.DeliveryTag, work.Redelivered,
-                                                work.Exchange!, work.RoutingKey!, work.BasicProperties!, work.Body.Memory)
+                                                    work.ConsumerTag!, work.DeliveryTag, work.Redelivered,
+                                                    work.Exchange!, work.RoutingKey!, work.BasicProperties!,
+                                                    work.Body.Memory,
+                                                    work.Consumer.Channel?.ShutdownCts.Token ?? default)
                                                 .ConfigureAwait(false);
                                         }
                                         break;

projects/RabbitMQ.Client/IChannel.cs

@@ -57,6 +57,8 @@ public interface IChannel : IAsyncDisposable, IDisposable
         /// or the cause of its closure otherwise.
         /// </summary>
         ShutdownEventArgs? CloseReason { get; }
+        
+        CancellationTokenSource ShutdownCts { get; }
 
         /// <summary>Signalled when an unexpected message is delivered.</summary>
         ///

projects/RabbitMQ.Client/Impl/AutorecoveringChannel.cs

@@ -136,6 +136,7 @@ public IEnumerable<string> ConsumerTags
         public int ChannelNumber => InnerChannel.ChannelNumber;
 
         public ShutdownEventArgs? CloseReason => InnerChannel.CloseReason;
+        public CancellationTokenSource ShutdownCts => InnerChannel.ShutdownCts;
 
         public IAsyncBasicConsumer? DefaultConsumer
         {

projects/RabbitMQ.Client/Impl/Channel.cs

@@ -66,6 +66,9 @@ internal partial class Channel : IChannel, IRecoverable
         private bool _disposed;
         private int _isDisposing;
 
+        private CancellationTokenSource _shutdownCts = new CancellationTokenSource();
+        public CancellationTokenSource ShutdownCts => _shutdownCts;
+        
         public Channel(ISession session, CreateChannelOptions createChannelOptions)
         {
             ContinuationTimeout = createChannelOptions.ContinuationTimeout;
@@ -208,6 +211,8 @@ public Task CloseAsync(ushort replyCode, string replyText, bool abort,
         public async Task CloseAsync(ShutdownEventArgs args, bool abort,
             CancellationToken cancellationToken)
         {
+            _shutdownCts.Cancel();
+            
             bool enqueued = false;
             var k = new ChannelCloseAsyncRpcContinuation(ContinuationTimeout, cancellationToken);