* Add a default consumer to see if perhaps that's why these tests sometimes fail.

Author: lukebakken

projects/RabbitMQ.Client/client/impl/ConsumerDispatching/ConsumerDispatcherBase.cs

@@ -7,7 +7,7 @@ namespace RabbitMQ.Client.ConsumerDispatching
 #nullable enable
     internal abstract class ConsumerDispatcherBase
     {
-        private static readonly FallbackConsumer fallbackConsumer = new FallbackConsumer();
+        private static readonly FallbackConsumer s_fallbackConsumer = new FallbackConsumer();
         private readonly Dictionary<string, IBasicConsumer> _consumers = new Dictionary<string, IBasicConsumer>();
 
         public IBasicConsumer? DefaultConsumer { get; set; }
@@ -74,7 +74,7 @@ private void DoShutdownConsumers(ShutdownEventArgs reason)
         [MethodImpl(MethodImplOptions.NoInlining)]
         private IBasicConsumer GetDefaultOrFallbackConsumer()
         {
-            return DefaultConsumer ?? fallbackConsumer;
+            return DefaultConsumer ?? s_fallbackConsumer;
         }
     }
 }

projects/Test/Integration/TestAsyncConsumer.cs

@@ -52,6 +52,8 @@ public TestAsyncConsumer(ITestOutputHelper output)
         [Fact]
         public async Task TestBasicRoundtripConcurrent()
         {
+            _channel.DefaultConsumer = new DefaultAsyncConsumer("_channel,", _output);
+
             QueueDeclareOk q = await _channel.QueueDeclareAsync();
 
             const int length = 4096;
@@ -129,6 +131,8 @@ public async Task TestBasicRoundtripConcurrent()
         [Fact]
         public async Task TestBasicRoundtripConcurrentManyMessages()
         {
+            _channel.DefaultConsumer = new DefaultAsyncConsumer("_channel,", _output);
+
             const int publish_total = 4096;
             const int length = 512;
             string queueName = GenerateQueueName();
@@ -185,6 +189,7 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
                                 };
                                 using (IChannel publishChannel = await publishConn.CreateChannelAsync())
                                 {
+                                    publishChannel.DefaultConsumer = new DefaultAsyncConsumer("publishChannel,", _output);
                                     publishChannel.ChannelShutdown += (o, ea) =>
                                     {
                                         HandleChannelShutdown(publishChannel, ea, (args) =>
@@ -221,6 +226,7 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
                                 };
                                 using (IChannel consumeChannel = await consumeConn.CreateChannelAsync())
                                 {
+                                    consumeChannel.DefaultConsumer = new DefaultAsyncConsumer("consumeChannel,", _output);
                                     consumeChannel.ChannelShutdown += (o, ea) =>
                                     {
                                         HandleChannelShutdown(consumeChannel, ea, (args) =>
@@ -592,5 +598,54 @@ private static bool ByteArraysEqual(ReadOnlySpan<byte> a1, ReadOnlySpan<byte> a2
         {
             return a1.SequenceEqual(a2);
         }
+
+        private class DefaultAsyncConsumer : AsyncDefaultBasicConsumer
+        {
+            private readonly string _logPrefix;
+            private readonly ITestOutputHelper _output;
+
+            public DefaultAsyncConsumer(string logPrefix, ITestOutputHelper output)
+            {
+                _logPrefix = logPrefix;
+                _output = output;
+            }
+
+            public override Task HandleBasicCancel(string consumerTag)
+            {
+                _output.WriteLine("[ERROR] {0} HandleBasicCancel {1}", _logPrefix, consumerTag);
+                return base.HandleBasicCancel(consumerTag);
+            }
+
+            public override Task HandleBasicCancelOk(string consumerTag)
+            {
+                _output.WriteLine("[ERROR] {0} HandleBasicCancelOk {1}", _logPrefix, consumerTag);
+                return base.HandleBasicCancelOk(consumerTag);
+            }
+
+            public override Task HandleBasicConsumeOk(string consumerTag)
+            {
+                _output.WriteLine("[ERROR] {0} HandleBasicConsumeOk {1}", _logPrefix, consumerTag);
+                return base.HandleBasicConsumeOk(consumerTag);
+            }
+
+            public override Task HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered,
+                string exchange, string routingKey, in ReadOnlyBasicProperties properties, ReadOnlyMemory<byte> body)
+            {
+                _output.WriteLine("[ERROR] {0} HandleBasicDeliver {1}", _logPrefix, consumerTag);
+                return base.HandleBasicDeliver(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body);
+            }
+
+            public override Task HandleChannelShutdown(object channel, ShutdownEventArgs reason)
+            {
+                _output.WriteLine("[ERROR] {0} HandleChannelShutdown", _logPrefix);
+                return base.HandleChannelShutdown(channel, reason);
+            }
+
+            public override Task OnCancel(params string[] consumerTags)
+            {
+                _output.WriteLine("[ERROR] {0} OnCancel {1}", _logPrefix, consumerTags[0]);
+                return base.OnCancel(consumerTags);
+            }
+        }
     }
 }