Pass through token into args on declaration

danielmarbach · danielmarbach · commit 75e87e483ef3 · 2024-09-19T11:18:56.000+02:00
diff --git a/projects/RabbitMQ.Client/client/events/AsyncEventArgs.cs b/projects/RabbitMQ.Client/client/events/AsyncEventArgs.cs
@@ -68,6 +68,16 @@ public AsyncEventArgs(CancellationToken cancellationToken = default)
         /// </summary>
         public CancellationToken CancellationToken { get; }
 
+        public static AsyncEventArgs CreateOrDefault(CancellationToken cancellationToken)
+        {
+            if (cancellationToken.CanBeCanceled)
+            {
+                return new AsyncEventArgs(cancellationToken);
+            }
+
+            return Empty;
+        }
+
         /// <summary>
         /// Provides a value to use with events that do not have event data.
         /// </summary>
diff --git a/projects/RabbitMQ.Client/client/framing/Channel.cs b/projects/RabbitMQ.Client/client/framing/Channel.cs
@@ -98,7 +98,7 @@ protected override Task<bool> DispatchCommandAsync(IncomingCommand cmd, Cancella
                 case ProtocolCommandId.BasicReturn:
                     {
                         // Note: always returns true
-                        return HandleBasicReturn(cmd);
+                        return HandleBasicReturn(cmd, cancellationToken);
                     }
                 case ProtocolCommandId.ChannelClose:
                     {
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs
@@ -213,7 +213,7 @@ await _connection.RecoverConsumersAsync(this, newChannel, recordedEntitiesSemaph
                         .ConfigureAwait(false);
                 }
 
-                await _innerChannel.RunRecoveryEventHandlers(this)
+                await _innerChannel.RunRecoveryEventHandlers(this, cancellationToken)
                     .ConfigureAwait(false);
 
                 return true;
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recovery.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recovery.cs
@@ -206,7 +206,7 @@ await RecoverChannelsAndItsConsumersAsync(recordedEntitiesSemaphoreHeld: true, c
 
                     ESLog.Info("Connection recovery completed");
                     ThrowIfDisposed();
-                    await _recoverySucceededAsyncWrapper.InvokeAsync(this, AsyncEventArgs.Empty)
+                    await _recoverySucceededAsyncWrapper.InvokeAsync(this, AsyncEventArgs.CreateOrDefault(cancellationToken))
                         .ConfigureAwait(false);
 
                     return true;
@@ -272,7 +272,7 @@ await maybeNewInnerConnection.OpenAsync(cancellationToken)
                 if (!_connectionRecoveryErrorAsyncWrapper.IsEmpty)
                 {
                     // Note: recordedEntities semaphore is _NOT_ held at this point
-                    await _connectionRecoveryErrorAsyncWrapper.InvokeAsync(this, new ConnectionRecoveryErrorEventArgs(e))
+                    await _connectionRecoveryErrorAsyncWrapper.InvokeAsync(this, new ConnectionRecoveryErrorEventArgs(e, cancellationToken))
                         .ConfigureAwait(false);
                 }
 
@@ -386,7 +386,7 @@ await RecordQueueAsync(new RecordedQueue(newName, recordedQueue),
                             try
                             {
                                 _recordedEntitiesSemaphore.Release();
-                                await _queueNameChangedAfterRecoveryAsyncWrapper.InvokeAsync(this, new QueueNameChangedAfterRecoveryEventArgs(oldName, newName))
+                                await _queueNameChangedAfterRecoveryAsyncWrapper.InvokeAsync(this, new QueueNameChangedAfterRecoveryEventArgs(oldName, newName, cancellationToken))
                                     .ConfigureAwait(false);
                             }
                             finally
@@ -520,7 +520,7 @@ internal async ValueTask RecoverConsumersAsync(AutorecoveringChannel channelToRe
                 try
                 {
                     _recordedEntitiesSemaphore.Release();
-                    await _recoveringConsumerAsyncWrapper.InvokeAsync(this, new RecoveringConsumerEventArgs(consumer.ConsumerTag, consumer.Arguments))
+                    await _recoveringConsumerAsyncWrapper.InvokeAsync(this, new RecoveringConsumerEventArgs(consumer.ConsumerTag, consumer.Arguments, cancellationToken))
                         .ConfigureAwait(false);
                 }
                 finally
@@ -542,7 +542,7 @@ await _recordedEntitiesSemaphore.WaitAsync(cancellationToken)
                         try
                         {
                             _recordedEntitiesSemaphore.Release();
-                            await _consumerTagChangeAfterRecoveryAsyncWrapper.InvokeAsync(this, new ConsumerTagChangedAfterRecoveryEventArgs(oldTag, newTag))
+                            await _consumerTagChangeAfterRecoveryAsyncWrapper.InvokeAsync(this, new ConsumerTagChangedAfterRecoveryEventArgs(oldTag, newTag, cancellationToken))
                                 .ConfigureAwait(false);
                         }
                         finally
diff --git a/projects/RabbitMQ.Client/client/impl/ChannelBase.cs b/projects/RabbitMQ.Client/client/impl/ChannelBase.cs
@@ -163,9 +163,9 @@ public event AsyncEventHandler<AsyncEventArgs> RecoveryAsync
 
         private AsyncEventingWrapper<AsyncEventArgs> _recoveryAsyncWrapper;
 
-        internal Task RunRecoveryEventHandlers(object sender)
+        internal Task RunRecoveryEventHandlers(object sender, CancellationToken cancellationToken)
         {
-            return _recoveryAsyncWrapper.InvokeAsync(sender, AsyncEventArgs.Empty);
+            return _recoveryAsyncWrapper.InvokeAsync(sender, AsyncEventArgs.CreateOrDefault(cancellationToken));
         }
 
         public int ChannelNumber => ((Session)Session).ChannelNumber;
@@ -494,7 +494,7 @@ await _channelShutdownAsyncWrapper.InvokeAsync(this, reason)
 
             if (ConfirmsAreEnabled)
             {
-                await _confirmSemaphore.WaitAsync()
+                await _confirmSemaphore.WaitAsync(reason.CancellationToken)
                     .ConfigureAwait(false);
                 try
                 {
@@ -582,7 +582,7 @@ protected async Task<bool> HandleBasicAck(IncomingCommand cmd, CancellationToken
             var ack = new BasicAck(cmd.MethodSpan);
             if (!_basicAcksAsyncWrapper.IsEmpty)
             {
-                var args = new BasicAckEventArgs(ack._deliveryTag, ack._multiple);
+                var args = new BasicAckEventArgs(ack._deliveryTag, ack._multiple, cancellationToken);
                 await _basicAcksAsyncWrapper.InvokeAsync(this, args)
                     .ConfigureAwait(false);
             }
@@ -598,7 +598,7 @@ protected async Task<bool> HandleBasicNack(IncomingCommand cmd, CancellationToke
             if (!_basicNacksAsyncWrapper.IsEmpty)
             {
                 var args = new BasicNackEventArgs(
-                    nack._deliveryTag, nack._multiple, nack._requeue);
+                    nack._deliveryTag, nack._multiple, nack._requeue, cancellationToken);
                 await _basicNacksAsyncWrapper.InvokeAsync(this, args)
                     .ConfigureAwait(false);
             }
@@ -641,14 +641,14 @@ protected virtual ulong AdjustDeliveryTag(ulong deliveryTag)
             return deliveryTag;
         }
 
-        protected async Task<bool> HandleBasicReturn(IncomingCommand cmd)
+        protected async Task<bool> HandleBasicReturn(IncomingCommand cmd, CancellationToken cancellationToken)
         {
             if (!_basicReturnAsyncWrapper.IsEmpty)
             {
                 var basicReturn = new BasicReturn(cmd.MethodSpan);
                 var e = new BasicReturnEventArgs(basicReturn._replyCode, basicReturn._replyText,
                     basicReturn._exchange, basicReturn._routingKey,
-                    new ReadOnlyBasicProperties(cmd.HeaderSpan), cmd.Body.Memory);
+                    new ReadOnlyBasicProperties(cmd.HeaderSpan), cmd.Body.Memory, cancellationToken);
                 await _basicReturnAsyncWrapper.InvokeAsync(this, e)
                     .ConfigureAwait(false);
             }
@@ -713,7 +713,7 @@ await ModelSendAsync(in method, cancellationToken).
 
             if (!_flowControlAsyncWrapper.IsEmpty)
             {
-                await _flowControlAsyncWrapper.InvokeAsync(this, new FlowControlEventArgs(active))
+                await _flowControlAsyncWrapper.InvokeAsync(this, new FlowControlEventArgs(active, cancellationToken))
                     .ConfigureAwait(false);
             }
 
@@ -723,7 +723,7 @@ await ModelSendAsync(in method, cancellationToken).
         protected async Task<bool> HandleConnectionBlockedAsync(IncomingCommand cmd, CancellationToken cancellationToken)
         {
             string reason = new ConnectionBlocked(cmd.MethodSpan)._reason;
-            await Session.Connection.HandleConnectionBlockedAsync(reason)
+            await Session.Connection.HandleConnectionBlockedAsync(reason, cancellationToken)
                 .ConfigureAwait(false);
             return true;
         }
@@ -801,7 +801,7 @@ await k.HandleCommandAsync(cmd)
 
         protected async Task<bool> HandleConnectionUnblockedAsync(CancellationToken cancellationToken)
         {
-            await Session.Connection.HandleConnectionUnblockedAsync()
+            await Session.Connection.HandleConnectionUnblockedAsync(cancellationToken)
                 .ConfigureAwait(false);
             return true;
         }
diff --git a/projects/RabbitMQ.Client/client/impl/Connection.Commands.cs b/projects/RabbitMQ.Client/client/impl/Connection.Commands.cs
@@ -54,20 +54,20 @@ internal void NotifyReceivedCloseOk()
             _closed = true;
         }
 
-        internal Task HandleConnectionBlockedAsync(string reason)
+        internal Task HandleConnectionBlockedAsync(string reason, CancellationToken cancellationToken)
         {
             if (!_connectionBlockedAsyncWrapper.IsEmpty)
             {
-                return _connectionBlockedAsyncWrapper.InvokeAsync(this, new ConnectionBlockedEventArgs(reason));
+                return _connectionBlockedAsyncWrapper.InvokeAsync(this, new ConnectionBlockedEventArgs(reason, cancellationToken));
             }
             return Task.CompletedTask;
         }
 
-        internal Task HandleConnectionUnblockedAsync()
+        internal Task HandleConnectionUnblockedAsync(CancellationToken cancellationToken)
         {
             if (!_connectionUnblockedAsyncWrapper.IsEmpty)
             {
-                return _connectionUnblockedAsyncWrapper.InvokeAsync(this, AsyncEventArgs.Empty);
+                return _connectionUnblockedAsyncWrapper.InvokeAsync(this, AsyncEventArgs.CreateOrDefault(cancellationToken));
             }
             return Task.CompletedTask;
         }

Original file line number	Diff line number	Diff line change
`@@ -98,7 +98,7 @@ protected override Task<bool> DispatchCommandAsync(IncomingCommand cmd, Cancella`
`98`	`98`	`case ProtocolCommandId.BasicReturn:`
`99`	`99`	`{`
`100`	`100`	`// Note: always returns true`
`101`		`- return HandleBasicReturn(cmd);`
	`101`	`+ return HandleBasicReturn(cmd, cancellationToken);`
`102`	`102`	`}`
`103`	`103`	`case ProtocolCommandId.ChannelClose:`
`104`	`104`	`{`
Original file line number	Diff line number	Diff line change
`@@ -213,7 +213,7 @@ await _connection.RecoverConsumersAsync(this, newChannel, recordedEntitiesSemaph`
`213`	`213`	`.ConfigureAwait(false);`
`214`	`214`	`}`
`215`	`215`
`216`		`- await _innerChannel.RunRecoveryEventHandlers(this)`
	`216`	`+ await _innerChannel.RunRecoveryEventHandlers(this, cancellationToken)`
`217`	`217`	`.ConfigureAwait(false);`
`218`	`218`
`219`	`219`	`return true;`
Original file line number	Diff line number	Diff line change
`@@ -206,7 +206,7 @@ await RecoverChannelsAndItsConsumersAsync(recordedEntitiesSemaphoreHeld: true, c`
`206`	`206`
`207`	`207`	`ESLog.Info("Connection recovery completed");`
`208`	`208`	`ThrowIfDisposed();`
`209`		`- await _recoverySucceededAsyncWrapper.InvokeAsync(this, AsyncEventArgs.Empty)`
	`209`	`+ await _recoverySucceededAsyncWrapper.InvokeAsync(this, AsyncEventArgs.CreateOrDefault(cancellationToken))`
`210`	`210`	`.ConfigureAwait(false);`
`211`	`211`
`212`	`212`	`return true;`
`@@ -272,7 +272,7 @@ await maybeNewInnerConnection.OpenAsync(cancellationToken)`
`272`	`272`	`if (!_connectionRecoveryErrorAsyncWrapper.IsEmpty)`
`273`	`273`	`{`
`274`	`274`	`// Note: recordedEntities semaphore is _NOT_ held at this point`
`275`		`- await _connectionRecoveryErrorAsyncWrapper.InvokeAsync(this, new ConnectionRecoveryErrorEventArgs(e))`
	`275`	`+ await _connectionRecoveryErrorAsyncWrapper.InvokeAsync(this, new ConnectionRecoveryErrorEventArgs(e, cancellationToken))`
`276`	`276`	`.ConfigureAwait(false);`
`277`	`277`	`}`
`278`	`278`
`@@ -386,7 +386,7 @@ await RecordQueueAsync(new RecordedQueue(newName, recordedQueue),`
`386`	`386`	`try`
`387`	`387`	`{`
`388`	`388`	`_recordedEntitiesSemaphore.Release();`
`389`		`- await _queueNameChangedAfterRecoveryAsyncWrapper.InvokeAsync(this, new QueueNameChangedAfterRecoveryEventArgs(oldName, newName))`
	`389`	`+ await _queueNameChangedAfterRecoveryAsyncWrapper.InvokeAsync(this, new QueueNameChangedAfterRecoveryEventArgs(oldName, newName, cancellationToken))`
`390`	`390`	`.ConfigureAwait(false);`
`391`	`391`	`}`
`392`	`392`	`finally`
`@@ -520,7 +520,7 @@ internal async ValueTask RecoverConsumersAsync(AutorecoveringChannel channelToRe`
`520`	`520`	`try`
`521`	`521`	`{`
`522`	`522`	`_recordedEntitiesSemaphore.Release();`
`523`		`- await _recoveringConsumerAsyncWrapper.InvokeAsync(this, new RecoveringConsumerEventArgs(consumer.ConsumerTag, consumer.Arguments))`
	`523`	`+ await _recoveringConsumerAsyncWrapper.InvokeAsync(this, new RecoveringConsumerEventArgs(consumer.ConsumerTag, consumer.Arguments, cancellationToken))`
`524`	`524`	`.ConfigureAwait(false);`
`525`	`525`	`}`
`526`	`526`	`finally`
`@@ -542,7 +542,7 @@ await _recordedEntitiesSemaphore.WaitAsync(cancellationToken)`
`542`	`542`	`try`
`543`	`543`	`{`
`544`	`544`	`_recordedEntitiesSemaphore.Release();`
`545`		`- await _consumerTagChangeAfterRecoveryAsyncWrapper.InvokeAsync(this, new ConsumerTagChangedAfterRecoveryEventArgs(oldTag, newTag))`
	`545`	`+ await _consumerTagChangeAfterRecoveryAsyncWrapper.InvokeAsync(this, new ConsumerTagChangedAfterRecoveryEventArgs(oldTag, newTag, cancellationToken))`
`546`	`546`	`.ConfigureAwait(false);`
`547`	`547`	`}`
`548`	`548`	`finally`
Original file line number	Diff line number	Diff line change
`@@ -163,9 +163,9 @@ public event AsyncEventHandler<AsyncEventArgs> RecoveryAsync`
`163`	`163`
`164`	`164`	`private AsyncEventingWrapper<AsyncEventArgs> _recoveryAsyncWrapper;`
`165`	`165`
`166`		`- internal Task RunRecoveryEventHandlers(object sender)`
	`166`	`+ internal Task RunRecoveryEventHandlers(object sender, CancellationToken cancellationToken)`
`167`	`167`	`{`
`168`		`- return _recoveryAsyncWrapper.InvokeAsync(sender, AsyncEventArgs.Empty);`
	`168`	`+ return _recoveryAsyncWrapper.InvokeAsync(sender, AsyncEventArgs.CreateOrDefault(cancellationToken));`
`169`	`169`	`}`
`170`	`170`
`171`	`171`	`public int ChannelNumber => ((Session)Session).ChannelNumber;`
`@@ -494,7 +494,7 @@ await _channelShutdownAsyncWrapper.InvokeAsync(this, reason)`
`494`	`494`
`495`	`495`	`if (ConfirmsAreEnabled)`
`496`	`496`	`{`
`497`		`- await _confirmSemaphore.WaitAsync()`
	`497`	`+ await _confirmSemaphore.WaitAsync(reason.CancellationToken)`
`498`	`498`	`.ConfigureAwait(false);`
`499`	`499`	`try`
`500`	`500`	`{`
`@@ -582,7 +582,7 @@ protected async Task<bool> HandleBasicAck(IncomingCommand cmd, CancellationToken`
`582`	`582`	`var ack = new BasicAck(cmd.MethodSpan);`
`583`	`583`	`if (!_basicAcksAsyncWrapper.IsEmpty)`
`584`	`584`	`{`
`585`		`- var args = new BasicAckEventArgs(ack._deliveryTag, ack._multiple);`
	`585`	`+ var args = new BasicAckEventArgs(ack._deliveryTag, ack._multiple, cancellationToken);`
`586`	`586`	`await _basicAcksAsyncWrapper.InvokeAsync(this, args)`
`587`	`587`	`.ConfigureAwait(false);`
`588`	`588`	`}`
`@@ -598,7 +598,7 @@ protected async Task<bool> HandleBasicNack(IncomingCommand cmd, CancellationToke`
`598`	`598`	`if (!_basicNacksAsyncWrapper.IsEmpty)`
`599`	`599`	`{`
`600`	`600`	`var args = new BasicNackEventArgs(`
`601`		`- nack._deliveryTag, nack._multiple, nack._requeue);`
	`601`	`+ nack._deliveryTag, nack._multiple, nack._requeue, cancellationToken);`
`602`	`602`	`await _basicNacksAsyncWrapper.InvokeAsync(this, args)`
`603`	`603`	`.ConfigureAwait(false);`
`604`	`604`	`}`
`@@ -641,14 +641,14 @@ protected virtual ulong AdjustDeliveryTag(ulong deliveryTag)`
`641`	`641`	`return deliveryTag;`
`642`	`642`	`}`
`643`	`643`
`644`		`- protected async Task<bool> HandleBasicReturn(IncomingCommand cmd)`
	`644`	`+ protected async Task<bool> HandleBasicReturn(IncomingCommand cmd, CancellationToken cancellationToken)`
`645`	`645`	`{`
`646`	`646`	`if (!_basicReturnAsyncWrapper.IsEmpty)`
`647`	`647`	`{`
`648`	`648`	`var basicReturn = new BasicReturn(cmd.MethodSpan);`
`649`	`649`	`var e = new BasicReturnEventArgs(basicReturn._replyCode, basicReturn._replyText,`
`650`	`650`	`basicReturn._exchange, basicReturn._routingKey,`
`651`		`- new ReadOnlyBasicProperties(cmd.HeaderSpan), cmd.Body.Memory);`
	`651`	`+ new ReadOnlyBasicProperties(cmd.HeaderSpan), cmd.Body.Memory, cancellationToken);`
`652`	`652`	`await _basicReturnAsyncWrapper.InvokeAsync(this, e)`
`653`	`653`	`.ConfigureAwait(false);`
`654`	`654`	`}`
`@@ -713,7 +713,7 @@ await ModelSendAsync(in method, cancellationToken).`
`713`	`713`
`714`	`714`	`if (!_flowControlAsyncWrapper.IsEmpty)`
`715`	`715`	`{`
`716`		`- await _flowControlAsyncWrapper.InvokeAsync(this, new FlowControlEventArgs(active))`
	`716`	`+ await _flowControlAsyncWrapper.InvokeAsync(this, new FlowControlEventArgs(active, cancellationToken))`
`717`	`717`	`.ConfigureAwait(false);`
`718`	`718`	`}`
`719`	`719`
`@@ -723,7 +723,7 @@ await ModelSendAsync(in method, cancellationToken).`
`723`	`723`	`protected async Task<bool> HandleConnectionBlockedAsync(IncomingCommand cmd, CancellationToken cancellationToken)`
`724`	`724`	`{`
`725`	`725`	`string reason = new ConnectionBlocked(cmd.MethodSpan)._reason;`
`726`		`- await Session.Connection.HandleConnectionBlockedAsync(reason)`
	`726`	`+ await Session.Connection.HandleConnectionBlockedAsync(reason, cancellationToken)`
`727`	`727`	`.ConfigureAwait(false);`
`728`	`728`	`return true;`
`729`	`729`	`}`
`@@ -801,7 +801,7 @@ await k.HandleCommandAsync(cmd)`
`801`	`801`
`802`	`802`	`protected async Task<bool> HandleConnectionUnblockedAsync(CancellationToken cancellationToken)`
`803`	`803`	`{`
`804`		`- await Session.Connection.HandleConnectionUnblockedAsync()`
	`804`	`+ await Session.Connection.HandleConnectionUnblockedAsync(cancellationToken)`
`805`	`805`	`.ConfigureAwait(false);`
`806`	`806`	`return true;`
`807`	`807`	`}`
Original file line number	Diff line number	Diff line change
`@@ -54,20 +54,20 @@ internal void NotifyReceivedCloseOk()`
`54`	`54`	`_closed = true;`
`55`	`55`	`}`
`56`	`56`
`57`		`- internal Task HandleConnectionBlockedAsync(string reason)`
	`57`	`+ internal Task HandleConnectionBlockedAsync(string reason, CancellationToken cancellationToken)`
`58`	`58`	`{`
`59`	`59`	`if (!_connectionBlockedAsyncWrapper.IsEmpty)`
`60`	`60`	`{`
`61`		`- return _connectionBlockedAsyncWrapper.InvokeAsync(this, new ConnectionBlockedEventArgs(reason));`
	`61`	`+ return _connectionBlockedAsyncWrapper.InvokeAsync(this, new ConnectionBlockedEventArgs(reason, cancellationToken));`
`62`	`62`	`}`
`63`	`63`	`return Task.CompletedTask;`
`64`	`64`	`}`
`65`	`65`
`66`		`- internal Task HandleConnectionUnblockedAsync()`
	`66`	`+ internal Task HandleConnectionUnblockedAsync(CancellationToken cancellationToken)`
`67`	`67`	`{`
`68`	`68`	`if (!_connectionUnblockedAsyncWrapper.IsEmpty)`
`69`	`69`	`{`
`70`		`- return _connectionUnblockedAsyncWrapper.InvokeAsync(this, AsyncEventArgs.Empty);`
	`70`	`+ return _connectionUnblockedAsyncWrapper.InvokeAsync(this, AsyncEventArgs.CreateOrDefault(cancellationToken));`
`71`	`71`	`}`
`72`	`72`	`return Task.CompletedTask;`
`73`	`73`	`}`