Propagate token into wrapper

Author: danielmarbach

projects/RabbitMQ.Client/client/events/CallbackExceptionEventArgs.cs

@@ -78,28 +78,29 @@ public class CallbackExceptionEventArgs : BaseExceptionEventArgs
         private const string ContextString = "context";
         private const string ConsumerString = "consumer";
 
+        // TODO Why is this public when there is a build method?
         public CallbackExceptionEventArgs(IDictionary<string, object> detail, Exception exception, CancellationToken cancellationToken = default)
             : base(detail, exception, cancellationToken)
         {
         }
 
-        public static CallbackExceptionEventArgs Build(Exception e, string context)
+        public static CallbackExceptionEventArgs Build(Exception e, string context, CancellationToken cancellationToken = default)
         {
             var details = new Dictionary<string, object>(1)
             {
                 {ContextString, context}
             };
-            return new CallbackExceptionEventArgs(details, e);
+            return new CallbackExceptionEventArgs(details, e, cancellationToken);
         }
 
-        public static CallbackExceptionEventArgs Build(Exception e, string context, object consumer)
+        public static CallbackExceptionEventArgs Build(Exception e, string context, object consumer, CancellationToken cancellationToken = default)
         {
             var details = new Dictionary<string, object>(2)
             {
                 {ContextString, context},
                 {ConsumerString, consumer}
             };
-            return new CallbackExceptionEventArgs(details, e);
+            return new CallbackExceptionEventArgs(details, e, cancellationToken);
         }
     }
 }

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

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using RabbitMQ.Client.Events;
 
@@ -9,11 +10,11 @@ internal struct AsyncEventingWrapper<T> where T : AsyncEventArgs
         private event AsyncEventHandler<T>? _event;
         private Delegate[]? _handlers;
         private string? _context;
-        private Func<Exception, string, Task>? _onException;
+        private Func<Exception, string, CancellationToken, Task>? _onException;
 
         public readonly bool IsEmpty => _event is null;
 
-        public AsyncEventingWrapper(string context, Func<Exception, string, Task> onException)
+        public AsyncEventingWrapper(string context, Func<Exception, string, CancellationToken, Task> onException)
         {
             _event = null;
             _handlers = null;
@@ -34,7 +35,7 @@ public void RemoveHandler(AsyncEventHandler<T>? handler)
         }
 
         // Do not make this function async! (This type is a struct that gets copied at the start of an async method => empty _handlers is copied)
-        public Task InvokeAsync(object sender, T parameter)
+        public Task InvokeAsync(object sender, T parameter, CancellationToken cancellationToken = default)
         {
             Delegate[]? handlers = _handlers;
             if (handlers is null)
@@ -48,10 +49,10 @@ public Task InvokeAsync(object sender, T parameter)
                 _handlers = handlers;
             }
 
-            return InternalInvoke(handlers, sender, parameter);
+            return InternalInvoke(handlers, sender, parameter, cancellationToken);
         }
 
-        private readonly async Task InternalInvoke(Delegate[] handlers, object sender, T parameter)
+        private readonly async Task InternalInvoke(Delegate[] handlers, object sender, T parameter, CancellationToken cancellationToken)
         {
             foreach (AsyncEventHandler<T> action in handlers)
             {
@@ -64,7 +65,7 @@ await action(sender, parameter)
                 {
                     if (_onException != null)
                     {
-                        await _onException(exception, _context!)
+                        await _onException(exception, _context!, cancellationToken)
                             .ConfigureAwait(false);
                     }
                     else

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

@@ -83,8 +83,8 @@ internal AutorecoveringConnection(ConnectionConfig config, IEndpointResolver end
             _recoveringConsumerAsyncWrapper =
                 new AsyncEventingWrapper<RecoveringConsumerEventArgs>("OnRecoveringConsumer", onExceptionAsync);
 
-            Task onExceptionAsync(Exception exception, string context) =>
-                _innerConnection.OnCallbackExceptionAsync(CallbackExceptionEventArgs.Build(exception, context));
+            Task onExceptionAsync(Exception exception, string context, CancellationToken cancellationToken) =>
+                _innerConnection.OnCallbackExceptionAsync(CallbackExceptionEventArgs.Build(exception, context, cancellationToken));
         }
 
         internal static async ValueTask<AutorecoveringConnection> CreateAsync(ConnectionConfig config, IEndpointResolver endpoints,

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

@@ -79,13 +79,13 @@ protected ChannelBase(ConnectionConfig config, ISession session,
             ContinuationTimeout = config.ContinuationTimeout;
             ConsumerDispatcher = new AsyncConsumerDispatcher(this,
                 perChannelConsumerDispatchConcurrency.GetValueOrDefault(config.ConsumerDispatchConcurrency));
-            Func<Exception, string, Task> onExceptionAsync = (exception, context) =>
-                OnCallbackExceptionAsync(CallbackExceptionEventArgs.Build(exception, context));
+            Func<Exception, string, CancellationToken, Task> onExceptionAsync = (exception, context, cancellationToken) =>
+                OnCallbackExceptionAsync(CallbackExceptionEventArgs.Build(exception, context, cancellationToken));
             _basicAcksAsyncWrapper = new AsyncEventingWrapper<BasicAckEventArgs>("OnBasicAck", onExceptionAsync);
             _basicNacksAsyncWrapper = new AsyncEventingWrapper<BasicNackEventArgs>("OnBasicNack", onExceptionAsync);
             _basicReturnAsyncWrapper = new AsyncEventingWrapper<BasicReturnEventArgs>("OnBasicReturn", onExceptionAsync);
             _callbackExceptionAsyncWrapper =
-                new AsyncEventingWrapper<CallbackExceptionEventArgs>(string.Empty, (exception, context) => Task.CompletedTask);
+                new AsyncEventingWrapper<CallbackExceptionEventArgs>(string.Empty, (exception, context, cancellationToken) => Task.CompletedTask);
             _flowControlAsyncWrapper = new AsyncEventingWrapper<FlowControlEventArgs>("OnFlowControl", onExceptionAsync);
             _channelShutdownAsyncWrapper = new AsyncEventingWrapper<ShutdownEventArgs>("OnChannelShutdownAsync", onExceptionAsync);
             _recoveryAsyncWrapper = new AsyncEventingWrapper<AsyncEventArgs>("OnChannelRecovery", onExceptionAsync);

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

@@ -66,7 +66,7 @@ internal Connection(ConnectionConfig config, IFrameHandler frameHandler)
 
             _callbackExceptionAsyncWrapper =
                 new AsyncEventingWrapper<CallbackExceptionEventArgs>(string.Empty,
-                    (exception, context) => Task.CompletedTask);
+                    (exception, context, cancellationToken) => Task.CompletedTask);
 
             _connectionBlockedAsyncWrapper =
                 new AsyncEventingWrapper<ConnectionBlockedEventArgs>("OnConnectionBlocked", onExceptionAsync);
@@ -89,8 +89,8 @@ internal Connection(ConnectionConfig config, IFrameHandler frameHandler)
 
             _mainLoopTask = Task.CompletedTask;
 
-            Task onExceptionAsync(Exception exception, string context) =>
-                OnCallbackExceptionAsync(CallbackExceptionEventArgs.Build(exception, context));
+            Task onExceptionAsync(Exception exception, string context, CancellationToken cancellationToken) =>
+                OnCallbackExceptionAsync(CallbackExceptionEventArgs.Build(exception, context, cancellationToken));
         }
 
         public Guid Id => _id;