Merge pull request #5 from mehyaa/rabbitmq-upgrade

RabbitMQ.Client package upgraded

Author: mehyaa

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

@@ -4,14 +4,16 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace Convey.MessageBrokers.RabbitMQ.Clients;
 
 internal sealed class RabbitMqClient : IRabbitMqClient
 {
     private const string EmptyContext = "{}";
 
-    private readonly object _lockObject = new();
+    private readonly SemaphoreSlim _semaphore = new(1, 1);
 
     private readonly AppOptions _appOptions;
     private readonly IConnection _connection;
@@ -24,7 +26,7 @@ internal sealed class RabbitMqClient : IRabbitMqClient
     private readonly bool _persistMessages;
     private readonly int _maxChannels;
 
-    private readonly ConcurrentDictionary<int, IModel> _channels = new();
+    private readonly ConcurrentDictionary<int, IChannel> _channels = new();
 
     private int _channelsCount;
 
@@ -48,21 +50,24 @@ public RabbitMqClient(
         _maxChannels = options.MaxProducerChannels <= 0 ? 1000 : options.MaxProducerChannels;
     }
 
-    public void Send(
+    public async Task SendAsync(
         object message,
         IConvention convention,
         string messageId = null,
         string correlationId = null,
         string spanContext = null,
         object messageContext = null,
-        IDictionary<string, object> headers = null)
+        IDictionary<string, object> headers = null,
+        CancellationToken cancellationToken = default)
     {
         var threadId = Environment.CurrentManagedThreadId;
 
         if (!_channels.TryGetValue(threadId, out var channel))
         {
-            lock (_lockObject)
+            try
             {
+                await _semaphore.WaitAsync(cancellationToken);
+
                 if (_channelsCount >= _maxChannels)
                 {
                     throw new InvalidOperationException(
@@ -71,7 +76,7 @@ public void Send(
                         "Modify `MaxProducerChannels` setting to allow more channels.");
                 }
 
-                channel = _connection.CreateModel();
+                channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
                 _channels.TryAdd(threadId, channel);
                 _channelsCount++;
 
@@ -84,6 +89,10 @@ public void Send(
                         _maxChannels);
                 }
             }
+            finally
+            {
+                _semaphore.Release();
+            }
         }
         else
         {
@@ -97,17 +106,18 @@ public void Send(
             }
         }
 
-        var properties = channel.CreateBasicProperties();
-
-        properties.AppId = _appOptions.Service;
-        properties.ContentEncoding = _serializer.ContentEncoding;
-        properties.ContentType = _serializer.ContentType;
-        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.Type = convention.Type?.Name;
-        properties.Headers = new Dictionary<string, object>();
+        var properties = new BasicProperties
+        {
+            AppId = _appOptions.Service,
+            ContentEncoding = _serializer.ContentEncoding,
+            ContentType = _serializer.ContentType,
+            Persistent = _persistMessages,
+            MessageId = string.IsNullOrWhiteSpace(messageId) ? Guid.NewGuid().ToString("N") : messageId,
+            CorrelationId = string.IsNullOrWhiteSpace(correlationId) ? Guid.NewGuid().ToString("N") : correlationId,
+            Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()),
+            Type = convention.Type?.Name,
+            Headers = new Dictionary<string, object>()
+        };
 
         if (_contextEnabled)
         {
@@ -144,11 +154,22 @@ public void Send(
 
         var body = _serializer.Serialize(message);
 
-        channel.BasicPublish(convention.Exchange, convention.RoutingKey, properties, body.ToArray());
+        await channel.BasicPublishAsync(
+            exchange: convention.Exchange,
+            routingKey: convention.RoutingKey,
+            mandatory: false,
+            basicProperties: properties,
+            body: body.ToArray(),
+            cancellationToken: cancellationToken);
     }
 
     private void IncludeMessageContext(object context, IBasicProperties properties)
     {
+        if (properties?.Headers is null)
+        {
+            return;
+        }
+
         if (context is not null)
         {
             properties.Headers.Add(_contextProvider.HeaderName, _serializer.Serialize(context).ToArray());

src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/Convey.MessageBrokers.RabbitMQ.csproj

@@ -13,7 +13,7 @@
     <PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="8.0.1" />
     <PackageReference Include="Microsoft.Extensions.Logging" Version="8.0.1" />
     <PackageReference Include="Polly" Version="8.5.2" />
-    <PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
+    <PackageReference Include="RabbitMQ.Client" Version="7.1.2" />
   </ItemGroup>
 
   <ItemGroup>

src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/Extensions.cs

@@ -100,7 +100,6 @@ public static IConveyBuilder AddRabbitMq(
             SocketWriteTimeout = options.SocketWriteTimeout,
             RequestedChannelMax = options.RequestedChannelMax,
             RequestedFrameMax = options.RequestedFrameMax,
-            DispatchConsumersAsync = true,
             ContinuationTimeout = options.ContinuationTimeout,
             HandshakeContinuationTimeout = options.HandshakeContinuationTimeout,
             NetworkRecoveryInterval = options.NetworkRecoveryInterval,
@@ -113,8 +112,8 @@ public static IConveyBuilder AddRabbitMq(
         connectionFactoryConfigurator?.Invoke(connectionFactory);
 
         logger.LogDebug("Connecting to RabbitMQ: '{Hostnames}'...", 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 = connectionFactory.CreateConnectionAsync(options.HostNames.ToList(), $"{options.ConnectionName}_consumer").GetAwaiter().GetResult();
+        var producerConnection = connectionFactory.CreateConnectionAsync(options.HostNames.ToList(), $"{options.ConnectionName}_producer").GetAwaiter().GetResult();
         logger.LogDebug("Connected to RabbitMQ: '{Hostnames}'", string.Join(", ", options.HostNames));
 
         builder.Services.AddSingleton(new ConsumerConnection(consumerConnection));

src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/IRabbitMqClient.cs

@@ -1,15 +1,18 @@
 using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace Convey.MessageBrokers.RabbitMQ;
 
 public interface IRabbitMqClient
 {
-    void Send(
+    Task SendAsync(
         object message,
         IConvention convention,
         string messageId = null,
         string correlationId = null,
         string spanContext = null,
         object messageContext = null,
-        IDictionary<string, object> headers = null);
+        IDictionary<string, object> headers = null,
+        CancellationToken cancellationToken = default);
 }

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

@@ -31,37 +31,39 @@ public RabbitMqExchangeInitializer(
         _loggerEnabled = _options.Logger?.Enabled == true;
     }
 
-    public Task InitializeAsync(CancellationToken cancellationToken)
+    public async Task InitializeAsync(CancellationToken cancellationToken)
     {
         var exchanges =
             AppDomain.CurrentDomain
                 .GetAssemblies()
                 .SelectMany(a => a.GetTypes())
                 .Where(t => t.IsDefined(typeof(MessageAttribute), false))
-                .Select(t => t.GetCustomAttribute<MessageAttribute>().Exchange)
+                .Select(t => t.GetCustomAttribute<MessageAttribute>()?.Exchange)
                 .Distinct()
                 .ToArray();
 
-        using var channel = _connection.CreateModel();
+        await using var channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
 
         if (_options.Exchange?.Declare == true)
         {
             Log(_options.Exchange.Name, _options.Exchange.Type);
 
-            channel.ExchangeDeclare(
+            await channel.ExchangeDeclareAsync(
                 _options.Exchange.Name,
                 _options.Exchange.Type,
                 _options.Exchange.Durable,
-                _options.Exchange.AutoDelete);
+                _options.Exchange.AutoDelete,
+                cancellationToken: cancellationToken);
 
             if (_options.DeadLetter?.Enabled is true &&
                 _options.DeadLetter?.Declare is true)
             {
-                channel.ExchangeDeclare(
+                await channel.ExchangeDeclareAsync(
                     $"{_options.DeadLetter.Prefix}{_options.Exchange.Name}{_options.DeadLetter.Suffix}",
                     ExchangeType.Direct,
                     _options.Exchange.Durable,
-                    _options.Exchange.AutoDelete);
+                    _options.Exchange.AutoDelete,
+                    cancellationToken: cancellationToken);
             }
         }
 
@@ -74,12 +76,10 @@ public Task InitializeAsync(CancellationToken cancellationToken)
 
             Log(exchange, DefaultType);
 
-            channel.ExchangeDeclare(exchange, DefaultType, true);
+            await channel.ExchangeDeclareAsync(exchange, DefaultType, true, cancellationToken: cancellationToken);
         }
 
-        channel.Close();
-
-        return Task.CompletedTask;
+        await channel.CloseAsync(cancellationToken: cancellationToken);
     }
 
     private void Log(string exchange, string type)