Prepare for RabbitMQ.Client v7.0.0

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.1" />
     <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;
         }
     }
 }

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

@@ -1,7 +1,6 @@
 namespace NServiceBus.Transport.RabbitMQ.AcceptanceTests
 {
     using System;
-    using System.Collections.Generic;
     using System.Threading;
     using System.Threading.Tasks;
     using AcceptanceTesting;
@@ -19,26 +18,24 @@ public class When_the_broker_connection_is_lost : NServiceBusAcceptanceTest
         [Test]
         public async Task Should_reconnect()
         {
-            var context = await Scenario.Define<MyContext>(myContext =>
+            var context = await Scenario.Define<MyContext>(ctx =>
             {
-                myContext.MessageId = Guid.NewGuid().ToString();
+                ctx.MessageId = Guid.NewGuid().ToString();
             })
-                .WithEndpoint<Receiver>()
-                .Done(c => c.GotTheMessage)
-                .Run();
+            .WithEndpoint<Receiver>()
+            .Done(c => c.GotTheMessage)
+            .Run();
 
             Assert.That(context.GotTheMessage, Is.True, "Should receive the message");
         }
 
         public class Receiver : EndpointConfigurationBuilder
         {
-            public Receiver()
-            {
+            public Receiver() =>
                 EndpointSetup<DefaultServer>(e =>
                 {
                     e.EnableFeature<ConnectionKillerFeature>();
                 });
-            }
 
             class ConnectionKillerFeature : Feature
             {
@@ -48,15 +45,9 @@ protected override void Setup(FeatureConfigurationContext context)
                     context.RegisterStartupTask(b => b.GetRequiredService<ConnectionKiller>());
                 }
 
-                class ConnectionKiller : FeatureStartupTask
+                class ConnectionKiller(IMessageDispatcher sender, IReadOnlySettings settings, MyContext context)
+                    : FeatureStartupTask
                 {
-                    public ConnectionKiller(IMessageDispatcher sender, IReadOnlySettings settings, MyContext context)
-                    {
-                        this.context = context;
-                        this.sender = sender;
-                        this.settings = settings;
-                    }
-
                     protected override async Task OnStart(IMessageSession session, CancellationToken cancellationToken = default)
                     {
                         await BreakConnectionBySendingInvalidMessage(cancellationToken);
@@ -70,7 +61,7 @@ async Task BreakConnectionBySendingInvalidMessage(CancellationToken cancellation
                     {
                         try
                         {
-                            var outgoingMessage = new OutgoingMessage("Foo", [], new byte[0]);
+                            var outgoingMessage = new OutgoingMessage("Foo", [], Array.Empty<byte>());
                             var props = new DispatchProperties
                             {
                                 DiscardIfNotReceivedBefore =
@@ -84,10 +75,6 @@ async Task BreakConnectionBySendingInvalidMessage(CancellationToken cancellation
                             // Don't care
                         }
                     }
-
-                    readonly MyContext context;
-                    readonly IMessageDispatcher sender;
-                    readonly IReadOnlySettings settings;
                 }
             }