Refactor RabbitMQ client to use async methods

Updated RabbitMQ client implementation to use asynchronous methods and interfaces, improving performance and scalability. Replaced `Send` with `SendAsync` in `IRabbitMQClient`. Updated `RabbitMQClient`, `RabbitMqExchangeInitializer`, `RabbitMqBackgroundService`, and `Extensions` classes to use async methods. Renamed `IRabbitMqSerializer` to `IRabbitMQSerializer` for consistency. Updated `ConcurrentDictionary` to use `IChannel` instead of `IModel`. Upgraded `RabbitMQ.Client` package to 7.0.0. Improved null safety with nullable reference types. Made minor logging and formatting improvements. Ensured consistent camel casing for RabbitMQ-related names.

Author: giovanninocco

src/Genocs.MessageBrokers.RabbitMQ/Clients/RabbitMqClient.cs

@@ -6,22 +6,25 @@ namespace Genocs.MessageBrokers.RabbitMQ.Clients;
 
 internal sealed class RabbitMQClient : IRabbitMQClient
 {
-    private readonly object _lockObject = new();
     private const string EmptyContext = "{}";
     private readonly IConnection _connection;
     private readonly IContextProvider _contextProvider;
-    private readonly IRabbitMqSerializer _serializer;
+    private readonly IRabbitMQSerializer _serializer;
     private readonly ILogger<RabbitMQClient> _logger;
     private readonly bool _contextEnabled;
     private readonly bool _loggerEnabled;
     private readonly string _spanContextHeader;
     private readonly bool _persistMessages;
-    private int _channelsCount;
-    private readonly ConcurrentDictionary<int, IModel> _channels = new();
+    private readonly ConcurrentDictionary<int, IChannel> _channels = new();
     private readonly int _maxChannels;
+    private int _channelsCount;
 
-    public RabbitMQClient(ProducerConnection connection, IContextProvider contextProvider, IRabbitMqSerializer serializer,
-        RabbitMQOptions options, ILogger<RabbitMQClient> logger)
+    public RabbitMQClient(
+                            ProducerConnection connection,
+                            IContextProvider contextProvider,
+                            IRabbitMQSerializer serializer,
+                            RabbitMQOptions options,
+                            ILogger<RabbitMQClient> logger)
     {
         _connection = connection.Connection;
         _contextProvider = contextProvider;
@@ -34,29 +37,34 @@ public RabbitMQClient(ProducerConnection connection, IContextProvider contextPro
         _maxChannels = options.MaxProducerChannels <= 0 ? 1000 : options.MaxProducerChannels;
     }
 
-    public void Send(object message, IConventions conventions, string messageId = null, string correlationId = null,
-        string spanContext = null, object messageContext = null, IDictionary<string, object> headers = null)
+    public async Task SendAsync(
+                                    object message,
+                                    IConventions conventions,
+                                    string? messageId = null,
+                                    string? correlationId = null,
+                                    string? spanContext = null,
+                                    object? messageContext = null,
+                                    IDictionary<string, object>? headers = null)
     {
-        var threadId = Thread.CurrentThread.ManagedThreadId;
+        int threadId = Thread.CurrentThread.ManagedThreadId;
         if (!_channels.TryGetValue(threadId, out var channel))
         {
-            lock (_lockObject)
+            if (_channelsCount >= _maxChannels)
             {
-                if (_channelsCount >= _maxChannels)
-                {
-                    throw new InvalidOperationException($"Cannot create RabbitMQ producer channel for thread: {threadId} " +
-                                                        $"(reached the limit of {_maxChannels} channels). " +
-                                                        "Modify `MaxProducerChannels` setting to allow more channels.");
-                }
+                throw new InvalidOperationException($"Cannot create RabbitMQ producer channel for thread: {threadId} " +
+                                                    $"(reached the limit of {_maxChannels} channels). " +
+                                                    "Modify `MaxProducerChannels` setting to allow more channels.");
 
-                channel = _connection.CreateModel();
-                _channels.TryAdd(threadId, channel);
-                _channelsCount++;
-                if (_loggerEnabled)
-                {
-                    _logger.LogTrace($"Created a channel for thread: {threadId}, total channels: {_channelsCount}/{_maxChannels}");
-                }
             }
+
+            channel = await _connection.CreateChannelAsync();
+            _channels.TryAdd(threadId, channel);
+            _channelsCount++;
+            if (_loggerEnabled)
+            {
+                _logger.LogTrace($"Created a channel for thread: {threadId}, total channels: {_channelsCount}/{_maxChannels}");
+            }
+
         }
         else
         {
@@ -67,16 +75,23 @@ public void Send(object message, IConventions conventions, string messageId = nu
         }
 
         var body = _serializer.Serialize(message);
-        var properties = channel.CreateBasicProperties();
+
+        BasicProperties properties = new BasicProperties();
+
+        // var properties = channel.BasicProperties();
         properties.Persistent = _persistMessages;
+
         properties.MessageId = string.IsNullOrWhiteSpace(messageId)
             ? Guid.NewGuid().ToString("N")
             : messageId;
+
         properties.CorrelationId = string.IsNullOrWhiteSpace(correlationId)
             ? Guid.NewGuid().ToString("N")
             : correlationId;
+
         properties.Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds());
-        properties.Headers = new Dictionary<string, object>();
+
+        properties.Headers = new Dictionary<string, object?>();
 
         if (_contextEnabled)
         {
@@ -108,11 +123,17 @@ public void Send(object message, IConventions conventions, string messageId = nu
                              $"[id: '{properties.MessageId}', correlation id: '{properties.CorrelationId}']");
         }
 
-        channel.BasicPublish(conventions.Exchange, conventions.RoutingKey, properties, body.ToArray());
+        await channel.BasicPublishAsync(conventions.Exchange, conventions.RoutingKey, true, properties, body.ToArray());
     }
 
-    private void IncludeMessageContext(object context, IBasicProperties properties)
+    private void IncludeMessageContext(object? context, IBasicProperties properties)
     {
+        if (context is null)
+            return;
+
+        if (properties.Headers is null)
+            return;
+
         if (context is not null)
         {
             properties.Headers.Add(_contextProvider.HeaderName, _serializer.Serialize(context).ToArray());

src/Genocs.MessageBrokers.RabbitMQ/Contexts/ContextProvider.cs

@@ -2,25 +2,25 @@ namespace Genocs.MessageBrokers.RabbitMQ.Contexts;
 
 internal sealed class ContextProvider : IContextProvider
 {
-    private readonly IRabbitMqSerializer _serializer;
+    private readonly IRabbitMQSerializer _serializer;
     public string HeaderName { get; }
 
-    public ContextProvider(IRabbitMqSerializer serializer, RabbitMQOptions options)
+    public ContextProvider(IRabbitMQSerializer serializer, RabbitMQOptions options)
     {
         _serializer = serializer;
         HeaderName = string.IsNullOrWhiteSpace(options.Context?.Header)
             ? "message_context"
             : options.Context.Header;
     }
 
-    public object Get(IDictionary<string, object> headers)
+    public object? Get(IDictionary<string, object>? headers)
     {
         if (headers is null)
         {
             return null;
         }
 
-        if (!headers.TryGetValue(HeaderName, out var context))
+        if (!headers.TryGetValue(HeaderName, out object? context))
         {
             return null;
         }

src/Genocs.MessageBrokers.RabbitMQ/Extensions.cs

@@ -35,12 +35,12 @@ public static class Extensions
     /// <param name="serializer">The serializer.</param>
     /// <returns></returns>
     /// <exception cref="ArgumentException">Raised when configuration is incorrect.</exception>
-    public static IGenocsBuilder AddRabbitMq(
+    public static async Task<IGenocsBuilder> AddRabbitMQAsync(
                                             this IGenocsBuilder builder,
                                             string sectionName = SectionName,
                                             Func<IRabbitMqPluginsRegistry, IRabbitMqPluginsRegistry>? plugins = null,
                                             Action<ConnectionFactory>? connectionFactoryConfigurator = null,
-                                            IRabbitMqSerializer? serializer = null)
+                                            IRabbitMQSerializer? serializer = null)
     {
         if (string.IsNullOrWhiteSpace(sectionName))
         {
@@ -51,7 +51,7 @@ public static IGenocsBuilder AddRabbitMq(
         builder.Services.AddSingleton(options);
         if (!builder.TryRegister(RegistryName))
         {
-            return builder;
+            return await Task.FromResult(builder);
         }
 
         if (options.HostNames is null || !options.HostNames.Any())
@@ -72,8 +72,8 @@ public static IGenocsBuilder AddRabbitMq(
         builder.Services.AddSingleton<IConventionsProvider, ConventionsProvider>();
         builder.Services.AddSingleton<IConventionsRegistry, ConventionsRegistry>();
         builder.Services.AddSingleton<IRabbitMQClient, RabbitMQClient>();
-        builder.Services.AddSingleton<IBusPublisher, RabbitMqPublisher>();
-        builder.Services.AddSingleton<IBusSubscriber, RabbitMqSubscriber>();
+        builder.Services.AddSingleton<IBusPublisher, RabbitMQPublisher>();
+        builder.Services.AddSingleton<IBusSubscriber, RabbitMQSubscriber>();
         builder.Services.AddSingleton<MessageSubscribersChannel>();
         builder.Services.AddTransient<RabbitMqExchangeInitializer>();
         builder.Services.AddHostedService<RabbitMqBackgroundService>();
@@ -85,7 +85,7 @@ public static IGenocsBuilder AddRabbitMq(
         }
         else
         {
-            builder.Services.AddSingleton<IRabbitMqSerializer, SystemTextJsonJsonRabbitMqSerializer>();
+            builder.Services.AddSingleton<IRabbitMQSerializer, SystemTextJsonJsonRabbitMQSerializer>();
         }
 
         var pluginsRegistry = new RabbitMqPluginsRegistry();
@@ -105,7 +105,6 @@ public static IGenocsBuilder AddRabbitMq(
             SocketWriteTimeout = options.SocketWriteTimeout,
             RequestedChannelMax = options.RequestedChannelMax,
             RequestedFrameMax = options.RequestedFrameMax,
-            DispatchConsumersAsync = true,
             ContinuationTimeout = options.ContinuationTimeout,
             HandshakeContinuationTimeout = options.HandshakeContinuationTimeout,
             NetworkRecoveryInterval = options.NetworkRecoveryInterval,
@@ -117,8 +116,8 @@ public static IGenocsBuilder AddRabbitMq(
         connectionFactoryConfigurator?.Invoke(connectionFactory);
 
         logger.LogDebug($"Connecting to RabbitMQ: '{string.Join(", ", options.HostNames)}'...");
-        var consumerConnection = connectionFactory.CreateConnection(options.HostNames.ToList(), $"{options.ConnectionName}_consumer");
-        var producerConnection = connectionFactory.CreateConnection(options.HostNames.ToList(), $"{options.ConnectionName}_producer");
+        var consumerConnection = await connectionFactory.CreateConnectionAsync(options.HostNames.ToList(), $"{options.ConnectionName}_consumer");
+        var producerConnection = await connectionFactory.CreateConnectionAsync(options.HostNames.ToList(), $"{options.ConnectionName}_producer");
         logger.LogDebug($"Connected to RabbitMQ: '{string.Join(", ", options.HostNames)}'.");
         builder.Services.AddSingleton(new ConsumerConnection(consumerConnection));
         builder.Services.AddSingleton(new ProducerConnection(producerConnection));
@@ -211,6 +210,6 @@ public static IGenocsBuilder AddExceptionToFailedMessageMapper<T>(this IGenocsBu
         return builder;
     }
 
-    public static IBusSubscriber UseRabbitMq(this IApplicationBuilder app)
-        => new RabbitMqSubscriber(app.ApplicationServices.GetRequiredService<MessageSubscribersChannel>());
+    public static IBusSubscriber UseRabbitMQ(this IApplicationBuilder app)
+        => new RabbitMQSubscriber(app.ApplicationServices.GetRequiredService<MessageSubscribersChannel>());
 }

src/Genocs.MessageBrokers.RabbitMQ/Genocs.MessageBrokers.RabbitMQ.csproj

@@ -28,7 +28,7 @@
     <ItemGroup>
         <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
         <PackageReference Include="Polly" Version="8.5.0" />
-        <PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
+        <PackageReference Include="RabbitMQ.Client" Version="7.0.0" />
     </ItemGroup>
 
 

src/Genocs.MessageBrokers.RabbitMQ/IRabbitMqClient.cs

@@ -2,12 +2,12 @@ namespace Genocs.MessageBrokers.RabbitMQ;
 
 public interface IRabbitMQClient
 {
-    void Send(
-                object? message,
-                IConventions conventions,
-                string? messageId = null,
-                string? correlationId = null,
-                string? spanContext = null,
-                object? messageContext = null,
-                IDictionary<string, object>? headers = null);
+    Task SendAsync(
+                    object message,
+                    IConventions conventions,
+                    string? messageId = null,
+                    string? correlationId = null,
+                    string? spanContext = null,
+                    object? messageContext = null,
+                    IDictionary<string, object>? headers = null);
 }

src/Genocs.MessageBrokers.RabbitMQ/IRabbitMqSerializer.cs

@@ -1,6 +1,6 @@
 namespace Genocs.MessageBrokers.RabbitMQ;
 
-public interface IRabbitMqSerializer
+public interface IRabbitMQSerializer
 {
     ReadOnlySpan<byte> Serialize(object value);
     object? Deserialize(ReadOnlySpan<byte> value, Type type);

src/Genocs.MessageBrokers.RabbitMQ/Initializers/RabbitMqExchangeInitializer.cs

@@ -1,7 +1,7 @@
+using System.Reflection;
 using Genocs.Common.Types;
 using Microsoft.Extensions.Logging;
 using RabbitMQ.Client;
-using System.Reflection;
 
 namespace Genocs.MessageBrokers.RabbitMQ.Initializers;
 
@@ -13,16 +13,18 @@ public class RabbitMqExchangeInitializer : IInitializer
     private readonly ILogger<RabbitMqExchangeInitializer> _logger;
     private readonly bool _loggerEnabled;
 
-    public RabbitMqExchangeInitializer(ProducerConnection connection, RabbitMQOptions options,
-        ILogger<RabbitMqExchangeInitializer> logger)
+    public RabbitMqExchangeInitializer(
+                                        ProducerConnection connection,
+                                        RabbitMQOptions options,
+                                        ILogger<RabbitMqExchangeInitializer> logger)
     {
         _connection = connection.Connection;
         _options = options;
         _logger = logger;
         _loggerEnabled = _options.Logger?.Enabled == true;
     }
 
-    public Task InitializeAsync()
+    public async Task InitializeAsync()
     {
         var exchanges = AppDomain.CurrentDomain
             .GetAssemblies()
@@ -32,41 +34,41 @@ public Task InitializeAsync()
             .Distinct()
             .ToList();
 
-        using var channel = _connection.CreateModel();
+        using var channel = await _connection.CreateChannelAsync();
         if (_options.Exchange?.Declare == true)
         {
             Log(_options.Exchange.Name, _options.Exchange.Type);
 
-            channel.ExchangeDeclare(
-                                    _options.Exchange.Name,
-                                    _options.Exchange.Type,
-                                    _options.Exchange.Durable,
-                                    _options.Exchange.AutoDelete);
+            await channel.ExchangeDeclareAsync(
+                                                _options.Exchange.Name,
+                                                _options.Exchange.Type,
+                                                _options.Exchange.Durable,
+                                                _options.Exchange.AutoDelete);
 
             if (_options.DeadLetter?.Enabled is true && _options.DeadLetter?.Declare is true)
             {
-                channel.ExchangeDeclare(
-                                        $"{_options.DeadLetter.Prefix}{_options.Exchange.Name}{_options.DeadLetter.Suffix}",
-                                        ExchangeType.Direct,
-                                        _options.Exchange.Durable,
-                                        _options.Exchange.AutoDelete);
+                await channel.ExchangeDeclareAsync(
+                                                    $"{_options.DeadLetter.Prefix}{_options.Exchange.Name}{_options.DeadLetter.Suffix}",
+                                                    ExchangeType.Direct,
+                                                    _options.Exchange.Durable,
+                                                    _options.Exchange.AutoDelete);
             }
         }
 
         foreach (string? exchange in exchanges)
         {
+            if (string.IsNullOrWhiteSpace(exchange)) continue;
+
             if (exchange.Equals(_options.Exchange?.Name, StringComparison.InvariantCultureIgnoreCase))
             {
                 continue;
             }
 
             Log(exchange, DefaultType);
-            channel.ExchangeDeclare(exchange, DefaultType, true);
+            await channel.ExchangeDeclareAsync(exchange, DefaultType, true);
         }
 
-        channel.Close();
-
-        return Task.CompletedTask;
+        await channel.CloseAsync();
     }
 
     private void Log(string exchange, string type)