Making sure AsyncEventHandlers are all run. Fixes issue rabbitmq#838.

Author: stebet

projects/RabbitMQ.Client/client/api/AsyncDefaultBasicConsumer.cs

@@ -140,11 +140,7 @@ public virtual Task HandleModelShutdown(object model, ShutdownEventArgs reason)
         public virtual async Task OnCancel(params string[] consumerTags)
         {
             IsRunning = false;
-            foreach (AsyncEventHandler<ConsumerEventArgs> h in ConsumerCancelled?.GetInvocationList() ?? Array.Empty<Delegate>())
-            {
-                await h(this, new ConsumerEventArgs(consumerTags)).ConfigureAwait(false);
-            }
-
+            await ConsumerCancelled.InvokeAsync(this, new ConsumerEventArgs(consumerTags)).ConfigureAwait(false);
             foreach (string consumerTag in consumerTags)
             {
                 _consumerTags.Remove(consumerTag);

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

@@ -4,4 +4,18 @@
 namespace RabbitMQ.Client.Events
 {
     public delegate Task AsyncEventHandler<in TEvent>(object sender, TEvent @event) where TEvent : EventArgs;
-}
+
+    internal static class AsyncEventHandlerExtensions
+    {
+        public static async Task InvokeAsync<TEvent>(this AsyncEventHandler<TEvent> eventHandler, object sender, TEvent @event) where TEvent : EventArgs
+        {
+            if(eventHandler != null)
+            {
+                foreach(AsyncEventHandler<TEvent> handlerInstance in eventHandler.GetInvocationList())
+                {
+                    await handlerInstance(sender, @event).ConfigureAwait(false);
+                }
+            }
+        }
+    }
+}

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

@@ -34,30 +34,28 @@ public AsyncEventingBasicConsumer(IModel model) : base(model)
         public override async Task HandleBasicCancelOk(string consumerTag)
         {
             await base.HandleBasicCancelOk(consumerTag).ConfigureAwait(false);
-            await (Unregistered?.Invoke(this, new ConsumerEventArgs(new[] { consumerTag })) ?? Task.CompletedTask).ConfigureAwait(false);
+            await Unregistered.InvokeAsync(this, new ConsumerEventArgs(new[] { consumerTag })).ConfigureAwait(false);
         }
 
         ///<summary>Fires when the server confirms successful consumer registration.</summary>
         public override async Task HandleBasicConsumeOk(string consumerTag)
         {
             await base.HandleBasicConsumeOk(consumerTag).ConfigureAwait(false);
-            await (Registered?.Invoke(this, new ConsumerEventArgs(new[] { consumerTag })) ?? Task.CompletedTask).ConfigureAwait(false);
+            await Registered.InvokeAsync(this, new ConsumerEventArgs(new[] { consumerTag })).ConfigureAwait(false);
         }
 
         ///<summary>Fires the Received event.</summary>
         public override async Task HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, ReadOnlyMemory<byte> body)
         {
             await base.HandleBasicDeliver(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body).ConfigureAwait(false);
-            await (Received?.Invoke(
-                this,
-                new BasicDeliverEventArgs(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body)) ?? Task.CompletedTask).ConfigureAwait(false);
+            await Received.InvokeAsync(this, new BasicDeliverEventArgs(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body)).ConfigureAwait(false);
         }
 
         ///<summary>Fires the Shutdown event.</summary>
         public override async Task HandleModelShutdown(object model, ShutdownEventArgs reason)
         {
             await base.HandleModelShutdown(model, reason).ConfigureAwait(false);
-            await (Shutdown?.Invoke(this, reason) ?? Task.CompletedTask).ConfigureAwait(false);
+            await Shutdown.InvokeAsync(this, reason).ConfigureAwait(false);
         }
     }
 }