Merge pull request #1384 from danielgerlag/copilot/fix-rabbitmq-client-support-issue

Add RabbitMQ.Client v7 support by migrating to async API

Author: danielgerlag

src/providers/WorkflowCore.QueueProviders.RabbitMQ/ServiceCollectionExtensions.cs

@@ -2,6 +2,8 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using WorkflowCore.Interface;
 using WorkflowCore.Models;
@@ -10,7 +12,7 @@
 
 namespace Microsoft.Extensions.DependencyInjection
 {
-    public delegate IConnection RabbitMqConnectionFactory(IServiceProvider sp, string clientProvidedName);
+    public delegate Task<IConnection> RabbitMqConnectionFactory(IServiceProvider sp, string clientProvidedName, CancellationToken cancellationToken = default);
 
     public static class ServiceCollectionExtensions
     {
@@ -20,7 +22,7 @@ public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options, IConnect
             if (connectionFactory == null) throw new ArgumentNullException(nameof(connectionFactory));
 
             return options
-                .UseRabbitMQ((sp, name) => connectionFactory.CreateConnection(name));
+                .UseRabbitMQ(async (sp, name, cancellationToken) => await connectionFactory.CreateConnectionAsync(name, cancellationToken));
         }
 
         public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options,
@@ -32,7 +34,7 @@ public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options,
             if (hostnames == null) throw new ArgumentNullException(nameof(hostnames));
 
             return options
-                .UseRabbitMQ((sp, name) => connectionFactory.CreateConnection(hostnames.ToList(), name));
+                .UseRabbitMQ(async (sp, name, cancellationToken) => await connectionFactory.CreateConnectionAsync(hostnames, name, cancellationToken));
         }
 
         public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options, RabbitMqConnectionFactory rabbitMqConnectionFactory)

src/providers/WorkflowCore.QueueProviders.RabbitMQ/Services/RabbitMQProvider.cs

@@ -37,11 +37,16 @@ public async Task QueueWork(string id, QueueType queue)
             if (_connection == null)
                 throw new InvalidOperationException("RabbitMQ provider not running");
 
-            using (var channel = _connection.CreateModel())
+            var channel = await _connection.CreateChannelAsync(new CreateChannelOptions(publisherConfirmationsEnabled: false, publisherConfirmationTrackingEnabled: false), CancellationToken.None);
+            try
             {
-                channel.QueueDeclare(queue: _queueNameProvider.GetQueueName(queue), durable: true, exclusive: false, autoDelete: false, arguments: null);
-                var body = Encoding.UTF8.GetBytes(id);
-                channel.BasicPublish(exchange: "", routingKey: _queueNameProvider.GetQueueName(queue), basicProperties: null, body: body);
+                await channel.QueueDeclareAsync(queue: _queueNameProvider.GetQueueName(queue), durable: true, exclusive: false, autoDelete: false, arguments: null, passive: false, noWait: false, CancellationToken.None);
+                var body = new ReadOnlyMemory<byte>(Encoding.UTF8.GetBytes(id));
+                await channel.BasicPublishAsync(exchange: "", routingKey: _queueNameProvider.GetQueueName(queue), mandatory: false, basicProperties: new BasicProperties(), body: body, CancellationToken.None);
+            }
+            finally
+            {
+                await channel.CloseAsync(200, "OK", abort: false, CancellationToken.None);
             }
         }
 
@@ -50,46 +55,54 @@ public async Task<string> DequeueWork(QueueType queue, CancellationToken cancell
             if (_connection == null)
                 throw new InvalidOperationException("RabbitMQ provider not running");
 
-            using (var channel = _connection.CreateModel())
+            var channel = await _connection.CreateChannelAsync(new CreateChannelOptions(publisherConfirmationsEnabled: false, publisherConfirmationTrackingEnabled: false), CancellationToken.None);
+            try
             {
-                channel.QueueDeclare(queue: _queueNameProvider.GetQueueName(queue),
-                                     durable: true,
-                                     exclusive: false,
-                                     autoDelete: false,
-                                     arguments: null);
+                await channel.QueueDeclareAsync(queue: _queueNameProvider.GetQueueName(queue),
+                                         durable: true,
+                                         exclusive: false,
+                                         autoDelete: false,
+                                         arguments: null,
+                                         passive: false,
+                                         noWait: false,
+                                         CancellationToken.None);
 
-                channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
+                await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false, CancellationToken.None);
 
-                var msg = channel.BasicGet(_queueNameProvider.GetQueueName(queue), false);
+                var msg = await channel.BasicGetAsync(_queueNameProvider.GetQueueName(queue), autoAck: false, CancellationToken.None);
                 if (msg != null)
                 {
                     var data = Encoding.UTF8.GetString(msg.Body.ToArray());
-                    channel.BasicAck(msg.DeliveryTag, false);
+                    await channel.BasicAckAsync(msg.DeliveryTag, multiple: false, CancellationToken.None);
                     return data;
                 }
                 return null;
             }
+            finally
+            {
+                await channel.CloseAsync(200, "OK", abort: false, CancellationToken.None);
+            }
         }
 
         public void Dispose()
         {
             if (_connection != null)
             {
                 if (_connection.IsOpen)
-                    _connection.Close();
+                    _connection.CloseAsync(200, "OK", TimeSpan.FromSeconds(10), abort: false, CancellationToken.None).GetAwaiter().GetResult();
             }
         }
 
         public async Task Start()
         {
-            _connection = _rabbitMqConnectionFactory(_serviceProvider, "Workflow-Core");
+            _connection = await _rabbitMqConnectionFactory(_serviceProvider, "Workflow-Core");
         }
 
         public async Task Stop()
         {
             if (_connection != null)
             {
-                _connection.Close();
+                await _connection.CloseAsync(200, "OK", TimeSpan.FromSeconds(10), abort: false, CancellationToken.None);
                 _connection = null;
             }
         }

src/providers/WorkflowCore.QueueProviders.RabbitMQ/WorkflowCore.QueueProviders.RabbitMQ.csproj

@@ -23,7 +23,7 @@
 
   <ItemGroup>
     <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
-    <PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
+    <PackageReference Include="RabbitMQ.Client" Version="7.0.0" />
   </ItemGroup>
 
 </Project>

test/WorkflowCore.Tests.QueueProviders.RabbitMQ/WorkflowCore.Tests.QueueProviders.RabbitMQ.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
     <PropertyGroup>
-      <TargetFrameworks>net6.0</TargetFrameworks>
+      <TargetFrameworks>net8.0</TargetFrameworks>
     </PropertyGroup>
 
     <ItemGroup>