* Add cancellation token to binding recovery

Author: lukebakken

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

@@ -179,8 +179,7 @@ await RecoverExchangesAsync(_innerConnection, recordedEntitiesSemaphoreHeld: tru
                                 .ConfigureAwait(false);
                             await RecoverQueuesAsync(_innerConnection, recordedEntitiesSemaphoreHeld: true, cancellationToken)
                                 .ConfigureAwait(false);
-                            // TODO cancellation token
-                            await RecoverBindingsAsync(_innerConnection, recordedEntitiesSemaphoreHeld: true)
+                            await RecoverBindingsAsync(_innerConnection, recordedEntitiesSemaphoreHeld: true, cancellationToken)
                                 .ConfigureAwait(false);
 
                         }
@@ -418,7 +417,7 @@ void UpdateConsumerQueue(string oldName, string newName)
         }
 
         private async ValueTask RecoverBindingsAsync(IConnection connection,
-            bool recordedEntitiesSemaphoreHeld = false)
+            bool recordedEntitiesSemaphoreHeld, CancellationToken cancellationToken)
         {
             if (_disposed)
             {
@@ -434,11 +433,11 @@ private async ValueTask RecoverBindingsAsync(IConnection connection,
             {
                 try
                 {
-                    using (IChannel ch = await connection.CreateChannelAsync().ConfigureAwait(false))
+                    using (IChannel ch = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false))
                     {
-                        await binding.RecoverAsync(ch)
+                        await binding.RecoverAsync(ch, cancellationToken)
                             .ConfigureAwait(false);
-                        await ch.CloseAsync()
+                        await ch.CloseAsync(cancellationToken)
                             .ConfigureAwait(false);
                     }
                 }
@@ -450,12 +449,13 @@ await ch.CloseAsync()
                         try
                         {
                             _recordedEntitiesSemaphore.Release();
+                            // TODO maybe cancellation token
                             await _config.TopologyRecoveryExceptionHandler.BindingRecoveryExceptionHandlerAsync(binding, ex, this)
                                 .ConfigureAwait(false);
                         }
                         finally
                         {
-                            await _recordedEntitiesSemaphore.WaitAsync()
+                            await _recordedEntitiesSemaphore.WaitAsync(cancellationToken)
                                 .ConfigureAwait(false);
                         }
                     }

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

@@ -31,6 +31,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace RabbitMQ.Client.Impl
@@ -67,15 +68,17 @@ public RecordedBinding(string destination, in RecordedBinding old)
             _arguments = old._arguments;
         }
 
-        public Task RecoverAsync(IChannel channel)
+        public Task RecoverAsync(IChannel channel, CancellationToken cancellationToken)
         {
             if (_isQueueBinding)
             {
-                return channel.QueueBindAsync(_destination, _source, _routingKey, _arguments, false);
+                return channel.QueueBindAsync(_destination, _source, _routingKey, _arguments, false,
+                    cancellationToken);
             }
             else
             {
-                return channel.ExchangeBindAsync(_destination, _source, _routingKey, _arguments, false);
+                return channel.ExchangeBindAsync(_destination, _source, _routingKey, _arguments, false,
+                    cancellationToken);
             }
         }