Use alpha version of RabbitMQ transport

Travis Nickels · bording · commit c1ae8e2448ea · 2025-02-26T11:26:45.000-05:00
diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props
@@ -38,7 +38,8 @@
     <PackageVersion Include="NServiceBus.Metrics" Version="5.0.1" />
     <PackageVersion Include="NServiceBus.Metrics.ServiceControl" Version="5.0.0" />
     <PackageVersion Include="NServiceBus.Persistence.NonDurable" Version="2.0.1" />
-    <PackageVersion Include="NServiceBus.RabbitMQ" Version="9.2.0" />
+    <PackageVersion Include="NServiceBus.RabbitMQ" Version="10.0.0-alpha.0.92" />
+    <!--<PackageVersion Include="NServiceBus.RabbitMQ" Version="9.2.0" />-->
     <PackageVersion Include="NServiceBus.SagaAudit" Version="5.0.2" />
     <PackageVersion Include="NServiceBus.Testing" Version="9.0.1" />
     <PackageVersion Include="NServiceBus.Transport.AzureServiceBus" Version="5.0.0" />
diff --git a/src/ServiceControl.Transports.RabbitMQ/QueueLengthProvider.cs b/src/ServiceControl.Transports.RabbitMQ/QueueLengthProvider.cs
@@ -75,13 +75,13 @@ async Task FetchQueueLengths(CancellationToken cancellationToken)
         {
             foreach (var endpointQueuePair in endpointQueues)
             {
-                await queryExecutor.Execute(m =>
+                await queryExecutor.Execute(async m =>
                 {
                     var queueName = endpointQueuePair.Value;
 
                     try
                     {
-                        var size = (int)m.MessageCount(queueName);
+                        var size = (int)await m.MessageCountAsync(queueName, cancellationToken).ConfigureAwait(false);
 
                         sizes.AddOrUpdate(queueName, _ => size, (_, __) => size);
                     }
@@ -120,7 +120,7 @@ public void Initialize()
                     null); // value would come from config API in actual transport
             }
 
-            public async Task Execute(Action<IModel> action, CancellationToken cancellationToken = default)
+            public async Task Execute(Action<IChannel> action, CancellationToken cancellationToken = default)
             {
                 try
                 {
@@ -132,14 +132,14 @@ public async Task Execute(Action<IModel> action, CancellationToken cancellationT
                         await Task.Delay(ReconnectionDelay, cancellationToken);
                     }
 
-                    if (model == null || model.IsClosed)
+                    if (channel == null || channel.IsClosed)
                     {
-                        model?.Dispose();
+                        channel?.Dispose();
 
-                        model = connection.CreateModel();
+                        channel = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
                     }
 
-                    action(model);
+                    action(channel);
                 }
                 catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
                 {
@@ -154,7 +154,7 @@ public async Task Execute(Action<IModel> action, CancellationToken cancellationT
             public void Dispose() => connection?.Dispose();
 
             IConnection connection;
-            IModel model;
+            IChannel channel;
             ConnectionFactory connectionFactory;
 
             static readonly TimeSpan ReconnectionDelay = TimeSpan.FromSeconds(5);

Original file line number	Diff line number	Diff line change
`@@ -75,13 +75,13 @@ async Task FetchQueueLengths(CancellationToken cancellationToken)`
`75`	`75`	`{`
`76`	`76`	`foreach (var endpointQueuePair in endpointQueues)`
`77`	`77`	`{`
`78`		`- await queryExecutor.Execute(m =>`
	`78`	`+ await queryExecutor.Execute(async m =>`
`79`	`79`	`{`
`80`	`80`	`var queueName = endpointQueuePair.Value;`
`81`	`81`
`82`	`82`	`try`
`83`	`83`	`{`
`84`		`- var size = (int)m.MessageCount(queueName);`
	`84`	`+ var size = (int)await m.MessageCountAsync(queueName, cancellationToken).ConfigureAwait(false);`
`85`	`85`
`86`	`86`	`sizes.AddOrUpdate(queueName, _ => size, (_, __) => size);`
`87`	`87`	`}`
`@@ -120,7 +120,7 @@ public void Initialize()`
`120`	`120`	`null); // value would come from config API in actual transport`
`121`	`121`	`}`
`122`	`122`
`123`		`- public async Task Execute(Action<IModel> action, CancellationToken cancellationToken = default)`
	`123`	`+ public async Task Execute(Action<IChannel> action, CancellationToken cancellationToken = default)`
`124`	`124`	`{`
`125`	`125`	`try`
`126`	`126`	`{`
`@@ -132,14 +132,14 @@ public async Task Execute(Action<IModel> action, CancellationToken cancellationT`
`132`	`132`	`await Task.Delay(ReconnectionDelay, cancellationToken);`
`133`	`133`	`}`
`134`	`134`
`135`		`- if (model == null \|\| model.IsClosed)`
	`135`	`+ if (channel == null \|\| channel.IsClosed)`
`136`	`136`	`{`
`137`		`- model?.Dispose();`
	`137`	`+ channel?.Dispose();`
`138`	`138`
`139`		`- model = connection.CreateModel();`
	`139`	`+ channel = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false);`
`140`	`140`	`}`
`141`	`141`
`142`		`- action(model);`
	`142`	`+ action(channel);`
`143`	`143`	`}`
`144`	`144`	`catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)`
`145`	`145`	`{`
`@@ -154,7 +154,7 @@ public async Task Execute(Action<IModel> action, CancellationToken cancellationT`
`154`	`154`	`public void Dispose() => connection?.Dispose();`
`155`	`155`
`156`	`156`	`IConnection connection;`
`157`		`- IModel model;`
	`157`	`+ IChannel channel;`
`158`	`158`	`ConnectionFactory connectionFactory;`
`159`	`159`
`160`	`160`	`static readonly TimeSpan ReconnectionDelay = TimeSpan.FromSeconds(5);`