Drain and log pending work when AsyncConsumerDispatcher loop ends.

rabbitmq · Feb 5, 2025 · 3fff3aa · 3fff3aa
1 parent 6a627dd
commit 3fff3aa
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 9 deletions.
diff --git a/projects/RabbitMQ.Client/ConsumerDispatching/AsyncConsumerDispatcher.cs b/projects/RabbitMQ.Client/ConsumerDispatching/AsyncConsumerDispatcher.cs
@@ -3,6 +3,7 @@
 using System.Threading.Tasks;
 using RabbitMQ.Client.Events;
 using RabbitMQ.Client.Impl;
+using RabbitMQ.Client.Logging;
 
 namespace RabbitMQ.Client.ConsumerDispatching
 {
@@ -71,6 +72,16 @@ await _channel.OnCallbackExceptionAsync(CallbackExceptionEventArgs.Build(e, work
                     throw;
                 }
             }
+            finally
+            {
+                while (_reader.TryRead(out WorkStruct work))
+                {
+                    using (work)
+                    {
+                        ESLog.Warn($"discarding consumer work: {work.WorkType}");
+                    }
+                }
+            }
         }
     }
 }
diff --git a/projects/RabbitMQ.Client/ConsumerDispatching/ConsumerDispatcherChannelBase.cs b/projects/RabbitMQ.Client/ConsumerDispatching/ConsumerDispatcherChannelBase.cs
@@ -161,17 +161,10 @@ public async Task WaitForShutdownAsync()
             {
                 try
                 {
-                    await _worker
-                        .ConfigureAwait(false);
-
-                    /*
-                     * rabbitmq/rabbitmq-dotnet-client#1751
-                     *
-                     * Wait for the worker first to ensure all items have been read out of the channel,
-                     * otherwise the following will never return (https://stackoverflow.com/a/66521303)
-                     */
                     await _reader.Completion
                         .ConfigureAwait(false);
+                    await _worker
+                        .ConfigureAwait(false);
                 }
                 catch (AggregateException aex)
                 {