Remove TODO

lukebakken · lukebakken · commit 08662693f7b7 · 2024-03-13T11:26:00.000-07:00
Add assertion

Remove assertion, add comment

* Add `ConfirmsAreEnabled` in `ChannelBase`

* Add `CancellationToken` to `CreateChannelAsync`

* Use `CancellationToken` argument in `OpenAsync`
diff --git a/projects/RabbitMQ.Client/PublicAPI.Unshipped.txt b/projects/RabbitMQ.Client/PublicAPI.Unshipped.txt
@@ -919,7 +919,7 @@ virtual RabbitMQ.Client.TcpClientAdapter.ReceiveTimeout.set -> void
 ~RabbitMQ.Client.IChannel.TxRollbackAsync() -> System.Threading.Tasks.Task
 ~RabbitMQ.Client.IChannel.TxSelectAsync() -> System.Threading.Tasks.Task
 ~RabbitMQ.Client.IConnection.CloseAsync(ushort reasonCode, string reasonText, System.TimeSpan timeout, bool abort, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task
-~RabbitMQ.Client.IConnection.CreateChannelAsync() -> System.Threading.Tasks.Task<RabbitMQ.Client.IChannel>
+~RabbitMQ.Client.IConnection.CreateChannelAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IChannel>
 ~RabbitMQ.Client.IConnection.UpdateSecretAsync(string newSecret, string reason) -> System.Threading.Tasks.Task
 ~RabbitMQ.Client.IConnectionFactory.CreateConnectionAsync(string clientProvidedName, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IConnection>
 ~RabbitMQ.Client.IConnectionFactory.CreateConnectionAsync(System.Collections.Generic.IEnumerable<RabbitMQ.Client.AmqpTcpEndpoint> endpoints, string clientProvidedName, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IConnection>
diff --git a/projects/RabbitMQ.Client/client/api/IConnection.cs b/projects/RabbitMQ.Client/client/api/IConnection.cs
@@ -225,13 +225,14 @@ public interface IConnection : INetworkConnection, IDisposable
         /// <param name="reasonText">A message indicating the reason for closing the connection.</param>
         /// <param name="timeout"></param>
         /// <param name="abort">Whether or not this close is an abort (ignores certain exceptions).</param>
-        /// <param name="cancellationToken"></param>
-        Task CloseAsync(ushort reasonCode, string reasonText, TimeSpan timeout, bool abort, CancellationToken cancellationToken = default);
+        /// <param name="cancellationToken">Cancellation token</param>
+        Task CloseAsync(ushort reasonCode, string reasonText, TimeSpan timeout, bool abort,
+            CancellationToken cancellationToken = default);
 
         /// <summary>
         /// Asynchronously create and return a fresh channel, session, and channel.
         /// </summary>
-        // TODO cancellation token
-        Task<IChannel> CreateChannelAsync();
+        /// <param name="cancellationToken">Cancellation token</param>
+        Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default);
     }
 }
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs
@@ -146,7 +146,7 @@ public IBasicConsumer DefaultConsumer
         public string CurrentQueue => InnerChannel.CurrentQueue;
 
         internal async Task AutomaticallyRecoverAsync(AutorecoveringConnection conn, bool recoverConsumers,
-            bool recordedEntitiesSemaphoreHeld = false)
+            bool recordedEntitiesSemaphoreHeld, CancellationToken cancellationToken)
         {
             if (false == recordedEntitiesSemaphoreHeld)
             {
@@ -156,7 +156,7 @@ internal async Task AutomaticallyRecoverAsync(AutorecoveringConnection conn, boo
             ThrowIfDisposed();
             _connection = conn;
 
-            RecoveryAwareChannel newChannel = await conn.CreateNonRecoveringChannelAsync()
+            RecoveryAwareChannel newChannel = await conn.CreateNonRecoveringChannelAsync(cancellationToken)
                 .ConfigureAwait(false);
             newChannel.TakeOver(_innerChannel);
 
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recording.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recording.cs
@@ -731,15 +731,15 @@ private void RecordChannel(in AutorecoveringChannel channel,
         }
 
         private async Task RecordChannelAsync(AutorecoveringChannel channel,
-            bool channelsSemaphoreHeld = false)
+            bool channelsSemaphoreHeld, CancellationToken cancellationToken)
         {
             if (channelsSemaphoreHeld)
             {
                 DoAddRecordedChannel(channel);
             }
             else
             {
-                await _channelsSemaphore.WaitAsync()
+                await _channelsSemaphore.WaitAsync(cancellationToken)
                     .ConfigureAwait(false);
                 try
                 {
@@ -757,6 +757,7 @@ private void DoAddRecordedChannel(AutorecoveringChannel channel)
             _channels.Add(channel);
         }
 
+        // TODO remove this unused method
         internal void DeleteRecordedChannel(in AutorecoveringChannel channel,
             bool channelsSemaphoreHeld, bool recordedEntitiesSemaphoreHeld)
         {
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recovery.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recovery.cs
@@ -184,7 +184,7 @@ await RecoverBindingsAsync(_innerConnection, recordedEntitiesSemaphoreHeld: true
                                 .ConfigureAwait(false);
 
                         }
-                        await RecoverChannelsAndItsConsumersAsync(recordedEntitiesSemaphoreHeld: true)
+                        await RecoverChannelsAndItsConsumersAsync(recordedEntitiesSemaphoreHeld: true, cancellationToken: cancellationToken)
                             .ConfigureAwait(false);
                     }
                     finally
@@ -541,7 +541,7 @@ void UpdateConsumer(string oldTag, string newTag, in RecordedConsumer consumer)
             }
         }
 
-        private async ValueTask RecoverChannelsAndItsConsumersAsync(bool recordedEntitiesSemaphoreHeld = false)
+        private async ValueTask RecoverChannelsAndItsConsumersAsync(bool recordedEntitiesSemaphoreHeld, CancellationToken cancellationToken)
         {
             if (false == recordedEntitiesSemaphoreHeld)
             {
@@ -551,7 +551,8 @@ private async ValueTask RecoverChannelsAndItsConsumersAsync(bool recordedEntitie
             foreach (AutorecoveringChannel channel in _channels)
             {
                 await channel.AutomaticallyRecoverAsync(this, _config.TopologyRecoveryEnabled,
-                    recordedEntitiesSemaphoreHeld: recordedEntitiesSemaphoreHeld)
+                    recordedEntitiesSemaphoreHeld: recordedEntitiesSemaphoreHeld,
+                    cancellationToken: cancellationToken)
                     .ConfigureAwait(false);
             }
         }
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs
@@ -175,11 +175,11 @@ public event EventHandler<RecoveringConsumerEventArgs> RecoveringConsumer
 
         public IProtocol Protocol => Endpoint.Protocol;
 
-        public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync()
+        public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync(CancellationToken cancellationToken)
         {
             ISession session = InnerConnection.CreateSession();
             var result = new RecoveryAwareChannel(_config, session);
-            return await result.OpenAsync()
+            return await result.OpenAsync(cancellationToken)
                 .ConfigureAwait(false) as RecoveryAwareChannel;
         }
 
@@ -241,13 +241,13 @@ await CloseInnerConnectionAsync()
             }
         }
 
-        public async Task<IChannel> CreateChannelAsync()
+        public async Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default)
         {
             EnsureIsOpen();
-            RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync()
+            RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(cancellationToken)
                 .ConfigureAwait(false);
             AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel);
