Allow semaphore to be awaited and cancelled

Author: danielmarbach

projects/RabbitMQ.Client/client/framing/Channel.cs

@@ -89,13 +89,11 @@ protected override Task<bool> DispatchCommandAsync(IncomingCommand cmd, Cancella
                     }
                 case ProtocolCommandId.BasicAck:
                     {
-                        HandleBasicAck(cmd);
-                        return Task.FromResult(true);
+                        return HandleBasicAck(cmd, cancellationToken);
                     }
                 case ProtocolCommandId.BasicNack:
                     {
-                        HandleBasicNack(cmd);
-                        return Task.FromResult(true);
+                        return HandleBasicNack(cmd, cancellationToken);
                     }
                 case ProtocolCommandId.BasicReturn:
                     {

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

@@ -589,7 +589,7 @@ public Task ConnectionTuneOkAsync(ushort channelMax, uint frameMax, ushort heart
             return ModelSendAsync(method, cancellationToken).AsTask();
         }
 
-        protected void HandleBasicAck(IncomingCommand cmd)
+        protected async Task<bool> HandleBasicAck(IncomingCommand cmd, CancellationToken cancellationToken)
         {
             var ack = new BasicAck(cmd.MethodSpan);
             if (!_basicAcksWrapper.IsEmpty)
@@ -598,10 +598,12 @@ protected void HandleBasicAck(IncomingCommand cmd)
                 _basicAcksWrapper.Invoke(this, args);
             }
 
-            HandleAckNack(ack._deliveryTag, ack._multiple, false);
+            await HandleAckNack(ack._deliveryTag, ack._multiple, false, cancellationToken)
+                .ConfigureAwait(false);
+            return true;
         }
 
-        protected void HandleBasicNack(IncomingCommand cmd)
+        protected async Task<bool> HandleBasicNack(IncomingCommand cmd, CancellationToken cancellationToken)
         {
             var nack = new BasicNack(cmd.MethodSpan);
             if (!_basicNacksWrapper.IsEmpty)
@@ -611,7 +613,9 @@ protected void HandleBasicNack(IncomingCommand cmd)
                 _basicNacksWrapper.Invoke(this, args);
             }
 
-            HandleAckNack(nack._deliveryTag, nack._multiple, true);
+            await HandleAckNack(nack._deliveryTag, nack._multiple, true, cancellationToken)
+                .ConfigureAwait(false);
+            return true;
         }
 
         protected async Task<bool> HandleBasicCancelAsync(IncomingCommand cmd, CancellationToken cancellationToken)
@@ -1829,7 +1833,7 @@ await tokenRegistration.DisposeAsync()
 
         // NOTE: this method is internal for its use in this test:
         // TestWaitForConfirmsWithTimeoutAsync_MessageNacked_WaitingHasTimedout_ReturnFalse
-        internal void HandleAckNack(ulong deliveryTag, bool multiple, bool isNack)
+        internal async Task HandleAckNack(ulong deliveryTag, bool multiple, bool isNack, CancellationToken cancellationToken = default)
         {
             // Only do this if confirms are enabled *and* the library is tracking confirmations
             if (ConfirmsAreEnabled && _trackConfirmations)
@@ -1839,7 +1843,8 @@ internal void HandleAckNack(ulong deliveryTag, bool multiple, bool isNack)
                     throw new InvalidOperationException(InternalConstants.BugFound);
                 }
                 // let's take a lock so we can assume that deliveryTags are unique, never duplicated and always sorted
-                _confirmSemaphore.Wait();
+                await _confirmSemaphore.WaitAsync(cancellationToken)
+                    .ConfigureAwait(false);
                 try
                 {
                     // No need to do anything if there are no delivery tags in the list

projects/Test/Integration/TestPublisherConfirms.cs

@@ -105,7 +105,7 @@ public Task TestWaitForConfirmsWithTimeoutAsync_MessageNacked_WaitingHasTimedout
             return TestWaitForConfirmsAsync(2000, async (ch) =>
             {
                 RecoveryAwareChannel actualChannel = ((AutorecoveringChannel)ch).InnerChannel;
-                actualChannel.HandleAckNack(10UL, false, true);
+                await actualChannel.HandleAckNack(10UL, false, true);
                 using (var cts = new CancellationTokenSource(ShortSpan))
                 {
                     Assert.False(await ch.WaitForConfirmsAsync(cts.Token));