Merge pull request #8 from mehyaa/feature/rabbitmq-serialization

Enhance RabbitMQ client with async serialization and content type sup…

Author: mehyaa

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

@@ -25,6 +25,8 @@ internal sealed class RabbitMqClient : IRabbitMqClient
     private readonly string _spanContextHeader;
     private readonly bool _persistMessages;
     private readonly int _maxChannels;
+    private readonly string _contentType;
+    private readonly string _contentEncoding;
 
     private readonly ConcurrentDictionary<int, IChannel> _channels = new();
 
@@ -48,6 +50,8 @@ public RabbitMqClient(
         _spanContextHeader = options.GetSpanContextHeader();
         _persistMessages = options?.MessagesPersisted ?? false;
         _maxChannels = options.MaxProducerChannels <= 0 ? 1000 : options.MaxProducerChannels;
+        _contentType = options.Publish?.ContentType ?? "application/json";
+        _contentEncoding = options.Publish?.ContentEncoding ?? "UTF-8";
     }
 
     public async Task SendAsync(
@@ -109,8 +113,8 @@ public async Task SendAsync(
         var properties = new BasicProperties
         {
             AppId = _appOptions.Service,
-            ContentEncoding = _serializer.ContentEncoding,
-            ContentType = _serializer.ContentType,
+            ContentEncoding = _contentEncoding,
+            ContentType = _contentType,
             Persistent = _persistMessages,
             MessageId = string.IsNullOrWhiteSpace(messageId) ? Guid.NewGuid().ToString("N") : messageId,
             CorrelationId = string.IsNullOrWhiteSpace(correlationId) ? Guid.NewGuid().ToString("N") : correlationId,
@@ -121,7 +125,7 @@ public async Task SendAsync(
 
         if (_contextEnabled)
         {
-            IncludeMessageContext(messageContext, properties);
+            await IncludeMessageContextAsync(messageContext, properties, cancellationToken);
         }
 
         if (!string.IsNullOrWhiteSpace(spanContext))
@@ -152,7 +156,7 @@ public async Task SendAsync(
                 properties.CorrelationId);
         }
 
-        var body = _serializer.Serialize(message);
+        var body = await _serializer.SerializeAsync(message, _contentType, cancellationToken);
 
         await channel.BasicPublishAsync(
             exchange: convention.Exchange,
@@ -163,7 +167,7 @@ await channel.BasicPublishAsync(
             cancellationToken: cancellationToken);
     }
 
-    private void IncludeMessageContext(object context, BasicProperties properties)
+    private async Task IncludeMessageContextAsync(object context, BasicProperties properties, CancellationToken cancellationToken)
     {
         if (properties?.Headers is null)
         {
@@ -172,7 +176,7 @@ private void IncludeMessageContext(object context, BasicProperties properties)
 
         if (context is not null)
         {
-            properties.Headers.Add(_contextProvider.HeaderName, _serializer.Serialize(context));
+            properties.Headers.Add(_contextProvider.HeaderName, await _serializer.SerializeAsync(context, _contentType, cancellationToken));
 
             return;
         }

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

@@ -1,4 +1,6 @@
 using System;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace Convey.MessageBrokers.RabbitMQ.Contexts;
 
@@ -18,7 +20,12 @@ public ContextProvider(IRabbitMqSerializer serializer, RabbitMqOptions options)
                 : options.Context.Header;
     }
 
-    public object Get(object message, Type messageType, MessageProperties messageProperties)
+    public async Task<object> GetAsync(
+        object message,
+        Type messageType,
+        MessageProperties messageProperties,
+        string contentType,
+        CancellationToken cancellationToken = default)
     {
         if (messageProperties.Headers is null)
         {
@@ -32,7 +39,7 @@ public object Get(object message, Type messageType, MessageProperties messagePro
 
         if (context is byte[] bytes)
         {
-            return _serializer.Deserialize(bytes);
+            return await _serializer.DeserializeAsync<object>(bytes, contentType, cancellationToken);
         }
 
         return null;

src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/Conventions/ConventionBuilder.cs

@@ -25,39 +25,39 @@ public ConventionBuilder(RabbitMqOptions options)
                 StringComparison.InvariantCultureIgnoreCase) == true;
     }
 
-    public string GetRoutingKey(Type type)
+    public string GetExchange(Type type)
     {
-        var routingKey = type.Name;
+        var exchange =
+            string.IsNullOrWhiteSpace(_options.Exchange?.Name)
+                ? type.Assembly.GetName().Name
+                : _options.Exchange.Name;
 
-        if (_options.Conventions?.MessageAttribute?.IgnoreRoutingKey is true)
+        if (_options.Conventions?.MessageAttribute?.IgnoreExchange is true)
         {
-            return WithCasing(routingKey);
+            return WithCasing(exchange);
         }
 
         var attribute = GeAttribute(type);
 
-        routingKey = string.IsNullOrWhiteSpace(attribute?.RoutingKey) ? routingKey : attribute.RoutingKey;
+        exchange = string.IsNullOrWhiteSpace(attribute?.Exchange) ? exchange : attribute.Exchange;
 
-        return WithCasing(routingKey);
+        return WithCasing(exchange);
     }
 
-    public string GetExchange(Type type)
+    public string GetRoutingKey(Type type)
     {
-        var exchange =
-            string.IsNullOrWhiteSpace(_options.Exchange?.Name)
-                ? type.Assembly.GetName().Name
-                : _options.Exchange.Name;
+        var routingKey = type.Name;
 
-        if (_options.Conventions?.MessageAttribute?.IgnoreExchange is true)
+        if (_options.Conventions?.MessageAttribute?.IgnoreRoutingKey is true)
         {
-            return WithCasing(exchange);
+            return WithCasing(routingKey);
         }
 
         var attribute = GeAttribute(type);
 
-        exchange = string.IsNullOrWhiteSpace(attribute?.Exchange) ? exchange : attribute.Exchange;
+        routingKey = string.IsNullOrWhiteSpace(attribute?.RoutingKey) ? routingKey : attribute.RoutingKey;
 
-        return WithCasing(exchange);
+        return WithCasing(routingKey);
     }
 
     public string GetQueue(Type type)

src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/Conventions/ConventionProvider.cs

@@ -16,8 +16,6 @@ public ConventionProvider(IConventionRegistry registry, IConventionBuilder build
         _builder = builder;
     }
 
-    public IConvention Get<T>() => Get(typeof(T));
-
     public IConvention Get(Type type)
     {
         if (_conventions.TryGetValue(type, out var convention))
@@ -29,8 +27,8 @@ public IConvention Get(Type type)
             _registry.Get(type) ??
             new MessageConvention(
                 type,
-                _builder.GetRoutingKey(type),
                 _builder.GetExchange(type),
+                _builder.GetRoutingKey(type),
                 _builder.GetQueue(type));
 
         _conventions.TryAdd(type, convention);

…geBrokers.RabbitMQ/ConventionRegistry.cs …bbitMQ/Conventions/ConventionRegistry.cssrc/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/ConventionRegistry.cs renamed to src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/Conventions/ConventionRegistry.cs src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/ConventionRegistry.cs renamed to src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/Conventions/ConventionRegistry.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Collections.Generic;
 
-namespace Convey.MessageBrokers.RabbitMQ;
+namespace Convey.MessageBrokers.RabbitMQ.Conventions;
 
 public class ConventionRegistry : IConventionRegistry
 {

src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/Conventions/MessageConvention.cs

@@ -5,15 +5,15 @@ namespace Convey.MessageBrokers.RabbitMQ.Conventions;
 public class MessageConvention : IConvention
 {
     public Type Type { get; }
-    public string RoutingKey { get; }
     public string Exchange { get; }
+    public string RoutingKey { get; }
     public string Queue { get; }
 
-    public MessageConvention(Type type, string routingKey, string exchange, string queue)
+    public MessageConvention(Type type, string exchange, string routingKey, string queue)
     {
         Type = type;
-        RoutingKey = routingKey;
         Exchange = exchange;
+        RoutingKey = routingKey;
         Queue = queue;
     }
 }

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

@@ -77,7 +77,7 @@ public static IConveyBuilder AddRabbitMq(
         }
         else
         {
-            builder.Services.AddSingleton<IRabbitMqSerializer, SystemTextJsonJsonRabbitMqSerializer>();
+            builder.Services.AddSingleton<IRabbitMqSerializer, DefaultRabbitMqSerializer>();
         }
 
         builder.Services.AddSingleton<IRabbitMqPluginsExecutor, RabbitMqPluginsExecutor>();
@@ -219,4 +219,7 @@ public static IConveyBuilder AddExceptionToFailedMessageMapper<T>(this IConveyBu
 
     public static IBusSubscriber UseRabbitMq(this IApplicationBuilder app)
         => app.ApplicationServices.GetRequiredService<IBusSubscriber>();
+    
+    public static IConvention Get<T>(this IConventionProvider conventionProvider)
+        => conventionProvider.Get(typeof(T));
 }

src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/IContextProvider.cs

@@ -1,9 +1,11 @@
 using System;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace Convey.MessageBrokers.RabbitMQ;
 
 public interface IContextProvider
 {
     string HeaderName { get; }
-    object Get(object message, Type messageType, MessageProperties messageProperties);
+    Task<object> GetAsync(object message, Type messageType, MessageProperties messageProperties, string contentType, CancellationToken cancellationToken = default);
 }

src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/IConvention.cs

@@ -5,7 +5,7 @@ namespace Convey.MessageBrokers.RabbitMQ;
 public interface IConvention
 {
     Type Type { get; }
-    string RoutingKey { get; }
     string Exchange { get; }
+    string RoutingKey { get; }
     string Queue { get; }
 }

src/Convey.MessageBrokers.RabbitMQ/src/Convey.MessageBrokers.RabbitMQ/IConventionBuilder.cs

@@ -4,7 +4,7 @@ namespace Convey.MessageBrokers.RabbitMQ;
 
 public interface IConventionBuilder
 {
-    string GetRoutingKey(Type type);
     string GetExchange(Type type);
+    string GetRoutingKey(Type type);
     string GetQueue(Type type);
 }