-            await RecordChannelAsync(channel, channelsSemaphoreHeld: false)
+            await RecordChannelAsync(channel, channelsSemaphoreHeld: false, cancellationToken: cancellationToken)
                 .ConfigureAwait(false);
             return channel;
         }
diff --git a/projects/RabbitMQ.Client/client/impl/ChannelBase.cs b/projects/RabbitMQ.Client/client/impl/ChannelBase.cs
@@ -59,7 +59,7 @@ internal abstract class ChannelBase : IChannel, IRecoverable
         private readonly RpcContinuationQueue _continuationQueue = new RpcContinuationQueue();
         private readonly ManualResetEventSlim _flowControlBlock = new ManualResetEventSlim(true);
 
-        private readonly object _confirmLock = new object();
+        private object _confirmLock;
         private readonly LinkedList<ulong> _pendingDeliveryTags = new LinkedList<ulong>();
 
         private bool _onlyAcksReceived = true;
@@ -183,7 +183,6 @@ public IBasicConsumer DefaultConsumer
 
         public bool IsOpen => CloseReason is null;
 
-        // TODO add private bool for Confirm mode
         public ulong NextPublishSeqNo { get; private set; }
 
         public string CurrentQueue { get; private set; }
@@ -376,19 +375,22 @@ protected bool Enqueue(IRpcContinuation k)
             }
         }
 
-        internal async Task<IChannel> OpenAsync()
+        internal async Task<IChannel> OpenAsync(CancellationToken cancellationToken)
         {
             bool enqueued = false;
             var k = new ChannelOpenAsyncRpcContinuation(ContinuationTimeout);
 
-            await _rpcSemaphore.WaitAsync(k.CancellationToken)
+            using CancellationTokenSource lts =
+                CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, k.CancellationToken);
+
+            await _rpcSemaphore.WaitAsync(lts.Token)
                 .ConfigureAwait(false);
             try
             {
                 enqueued = Enqueue(k);
 
                 var method = new ChannelOpen();
-                await ModelSendAsync(method, k.CancellationToken)
+                await ModelSendAsync(method, lts.Token)
                     .ConfigureAwait(false);
 
                 bool result = await k;
@@ -416,6 +418,8 @@ internal void FinishClose()
             m_connectionStartCell?.TrySetResult(null);
         }
 
