Prepare for RabbitMQ.Client v7.0.0 (#1446)

* Wip upgrade

* More changes

* More changes

* Fix warnings

* Remove our now unneeded publisher confirms handling

* Use CatchAsync

* Add awaits to tests

* Add missing await

* ConfirmsAwareChannel init for now

* Wire up unregister event for consumer cancellation

* Headers can be null when nothing was sent

* Add a done condition (still not an entirely perfect test but we'll go from there)

* Cleanup test a bit

* Update approval file

* Don't cancel the connection close

* ToHashSet

Co-authored-by: Paulo Morgado <470455+paulomorgado@users.noreply.github.com>

* Cleanup test

* Switch regex approach

* Async continuation

* Forward cancellation token with exception handling as a spike

* Update to 7.0.0-rc.9

This version restores the parameter order from 6.x for `BasicPublishAsync`

* Cleanup

* More cleanup

* Add more info to test exception message

* Add more logging

* Simplify

* Return channel after all tasks are completed now that send is async

* Put return in finally block

* Try adding our own tracking back

* Revert "Try adding our own tracking back"

This reverts commit 99bc699.

* WaitForConfirmsOrDieAsync for confirms

* Upgrade to RC11

* Try adding our own tracking back

* Comment

* Synchronize for now the sequence number access around publish (deliberate wider scope for safety as a trial)

* Move NextPublishSeqNo acquisition inside semaphore

* Remove TODOs that are not necessary
For example channel operations should not be parallelized.

* Non-generic task completion source

* readonly

* It seems we have to pass the non-readonly version

* UnsafeRegister

* Bump version to 10.0

---------

Co-authored-by: Daniel Marbach <danielmarbach@users.noreply.github.com>
Co-authored-by: Brandon Ording <bording@gmail.com>
Co-authored-by: Paulo Morgado <470455+paulomorgado@users.noreply.github.com>
Co-authored-by: Luke Bakken <luke@bakken.io>

Author: 4 people

src/Custom.Build.props

@@ -1,7 +1,7 @@
 <Project>
 
   <PropertyGroup>
-    <MinVerMinimumMajorMinor>9.0</MinVerMinimumMajorMinor>
+    <MinVerMinimumMajorMinor>10.0</MinVerMinimumMajorMinor>
     <MinVerAutoIncrement>minor</MinVerAutoIncrement>
   </PropertyGroup>
 

src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/ChannelExtensions.cs

@@ -1,21 +1,22 @@
 namespace NServiceBus.Transport.RabbitMQ.AcceptanceTests
 {
     using System.Collections.Generic;
+    using System.Threading.Tasks;
     using global::RabbitMQ.Client;
 
     static class ChannelExtensions
     {
-        public static void DeclareQuorumQueue(this IModel channel, string queueName)
+        public static Task<QueueDeclareOk> DeclareQuorumQueue(this IChannel channel, string queueName)
         {
-            channel.QueueDeclare(queueName, true, false, false, new Dictionary<string, object>
+            return channel.QueueDeclareAsync(queueName, true, false, false, new Dictionary<string, object>
             {
                 { "x-queue-type", "quorum" }
             });
         }
 
-        public static void DeclareClassicQueue(this IModel channel, string queueName)
+        public static Task<QueueDeclareOk> DeclareClassicQueue(this IChannel channel, string queueName)
         {
-            channel.QueueDeclare(queueName, true, false, false);
+            return channel.QueueDeclareAsync(queueName, true, false, false);
         }
     }
 }

src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/ConfigureEndpointRabbitMQTransport.cs

@@ -27,30 +27,28 @@ public Task Configure(string endpointName, EndpointConfiguration configuration,
         return Task.CompletedTask;
     }
 
-    public Task Cleanup()
+    public async Task Cleanup()
     {
-        PurgeQueues();
-
-        return Task.CompletedTask;
+        await PurgeQueues();
     }
 
-    void PurgeQueues()
+    async Task PurgeQueues()
     {
         if (transport == null)
         {
             return;
         }
 
-        var queues = transport.QueuesToCleanup.Distinct().ToArray();
+        var queues = transport.QueuesToCleanup.ToHashSet();
 
-        using (var connection = ConnectionHelper.ConnectionFactory.CreateConnection("Test Queue Purger"))
-        using (var channel = connection.CreateModel())
+        using (var connection = await ConnectionHelper.ConnectionFactory.CreateConnectionAsync("Test Queue Purger"))
+        using (var channel = await connection.CreateChannelAsync())
         {
             foreach (var queue in queues)
             {
                 try
                 {
-                    channel.QueuePurge(queue);
+                    await channel.QueuePurgeAsync(queue);
                 }
                 catch (Exception ex)
                 {

src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/NServiceBus.Transport.RabbitMQ.AcceptanceTests.csproj

@@ -21,7 +21,7 @@
   <ItemGroup>
     <PackageReference Include="BitFaster.Caching" Version="2.5.2" />
     <PackageReference Include="NServiceBus.AcceptanceTests.Sources" Version="9.2.2" GeneratePathProperty="true" />
-    <PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
+    <PackageReference Include="RabbitMQ.Client" Version="7.0.0-rc.11" />
   </ItemGroup>
 
   <ItemGroup>

src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/QuorumQueues/When_classic_endpoint_uses_quorum_error_queue.cs

@@ -1,6 +1,7 @@
 namespace NServiceBus.Transport.RabbitMQ.AcceptanceTests
 {
     using System;
+    using System.Threading.Tasks;
     using AcceptanceTesting;
     using NServiceBus.AcceptanceTests;
     using NServiceBus.AcceptanceTests.EndpointTemplates;
@@ -9,12 +10,12 @@
     public class When_classic_endpoint_uses_quorum_error_queue : NServiceBusAcceptanceTest
     {
         [Test]
-        public void Should_fail_to_start()
+        public async Task Should_fail_to_start()
         {
-            using (var connection = ConnectionHelper.ConnectionFactory.CreateConnection())
-            using (var channel = connection.CreateModel())
+            using (var connection = await ConnectionHelper.ConnectionFactory.CreateConnectionAsync())
+            using (var channel = await connection.CreateChannelAsync())
             {
-                channel.DeclareQuorumQueue("rabbitmq.transport.tests.quorum-error");
+                await channel.DeclareQuorumQueue("rabbitmq.transport.tests.quorum-error");
             }
 
             var exception = Assert.CatchAsync<Exception>(async () => await Scenario.Define<ScenarioContext>()

src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/QuorumQueues/When_classic_endpoint_uses_quorum_queue.cs

@@ -1,6 +1,7 @@
 namespace NServiceBus.Transport.RabbitMQ.AcceptanceTests
 {
     using System;
+    using System.Threading.Tasks;
     using AcceptanceTesting;
     using AcceptanceTesting.Customization;
     using NServiceBus.AcceptanceTests;
@@ -10,12 +11,12 @@
     public class When_classic_endpoint_uses_quorum_queue : NServiceBusAcceptanceTest
     {
         [Test]
-        public void Should_fail_to_start()
+        public async Task Should_fail_to_start()
         {
-            using (var connection = ConnectionHelper.ConnectionFactory.CreateConnection())
-            using (var channel = connection.CreateModel())
+            using (var connection = await ConnectionHelper.ConnectionFactory.CreateConnectionAsync())
+            using (var channel = await connection.CreateChannelAsync())
             {
-                channel.DeclareQuorumQueue(Conventions.EndpointNamingConvention(typeof(ClassicQueueEndpoint)));
+                await channel.DeclareQuorumQueue(Conventions.EndpointNamingConvention(typeof(ClassicQueueEndpoint)));
             }
 
             var exception = Assert.CatchAsync<Exception>(async () => await Scenario.Define<ScenarioContext>()

src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/QuorumQueues/When_quorum_endpoint_uses_classic_error_queue.cs

@@ -1,19 +1,20 @@
 namespace NServiceBus.Transport.RabbitMQ.AcceptanceTests
 {
     using System;
+    using System.Threading.Tasks;
     using AcceptanceTesting;
     using NServiceBus.AcceptanceTests;
     using NUnit.Framework;
 
     public class When_quorum_endpoint_uses_classic_error_queue : NServiceBusAcceptanceTest
     {
         [Test]
-        public void Should_fail_to_start()
+        public async Task Should_fail_to_start()
         {
-            using (var connection = ConnectionHelper.ConnectionFactory.CreateConnection())
-            using (var channel = connection.CreateModel())
+            using (var connection = await ConnectionHelper.ConnectionFactory.CreateConnectionAsync())
+            using (var channel = await connection.CreateChannelAsync())
             {
-                channel.DeclareClassicQueue("rabbitmq.transport.tests.classic-error");
+                await channel.DeclareClassicQueue("rabbitmq.transport.tests.classic-error");
             }
 
 

src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/QuorumQueues/When_quorum_endpoint_uses_classic_queue.cs

@@ -1,6 +1,7 @@
 namespace NServiceBus.Transport.RabbitMQ.AcceptanceTests
 {
     using System;
+    using System.Threading.Tasks;
     using AcceptanceTesting;
     using AcceptanceTesting.Customization;
     using NServiceBus.AcceptanceTests;
@@ -9,12 +10,12 @@
     public class When_quorum_endpoint_uses_classic_queue : NServiceBusAcceptanceTest
     {
         [Test]
-        public void Should_fail_to_start()
+        public async Task Should_fail_to_start()
         {
-            using (var connection = ConnectionHelper.ConnectionFactory.CreateConnection())
-            using (var channel = connection.CreateModel())
+            using (var connection = await ConnectionHelper.ConnectionFactory.CreateConnectionAsync())
+            using (var channel = await connection.CreateChannelAsync())
             {
-                channel.DeclareClassicQueue(Conventions.EndpointNamingConvention(typeof(QuorumQueueEndpoint)));
+                await channel.DeclareClassicQueue(Conventions.EndpointNamingConvention(typeof(QuorumQueueEndpoint)));
             }
 
             var exception = Assert.CatchAsync<Exception>(async () => await Scenario.Define<ScenarioContext>()

src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/QuorumQueues/When_starting_endpoint_using_quorum_queues.cs

@@ -15,12 +15,12 @@ public async Task Should_create_receiving_queues_as_quorum_queues()
         {
             var endpointInputQueue = Conventions.EndpointNamingConvention(typeof(QuorumQueueEndpoint));
 
-            using (var connection = ConnectionHelper.ConnectionFactory.CreateConnection())
-            using (var channel = connection.CreateModel())
+            using (var connection = await ConnectionHelper.ConnectionFactory.CreateConnectionAsync())
+            using (var channel = await connection.CreateChannelAsync())
             {
-                channel.QueueDelete(endpointInputQueue, false, false);
-                channel.QueueDelete(endpointInputQueue + "-disc", false, false);
-                channel.QueueDelete("QuorumQueueSatelliteReceiver", false, false);
+                await channel.QueueDeleteAsync(endpointInputQueue, false, false);
+                await channel.QueueDeleteAsync(endpointInputQueue + "-disc", false, false);
+                await channel.QueueDeleteAsync("QuorumQueueSatelliteReceiver", false, false);
             }
 
             await Scenario.Define<ScenarioContext>()
@@ -29,23 +29,23 @@ await Scenario.Define<ScenarioContext>()
                 .Run();
 
             // try to declare the same queue as a non-quorum queue, which should fail:
-            using (var connection = ConnectionHelper.ConnectionFactory.CreateConnection())
+            using (var connection = await ConnectionHelper.ConnectionFactory.CreateConnectionAsync())
             {
-                using (var channel = connection.CreateModel())
+                using (var channel = await connection.CreateChannelAsync())
                 {
-                    var mainQueueException = Assert.Catch<RabbitMQClientException>(() => channel.DeclareClassicQueue(endpointInputQueue));
+                    var mainQueueException = Assert.CatchAsync<RabbitMQClientException>(async () => await channel.DeclareClassicQueue(endpointInputQueue));
                     Assert.That(mainQueueException.Message, Does.Contain("PRECONDITION_FAILED - inequivalent arg 'x-queue-type'"));
                 }
 
-                using (var channel = connection.CreateModel())
+                using (var channel = await connection.CreateChannelAsync())
                 {
-                    var instanceSpecificQueueException = Assert.Catch<RabbitMQClientException>(() => channel.DeclareClassicQueue(endpointInputQueue + "-disc"));
+                    var instanceSpecificQueueException = Assert.CatchAsync<RabbitMQClientException>(async () => await channel.DeclareClassicQueue(endpointInputQueue + "-disc"));
                     Assert.That(instanceSpecificQueueException.Message, Does.Contain("PRECONDITION_FAILED - inequivalent arg 'x-queue-type'"));
                 }
 
-                using (var channel = connection.CreateModel())
+                using (var channel = await connection.CreateChannelAsync())
                 {
-                    var satelliteReceiver = Assert.Catch<RabbitMQClientException>(() => channel.DeclareClassicQueue("QuorumQueueSatelliteReceiver"));
+                    var satelliteReceiver = Assert.CatchAsync<RabbitMQClientException>(async () => await channel.DeclareClassicQueue("QuorumQueueSatelliteReceiver"));
                     Assert.That(satelliteReceiver.Message, Does.Contain("PRECONDITION_FAILED - inequivalent arg 'x-queue-type'"));
                 }
             }

src/NServiceBus.Transport.RabbitMQ.AcceptanceTests/When_scaling_out_subscribers.cs

@@ -1,6 +1,7 @@
 namespace NServiceBus.Transport.RabbitMQ.AcceptanceTests
 {
     using System;
+    using System.Threading;
     using System.Threading.Tasks;
     using AcceptanceTesting;
     using NServiceBus.AcceptanceTests;
@@ -32,59 +33,42 @@ public async Task Should_only_deliver_event_to_one_of_the_instances()
                            c.ServerBSubscribed = true;
                        });
                    })
+                   .Done(ctx => ctx.Counter > 0)
                    .Run(TimeSpan.FromSeconds(10));
 
             Assert.That(context.Counter, Is.EqualTo(1), "One of the scaled out instances should get the event");
         }
 
         public class ScaledOutSubscriber : EndpointConfigurationBuilder
         {
-            public ScaledOutSubscriber()
-            {
-                EndpointSetup<DefaultPublisher>();
-            }
+            public ScaledOutSubscriber() => EndpointSetup<DefaultPublisher>();
 
-            class MyEventHandler : IHandleMessages<MyEvent>
+            class MyEventHandler(MyContext myContext) : IHandleMessages<MyEvent>
             {
-                readonly MyContext myContext;
-
-                public MyEventHandler(MyContext context)
-                {
-                    myContext = context;
-                }
-
                 public Task Handle(MyEvent message, IMessageHandlerContext context)
                 {
-                    lock (objLock)
-                    {
-                        myContext.Counter++;
-                    }
-
+                    myContext.IncrementCounter();
                     return Task.CompletedTask;
                 }
-
-                static object objLock = new object();
             }
         }
 
         public class Publisher : EndpointConfigurationBuilder
         {
-            public Publisher()
-            {
-                EndpointSetup<DefaultPublisher>();
-            }
+            public Publisher() => EndpointSetup<DefaultPublisher>();
         }
 
-        public class MyEvent : IEvent
-        {
-
-        }
+        public class MyEvent : IEvent;
 
         class MyContext : ScenarioContext
         {
             public bool ServerASubscribed { get; set; }
             public bool ServerBSubscribed { get; set; }
-            public int Counter { get; set; }
+            public int Counter => counter;
+
+            public void IncrementCounter() => Interlocked.Increment(ref counter);
+
+            int counter;
         }
     }
 }