Work on removing polling from RabbitMQ transport

Rebus.RabbitMq.Tests/AssemblyInfo.cs

@@ -0,0 +1,3 @@
+using NUnit.Framework;
+
+[assembly: Timeout(100000)]

Rebus.RabbitMq/Internals/ConnectionManager.cs

@@ -63,7 +63,8 @@ public ConnectionManager(IList<ConnectionEndpoint> endpoints, string inputQueueA
                 Uri = endpoints.First().ConnectionUri, //Use the first URI in the list for ConnectionFactory to pick the AMQP credentials, VirtualHost (if any)
                 AutomaticRecoveryEnabled = true,
                 NetworkRecoveryInterval = TimeSpan.FromSeconds(30),
-                ClientProperties = CreateClientProperties(inputQueueAddress)
+                ClientProperties = CreateClientProperties(inputQueueAddress),
+                DispatchConsumersAsync = true,
             };
 
             if (customizer != null)
@@ -120,7 +121,8 @@ public ConnectionManager(string connectionString, string inputQueueAddress, IReb
                 Uri = new Uri(uriStrings.First()), //Use the first URI in the list for ConnectionFactory to pick the AMQP credentials (if any)
                 AutomaticRecoveryEnabled = true,
                 NetworkRecoveryInterval = TimeSpan.FromSeconds(30),
-                ClientProperties = CreateClientProperties(inputQueueAddress)
+                ClientProperties = CreateClientProperties(inputQueueAddress),
+                DispatchConsumersAsync = true,
             };
 
             if (customizer != null)

Rebus.RabbitMq/Internals/CustomQueueingConsumer.cs

@@ -3,19 +3,24 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
 
 namespace Rebus.Internals
 {
-    class CustomQueueingConsumer : DefaultBasicConsumer
+    class CustomQueueingConsumer : AsyncDefaultBasicConsumer
     {
-        public SharedQueue<BasicDeliverEventArgs> Queue { get; } = new SharedQueue<BasicDeliverEventArgs>();
-        public CustomQueueingConsumer(IModel model) : base(model)
+        private Channel<BasicDeliverEventArgs> Queue { get; }
+        public CustomQueueingConsumer(IModel model, int maxSize) : base(model)
         {
+            // Use a bounded queue to avoid a memory explosion somewhere in case something goes haywire.
+            Queue = Channel.CreateBounded<BasicDeliverEventArgs>(maxSize);
         }
 
-        public override void HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, ReadOnlyMemory<byte> body)
+        public override async Task HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, ReadOnlyMemory<byte> body)
         {
-            Queue.Enqueue(new BasicDeliverEventArgs
+            await Queue.Writer.WriteAsync(new BasicDeliverEventArgs
             {
                 ConsumerTag = consumerTag,
                 DeliveryTag = deliveryTag,
@@ -27,10 +32,16 @@ public override void HandleBasicDeliver(string consumerTag, ulong deliveryTag, b
             });
         }
 
-        public override void OnCancel(params string[] consumerTags)
+        public override Task OnCancel(params string[] consumerTags)
         {
             base.OnCancel(consumerTags);
-            Queue.Close();
+            Queue.Writer.Complete();
+            return Task.CompletedTask;
+        }
+
+        public ValueTask<BasicDeliverEventArgs> GetNext(CancellationToken cancellationToken)
+        {
+            return Queue.Reader.ReadAsync(cancellationToken);
         }
 
         public void Dispose()