Allow WaitForShutdown to be cancelled

Author: danielmarbach

projects/RabbitMQ.Client/ConsumerDispatching/ConsumerDispatcherChannelBase.cs

@@ -164,7 +164,7 @@ public void Quiesce()
             }
         }
 
-        public async Task WaitForShutdownAsync()
+        public async Task WaitForShutdownAsync(CancellationToken cancellationToken)
         {
             if (_disposed)
             {
@@ -183,7 +183,7 @@ public async Task WaitForShutdownAsync()
                      *
                      * await _reader.Completion.ConfigureAwait(false);
                      */
-                    await _worker
+                    await _worker.WaitAsync(cancellationToken)
                         .ConfigureAwait(false);
                 }
                 catch (AggregateException aex)

projects/RabbitMQ.Client/ConsumerDispatching/IConsumerDispatcher.cs

@@ -64,6 +64,6 @@ ValueTask HandleBasicDeliverAsync(string consumerTag,
         void Quiesce();
 
         Task ShutdownAsync(ShutdownEventArgs reason);
-        Task WaitForShutdownAsync();
+        Task WaitForShutdownAsync(CancellationToken cancellationToken);
     }
 }

projects/RabbitMQ.Client/Impl/Channel.cs

@@ -208,14 +208,10 @@ public Task CloseAsync(ushort replyCode, string replyText, bool abort,
         public async Task CloseAsync(ShutdownEventArgs args, bool abort)
         {
             CancellationToken cancellationToken = args.CancellationToken;
-            if (IsOpen)
-            {
-                // Note: we really do need to try and close this channel!
-                cancellationToken = CancellationToken.None;
-            }
 
             bool enqueued = false;
-            var k = new ChannelCloseAsyncRpcContinuation(ContinuationTimeout, cancellationToken);
+            // We should really try to clsoe the connection and therefore we don't allow this to be canceled by the user
+            var k = new ChannelCloseAsyncRpcContinuation(ContinuationTimeout, IsOpen ? CancellationToken.None : cancellationToken);
 
             await _rpcSemaphore.WaitAsync(k.CancellationToken)
                 .ConfigureAwait(false);
@@ -235,7 +231,7 @@ await ModelSendAsync(in method, k.CancellationToken)
 
                 AssertResultIsTrue(await k);
 
-                await ConsumerDispatcher.WaitForShutdownAsync()
+                await ConsumerDispatcher.WaitForShutdownAsync(cancellationToken)
                     .ConfigureAwait(false);
             }
             catch (AlreadyClosedException)