+        private bool ConfirmsAreEnabled => _confirmLock != null;
+
         private async Task HandleCommandAsync(IncomingCommand cmd, CancellationToken cancellationToken)
         {
             /*
@@ -475,17 +479,21 @@ private void OnChannelShutdown(ShutdownEventArgs reason)
         {
             _continuationQueue.HandleChannelShutdown(reason);
             _channelShutdownWrapper.Invoke(this, reason);
-            lock (_confirmLock)
+
+            if (ConfirmsAreEnabled)
             {
-                if (_confirmsTaskCompletionSources?.Count > 0)
+                lock (_confirmLock)
                 {
-                    var exception = new AlreadyClosedException(reason);
-                    foreach (var confirmsTaskCompletionSource in _confirmsTaskCompletionSources)
+                    if (_confirmsTaskCompletionSources?.Count > 0)
                     {
-                        confirmsTaskCompletionSource.TrySetException(exception);
-                    }
+                        var exception = new AlreadyClosedException(reason);
+                        foreach (var confirmsTaskCompletionSource in _confirmsTaskCompletionSources)
+                        {
+                            confirmsTaskCompletionSource.TrySetException(exception);
+                        }
 
-                    _confirmsTaskCompletionSources.Clear();
+                        _confirmsTaskCompletionSources.Clear();
+                    }
                 }
             }
 
@@ -581,7 +589,7 @@ protected void HandleBasicNack(in IncomingCommand cmd)
         protected void HandleAckNack(ulong deliveryTag, bool multiple, bool isNack)
         {
             // No need to do this if publisher confirms have never been enabled.
-            if (NextPublishSeqNo > 0)
+            if (ConfirmsAreEnabled)
             {
                 // let's take a lock so we can assume that deliveryTags are unique, never duplicated and always sorted
                 lock (_confirmLock)
@@ -1017,7 +1025,7 @@ await ModelSendAsync(method, k.CancellationToken)
         public async ValueTask BasicPublishAsync<TProperties>(string exchange, string routingKey, TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
             where TProperties : IReadOnlyBasicProperties, IAmqpHeader
         {
-            if (NextPublishSeqNo > 0)
+            if (ConfirmsAreEnabled)
             {
                 lock (_confirmLock)
                 {
@@ -1047,7 +1055,7 @@ public async ValueTask BasicPublishAsync<TProperties>(string exchange, string ro
             }
             catch
             {
-                if (NextPublishSeqNo > 0)
+                if (ConfirmsAreEnabled)
                 {
                     lock (_confirmLock)
                     {
@@ -1078,7 +1086,7 @@ public async void BasicPublish<TProperties>(CachedString exchange, CachedString
             TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
             where TProperties : IReadOnlyBasicProperties, IAmqpHeader
         {
-            if (NextPublishSeqNo > 0)
+            if (ConfirmsAreEnabled)
             {
                 lock (_confirmLock)
                 {
@@ -1109,7 +1117,7 @@ public async void BasicPublish<TProperties>(CachedString exchange, CachedString
             }
             catch
             {
-                if (NextPublishSeqNo > 0)
+                if (ConfirmsAreEnabled)
                 {
                     lock (_confirmLock)
                     {
@@ -1126,7 +1134,7 @@ public async ValueTask BasicPublishAsync<TProperties>(CachedString exchange, Cac
             TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
             where TProperties : IReadOnlyBasicProperties, IAmqpHeader
         {
-            if (NextPublishSeqNo > 0)
+            if (ConfirmsAreEnabled)
             {
                 lock (_confirmLock)
                 {
@@ -1157,7 +1165,7 @@ public async ValueTask BasicPublishAsync<TProperties>(CachedString exchange, Cac
             }
             catch
             {
-                if (NextPublishSeqNo > 0)
+                if (ConfirmsAreEnabled)
                 {
                     lock (_confirmLock)
                     {
@@ -1263,6 +1271,10 @@ await ModelSendAsync(method, k.CancellationToken)
                 bool result = await k;
                 Debug.Assert(result);
 
+                // Note:
+                // Non-null means confirms are enabled
+                _confirmLock = new object();
+
                 return;
             }
             finally
@@ -1747,7 +1759,7 @@ await ModelSendAsync(method, k.CancellationToken)
 
         public Task<bool> WaitForConfirmsAsync(CancellationToken token = default)
         {
-            if (NextPublishSeqNo == 0UL)
+            if (false == ConfirmsAreEnabled)
             {
                 throw new InvalidOperationException("Confirms not selected");
             }
@@ -1821,17 +1833,15 @@ public async Task WaitForConfirmsOrDieAsync(CancellationToken token = default)
                 await CloseAsync(ea, false)
                     .ConfigureAwait(false);
             }
-            catch (TaskCanceledException)
+            catch (OperationCanceledException ex)
             {
                 const string msg = "timed out waiting for acks";
-
-                var ex = new IOException(msg);
                 var ea = new ShutdownEventArgs(ShutdownInitiator.Library, Constants.ReplySuccess, msg, ex);
 
                 await CloseAsync(ea, false)
                     .ConfigureAwait(false);
 
-                throw ex;
+                throw;
             }
         }
 
diff --git a/projects/RabbitMQ.Client/client/impl/Connection.cs b/projects/RabbitMQ.Client/client/impl/Connection.cs
@@ -256,12 +256,12 @@ await CloseAsync(ea, true,
             }
         }
 
-        public Task<IChannel> CreateChannelAsync()
+        public Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default)
         {
             EnsureIsOpen();
             ISession session = CreateSession();
             var channel = new Channel(_config, session);
-            return channel.OpenAsync();
+            return channel.OpenAsync(cancellationToken);
         }
 
         internal ISession CreateSession()
@@ -285,7 +285,8 @@ internal void EnsureIsOpen()
         }
 
         ///<summary>Asynchronous API-side invocation of connection.close with timeout.</summary>
-        public Task CloseAsync(ushort reasonCode, string reasonText, TimeSpan timeout, bool abort, CancellationToken cancellationToken)
+        public Task CloseAsync(ushort reasonCode, string reasonText, TimeSpan timeout, bool abort,
+            CancellationToken cancellationToken = default)
         {
             var reason = new ShutdownEventArgs(ShutdownInitiator.Application, reasonCode, reasonText);
             return CloseAsync(reason, abort, timeout, cancellationToken);
diff --git a/projects/RabbitMQ.Client/client/impl/Frame.cs b/projects/RabbitMQ.Client/client/impl/Frame.cs
@@ -313,8 +313,14 @@ internal static bool TryReadFrame(ref ReadOnlySequence<byte> buffer, uint maxMes
                 return false;
             }
 
-            // TODO check this?
-            // buffer.IsSingleSegment;
+            /*
+             * Note:
+             * The use of buffer.Slice seems to take all segments into account, thus there appears to be no need to check IsSingleSegment
+             * Debug.Assert(buffer.IsSingleSegment);
+             * In addition, the TestBasicRoundtripConcurrentManyMessages asserts that the consumed message bodies are equivalent to
+             * the published bodies, and if there were an issue parsing frames, it would show up there for sure.
+             * https://github.com/rabbitmq/rabbitmq-dotnet-client/pull/1516#issuecomment-1991943017
+             */
 
             byte firstByte = buffer.First.Span[0];
             if (firstByte == 'A')
diff --git a/projects/RabbitMQ.Client/client/impl/SocketFrameHandler.cs b/projects/RabbitMQ.Client/client/impl/SocketFrameHandler.cs
@@ -428,7 +428,6 @@ await tcpClient.ConnectAsync(endpoint.Address, endpoint.Port, linkedTokenSource.
             {
                 if (timeoutTokenSource.Token.IsCancellationRequested)
                 {
-                    // TODO maybe do not use System.TimeoutException here
                     var timeoutException = new TimeoutException(msg, e);
                     throw new ConnectFailureException(msg, timeoutException);
                 }

Original file line number	Diff line number	Diff line change
`@@ -146,7 +146,7 @@ public IBasicConsumer DefaultConsumer`
`146`	`146`	`public string CurrentQueue => InnerChannel.CurrentQueue;`
`147`	`147`
`148`	`148`	`internal async Task AutomaticallyRecoverAsync(AutorecoveringConnection conn, bool recoverConsumers,`
`149`		`- bool recordedEntitiesSemaphoreHeld = false)`
	`149`	`+ bool recordedEntitiesSemaphoreHeld, CancellationToken cancellationToken)`
`150`	`150`	`{`
`151`	`151`	`if (false == recordedEntitiesSemaphoreHeld)`
`152`	`152`	`{`
`@@ -156,7 +156,7 @@ internal async Task AutomaticallyRecoverAsync(AutorecoveringConnection conn, boo`
`156`	`156`	`ThrowIfDisposed();`
`157`	`157`	`_connection = conn;`
`158`	`158`
`159`		`- RecoveryAwareChannel newChannel = await conn.CreateNonRecoveringChannelAsync()`
	`159`	`+ RecoveryAwareChannel newChannel = await conn.CreateNonRecoveringChannelAsync(cancellationToken)`
`160`	`160`	`.ConfigureAwait(false);`
`161`	`161`	`newChannel.TakeOver(_innerChannel);`
`162`	`162`
Original file line number	Diff line number	Diff line change
`@@ -731,15 +731,15 @@ private void RecordChannel(in AutorecoveringChannel channel,`
`731`	`731`	`}`
`732`	`732`
`733`	`733`	`private async Task RecordChannelAsync(AutorecoveringChannel channel,`
`734`		`- bool channelsSemaphoreHeld = false)`
	`734`	`+ bool channelsSemaphoreHeld, CancellationToken cancellationToken)`
`735`	`735`	`{`
`736`	`736`	`if (channelsSemaphoreHeld)`
`737`	`737`	`{`
`738`	`738`	`DoAddRecordedChannel(channel);`
`739`	`739`	`}`
`740`	`740`	`else`
`741`	`741`	`{`
`742`		`- await _channelsSemaphore.WaitAsync()`
	`742`	`+ await _channelsSemaphore.WaitAsync(cancellationToken)`
`743`	`743`	`.ConfigureAwait(false);`
`744`	`744`	`try`
`745`	`745`	`{`
`@@ -757,6 +757,7 @@ private void DoAddRecordedChannel(AutorecoveringChannel channel)`
`757`	`757`	`_channels.Add(channel);`
`758`	`758`	`}`
`759`	`759`
	`760`	`+ // TODO remove this unused method`
`760`	`761`	`internal void DeleteRecordedChannel(in AutorecoveringChannel channel,`
`761`	`762`	`bool channelsSemaphoreHeld, bool recordedEntitiesSemaphoreHeld)`
`762`	`763`	`{`
Original file line number	Diff line number	Diff line change
`@@ -184,7 +184,7 @@ await RecoverBindingsAsync(_innerConnection, recordedEntitiesSemaphoreHeld: true`
`184`	`184`	`.ConfigureAwait(false);`
`185`	`185`
`186`	`186`	`}`
`187`		`- await RecoverChannelsAndItsConsumersAsync(recordedEntitiesSemaphoreHeld: true)`
	`187`	`+ await RecoverChannelsAndItsConsumersAsync(recordedEntitiesSemaphoreHeld: true, cancellationToken: cancellationToken)`
`188`	`188`	`.ConfigureAwait(false);`
`189`	`189`	`}`
`190`	`190`	`finally`
`@@ -541,7 +541,7 @@ void UpdateConsumer(string oldTag, string newTag, in RecordedConsumer consumer)`
`541`	`541`	`}`
`542`	`542`	`}`
`543`	`543`
`544`		`- private async ValueTask RecoverChannelsAndItsConsumersAsync(bool recordedEntitiesSemaphoreHeld = false)`
	`544`	`+ private async ValueTask RecoverChannelsAndItsConsumersAsync(bool recordedEntitiesSemaphoreHeld, CancellationToken cancellationToken)`
`545`	`545`	`{`
`546`	`546`	`if (false == recordedEntitiesSemaphoreHeld)`
`547`	`547`	`{`
`@@ -551,7 +551,8 @@ private async ValueTask RecoverChannelsAndItsConsumersAsync(bool recordedEntitie`
`551`	`551`	`foreach (AutorecoveringChannel channel in _channels)`
`552`	`552`	`{`
`553`	`553`	`await channel.AutomaticallyRecoverAsync(this, _config.TopologyRecoveryEnabled,`
`554`		`- recordedEntitiesSemaphoreHeld: recordedEntitiesSemaphoreHeld)`
	`554`	`+ recordedEntitiesSemaphoreHeld: recordedEntitiesSemaphoreHeld,`
	`555`	`+ cancellationToken: cancellationToken)`
`555`	`556`	`.ConfigureAwait(false);`
`556`	`557`	`}`
`557`	`558`	`}`
Original file line number	Diff line number	Diff line change
`@@ -175,11 +175,11 @@ public event EventHandler<RecoveringConsumerEventArgs> RecoveringConsumer`
`175`	`175`
`176`	`176`	`public IProtocol Protocol => Endpoint.Protocol;`
`177`	`177`
`178`		`- public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync()`
	`178`	`+ public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync(CancellationToken cancellationToken)`
`179`	`179`	`{`
`180`	`180`	`ISession session = InnerConnection.CreateSession();`
`181`	`181`	`var result = new RecoveryAwareChannel(_config, session);`
`182`		`- return await result.OpenAsync()`
	`182`	`+ return await result.OpenAsync(cancellationToken)`
`183`	`183`	`.ConfigureAwait(false) as RecoveryAwareChannel;`
`184`	`184`	`}`
`185`	`185`
`@@ -241,13 +241,13 @@ await CloseInnerConnectionAsync()`
`241`	`241`	`}`
`242`	`242`	`}`
`243`	`243`
`244`		`- public async Task<IChannel> CreateChannelAsync()`
	`244`	`+ public async Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default)`
`245`	`245`	`{`
`246`	`246`	`EnsureIsOpen();`
`247`		`- RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync()`
	`247`	`+ RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(cancellationToken)`
`248`	`248`	`.ConfigureAwait(false);`
`249`	`249`	`AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel);`
`250`		`- await RecordChannelAsync(channel, channelsSemaphoreHeld: false)`
	`250`	`+ await RecordChannelAsync(channel, channelsSemaphoreHeld: false, cancellationToken: cancellationToken)`
`251`	`251`	`.ConfigureAwait(false);`
`252`	`252`	`return channel;`
`253`	`253`	`}`
Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,7 @@ internal abstract class ChannelBase : IChannel, IRecoverable`
`59`	`59`	`private readonly RpcContinuationQueue _continuationQueue = new RpcContinuationQueue();`
`60`	`60`	`private readonly ManualResetEventSlim _flowControlBlock = new ManualResetEventSlim(true);`
`61`	`61`
`62`		`- private readonly object _confirmLock = new object();`
	`62`	`+ private object _confirmLock;`
`63`	`63`	`private readonly LinkedList<ulong> _pendingDeliveryTags = new LinkedList<ulong>();`
`64`	`64`
`65`	`65`	`private bool _onlyAcksReceived = true;`
`@@ -183,7 +183,6 @@ public IBasicConsumer DefaultConsumer`
`183`	`183`
`184`	`184`	`public bool IsOpen => CloseReason is null;`
`185`	`185`
`186`		`- // TODO add private bool for Confirm mode`
`187`	`186`	`public ulong NextPublishSeqNo { get; private set; }`
`188`	`187`
`189`	`188`	`public string CurrentQueue { get; private set; }`
`@@ -376,19 +375,22 @@ protected bool Enqueue(IRpcContinuation k)`
`376`	`375`	`}`
`377`	`376`	`}`
`378`	`377`
`379`		`- internal async Task<IChannel> OpenAsync()`
	`378`	`+ internal async Task<IChannel> OpenAsync(CancellationToken cancellationToken)`
`380`	`379`	`{`
`381`	`380`	`bool enqueued = false;`
`382`	`381`	`var k = new ChannelOpenAsyncRpcContinuation(ContinuationTimeout);`
`383`	`382`
`384`		`- await _rpcSemaphore.WaitAsync(k.CancellationToken)`
	`383`	`+ using CancellationTokenSource lts =`
	`384`	`+ CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, k.CancellationToken);`
	`385`	`+`
	`386`	`+ await _rpcSemaphore.WaitAsync(lts.Token)`
`385`	`387`	`.ConfigureAwait(false);`
`386`	`388`	`try`
`387`	`389`	`{`
`388`	`390`	`enqueued = Enqueue(k);`
`389`	`391`
`390`	`392`	`var method = new ChannelOpen();`
`391`		`- await ModelSendAsync(method, k.CancellationToken)`
	`393`	`+ await ModelSendAsync(method, lts.Token)`
`392`	`394`	`.ConfigureAwait(false);`
`393`	`395`
`394`	`396`	`bool result = await k;`
`@@ -416,6 +418,8 @@ internal void FinishClose()`
`416`	`418`	`m_connectionStartCell?.TrySetResult(null);`
`417`	`419`	`}`
`418`	`420`
	`421`	`+ private bool ConfirmsAreEnabled => _confirmLock != null;`
	`422`	`+`
`419`	`423`	`private async Task HandleCommandAsync(IncomingCommand cmd, CancellationToken cancellationToken)`
`420`	`424`	`{`
`421`	`425`	`/*`
`@@ -475,17 +479,21 @@ private void OnChannelShutdown(ShutdownEventArgs reason)`
`475`	`479`	`{`
`476`	`480`	`_continuationQueue.HandleChannelShutdown(reason);`
`477`	`481`	`_channelShutdownWrapper.Invoke(this, reason);`
`478`		`- lock (_confirmLock)`
	`482`	`+`
	`483`	`+ if (ConfirmsAreEnabled)`
`479`	`484`	`{`
`480`		`- if (_confirmsTaskCompletionSources?.Count > 0)`
	`485`	`+ lock (_confirmLock)`
`481`	`486`	`{`
`482`		`- var exception = new AlreadyClosedException(reason);`
`483`		`- foreach (var confirmsTaskCompletionSource in _confirmsTaskCompletionSources)`
	`487`	`+ if (_confirmsTaskCompletionSources?.Count > 0)`
`484`	`488`	`{`
`485`		`- confirmsTaskCompletionSource.TrySetException(exception);`
`486`		`- }`
	`489`	`+ var exception = new AlreadyClosedException(reason);`
	`490`	`+ foreach (var confirmsTaskCompletionSource in _confirmsTaskCompletionSources)`
	`491`	`+ {`
	`492`	`+ confirmsTaskCompletionSource.TrySetException(exception);`
	`493`	`+ }`
`487`	`494`
`488`		`- _confirmsTaskCompletionSources.Clear();`
	`495`	`+ _confirmsTaskCompletionSources.Clear();`
	`496`	`+ }`
`489`	`497`	`}`
`490`	`498`	`}`
`491`	`499`
`@@ -581,7 +589,7 @@ protected void HandleBasicNack(in IncomingCommand cmd)`
`581`	`589`	`protected void HandleAckNack(ulong deliveryTag, bool multiple, bool isNack)`
`582`	`590`	`{`
`583`	`591`	`// No need to do this if publisher confirms have never been enabled.`
`584`		`- if (NextPublishSeqNo > 0)`
	`592`	`+ if (ConfirmsAreEnabled)`
`585`	`593`	`{`
`586`	`594`	`// let's take a lock so we can assume that deliveryTags are unique, never duplicated and always sorted`
`587`	`595`	`lock (_confirmLock)`
`@@ -1017,7 +1025,7 @@ await ModelSendAsync(method, k.CancellationToken)`
`1017`	`1025`	`public async ValueTask BasicPublishAsync<TProperties>(string exchange, string routingKey, TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)`
`1018`	`1026`	`where TProperties : IReadOnlyBasicProperties, IAmqpHeader`
`1019`	`1027`	`{`
`1020`		`- if (NextPublishSeqNo > 0)`
	`1028`	`+ if (ConfirmsAreEnabled)`
`1021`	`1029`	`{`
`1022`	`1030`	`lock (_confirmLock)`
`1023`	`1031`	`{`
`@@ -1047,7 +1055,7 @@ public async ValueTask BasicPublishAsync<TProperties>(string exchange, string ro`
`1047`	`1055`	`}`
`1048`	`1056`	`catch`
`1049`	`1057`	`{`
`1050`		`- if (NextPublishSeqNo > 0)`
	`1058`	`+ if (ConfirmsAreEnabled)`
`1051`	`1059`	`{`
`1052`	`1060`	`lock (_confirmLock)`
`1053`	`1061`	`{`
`@@ -1078,7 +1086,7 @@ public async void BasicPublish<TProperties>(CachedString exchange, CachedString`
`1078`	`1086`	`TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)`
`1079`	`1087`	`where TProperties : IReadOnlyBasicProperties, IAmqpHeader`
`1080`	`1088`	`{`
`1081`		`- if (NextPublishSeqNo > 0)`
	`1089`	`+ if (ConfirmsAreEnabled)`
`1082`	`1090`	`{`
`1083`	`1091`	`lock (_confirmLock)`
`1084`	`1092`	`{`
`@@ -1109,7 +1117,7 @@ public async void BasicPublish<TProperties>(CachedString exchange, CachedString`
`1109`	`1117`	`}`
`1110`	`1118`	`catch`
`1111`	`1119`	`{`
`1112`		`- if (NextPublishSeqNo > 0)`
	`1120`	`+ if (ConfirmsAreEnabled)`
`1113`	`1121`	`{`
`1114`	`1122`	`lock (_confirmLock)`
`1115`	`1123`	`{`
`@@ -1126,7 +1134,7 @@ public async ValueTask BasicPublishAsync<TProperties>(CachedString exchange, Cac`
`1126`	`1134`	`TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)`
`1127`	`1135`	`where TProperties : IReadOnlyBasicProperties, IAmqpHeader`
`1128`	`1136`	`{`
`1129`		`- if (NextPublishSeqNo > 0)`
	`1137`	`+ if (ConfirmsAreEnabled)`
`1130`	`1138`	`{`
`1131`	`1139`	`lock (_confirmLock)`
`1132`	`1140`	`{`
`@@ -1157,7 +1165,7 @@ public async ValueTask BasicPublishAsync<TProperties>(CachedString exchange, Cac`
`1157`	`1165`	`}`
`1158`	`1166`	`catch`
`1159`	`1167`	`{`
`1160`		`- if (NextPublishSeqNo > 0)`
	`1168`	`+ if (ConfirmsAreEnabled)`
`1161`	`1169`	`{`
`1162`	`1170`	`lock (_confirmLock)`
`1163`	`1171`	`{`
`@@ -1263,6 +1271,10 @@ await ModelSendAsync(method, k.CancellationToken)`
`1263`	`1271`	`bool result = await k;`
`1264`	`1272`	`Debug.Assert(result);`
`1265`	`1273`
	`1274`	`+ // Note:`
	`1275`	`+ // Non-null means confirms are enabled`
	`1276`	`+ _confirmLock = new object();`
	`1277`	`+`
`1266`	`1278`	`return;`
`1267`	`1279`	`}`
`1268`	`1280`	`finally`
`@@ -1747,7 +1759,7 @@ await ModelSendAsync(method, k.CancellationToken)`
`1747`	`1759`
`1748`	`1760`	`public Task<bool> WaitForConfirmsAsync(CancellationToken token = default)`
`1749`	`1761`	`{`
`1750`		`- if (NextPublishSeqNo == 0UL)`
	`1762`	`+ if (false == ConfirmsAreEnabled)`
`1751`	`1763`	`{`
`1752`	`1764`	`throw new InvalidOperationException("Confirms not selected");`
`1753`	`1765`	`}`
`@@ -1821,17 +1833,15 @@ public async Task WaitForConfirmsOrDieAsync(CancellationToken token = default)`
`1821`	`1833`	`await CloseAsync(ea, false)`
`1822`	`1834`	`.ConfigureAwait(false);`
`1823`	`1835`	`}`
`1824`		`- catch (TaskCanceledException)`
	`1836`	`+ catch (OperationCanceledException ex)`
`1825`	`1837`	`{`
`1826`	`1838`	`const string msg = "timed out waiting for acks";`
`1827`		`-`
`1828`		`- var ex = new IOException(msg);`
`1829`	`1839`	`var ea = new ShutdownEventArgs(ShutdownInitiator.Library, Constants.ReplySuccess, msg, ex);`
`1830`	`1840`
`1831`	`1841`	`await CloseAsync(ea, false)`
`1832`	`1842`	`.ConfigureAwait(false);`
`1833`	`1843`
`1834`		`- throw ex;`
	`1844`	`+ throw;`
`1835`	`1845`	`}`
`1836`	`1846`	`}`
`1837`	`1847`
Original file line number	Diff line number	Diff line change
`@@ -256,12 +256,12 @@ await CloseAsync(ea, true,`
`256`	`256`	`}`
`257`	`257`	`}`
`258`	`258`
`259`		`- public Task<IChannel> CreateChannelAsync()`
	`259`	`+ public Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default)`
`260`	`260`	`{`
`261`	`261`	`EnsureIsOpen();`
`262`	`262`	`ISession session = CreateSession();`
`263`	`263`	`var channel = new Channel(_config, session);`
`264`		`- return channel.OpenAsync();`
	`264`	`+ return channel.OpenAsync(cancellationToken);`
`265`	`265`	`}`
`266`	`266`
`267`	`267`	`internal ISession CreateSession()`
`@@ -285,7 +285,8 @@ internal void EnsureIsOpen()`
`285`	`285`	`}`
`286`	`286`
`287`	`287`	`///<summary>Asynchronous API-side invocation of connection.close with timeout.</summary>`
`288`		`- public Task CloseAsync(ushort reasonCode, string reasonText, TimeSpan timeout, bool abort, CancellationToken cancellationToken)`
	`288`	`+ public Task CloseAsync(ushort reasonCode, string reasonText, TimeSpan timeout, bool abort,`
	`289`	`+ CancellationToken cancellationToken = default)`
`289`	`290`	`{`
`290`	`291`	`var reason = new ShutdownEventArgs(ShutdownInitiator.Application, reasonCode, reasonText);`
`291`	`292`	`return CloseAsync(reason, abort, timeout, cancellationToken);`
Original file line number	Diff line number	Diff line change
`@@ -428,7 +428,6 @@ await tcpClient.ConnectAsync(endpoint.Address, endpoint.Port, linkedTokenSource.`
`428`	`428`	`{`
`429`	`429`	`if (timeoutTokenSource.Token.IsCancellationRequested)`
`430`	`430`	`{`
`431`		`- // TODO maybe do not use System.TimeoutException here`
`432`	`431`	`var timeoutException = new TimeoutException(msg, e);`
`433`	`432`	`throw new ConnectFailureException(msg, timeoutException);`
`434`	`433`	`}`