FEAT: Adding new changes

Author: Felipe Mattioli

src/AzureServiceBusFlow/Abstractions/ICommandProducer.cs

@@ -4,7 +4,23 @@ namespace AzureServiceBusFlow.Abstractions;
 
 public interface ICommandProducer<in TCommand> where TCommand : class, IServiceBusMessage
 {
+    /// <summary>
+    /// Produces the command without additional options.
+    /// </summary>
     Task ProduceCommandAsync(TCommand command, CancellationToken cancellationToken);
 
+    /// <summary>
+    /// Produces the command using explicit message options such as delay or application properties.
+    /// </summary>
     Task ProduceCommandAsync(TCommand command, MessageOptions messageOptions, CancellationToken cancellationToken);
+
+    /// <summary>
+    /// Produces the command with a set of application properties that will be attached to the message.
+    /// </summary>
+    Task ProduceCommandAsync(TCommand command, IDictionary<string, object> applicationProperties, CancellationToken cancellationToken);
+
+    /// <summary>
+    /// Produces the command with a delivery delay before it becomes available for processing.
+    /// </summary>
+    Task ProduceCommandAsync(TCommand command, TimeSpan delay, CancellationToken cancellationToken);
 }

src/AzureServiceBusFlow/Abstractions/IServiceBusProducer.cs

@@ -47,7 +47,7 @@ Task ProduceAsync(
         /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
         Task ProduceAsync(
             TMessage message,
-            IDictionary<string, string> applicationProperties,
+            IDictionary<string, object> applicationProperties,
             CancellationToken cancellationToken);
     }
 }

src/AzureServiceBusFlow/Models/MessageOptions.cs

@@ -1,4 +1,4 @@
 namespace AzureServiceBusFlow.Models
 {
-    public record MessageOptions(TimeSpan? Delay, IDictionary<string, string>? ApplicationProperties);
+    public record MessageOptions(TimeSpan? Delay, IDictionary<string, object>? ApplicationProperties);
 }

src/AzureServiceBusFlow/Producers/CommandProducer.cs

@@ -35,7 +35,7 @@ public Task ProduceCommandAsync(TCommand command, TimeSpan delay, CancellationTo
         /// <summary>
         /// Produces a command with application properties.
         /// </summary>
-        public Task ProduceCommandAsync(TCommand command, IDictionary<string, string> applicationProperties, CancellationToken cancellationToken)
+        public Task ProduceCommandAsync(TCommand command, IDictionary<string, object> applicationProperties, CancellationToken cancellationToken)
         {
             return _producer.ProduceAsync(command, applicationProperties, cancellationToken);
         }

src/AzureServiceBusFlow/Producers/ServiceBusProducer.cs

@@ -15,15 +15,29 @@ public ServiceBusProducer(AzureServiceBusConfiguration azureServiceBusConfigurat
         {
             var client = new ServiceBusClient(azureServiceBusConfiguration.ConnectionString);
             _sender = client.CreateSender(queueOrTopicName);
+
             _logger = logger;
         }
 
         public async Task ProduceAsync(TMessage message, CancellationToken cancellationToken)
         {
-            await ProduceAsync(message, producerOptions: null, cancellationToken);
+            var json = JsonConvert.SerializeObject(message);
+            var serviceBusMessage = new ServiceBusMessage(json)
+            {
+                Subject = message.RoutingKey,
+                ApplicationProperties =
+                {
+                    { "MessageType", message.GetType().FullName },
+                    { "CreatedAt", message.CreatedDate.ToString("O") }
+                }
+            };
+
+            await _sender.SendMessageAsync(serviceBusMessage, cancellationToken);
+
+            _logger.LogInformation("Message {MessageType} published with successfully!", message.GetType().Name);
         }
 
-        public async Task ProduceAsync(TMessage message, MessageOptions? producerOptions, CancellationToken cancellationToken)
+        public Task ProduceAsync(TMessage message, MessageOptions producerOptions, CancellationToken cancellationToken)
         {
             var json = JsonConvert.SerializeObject(message);
             var serviceBusMessage = new ServiceBusMessage(json)
@@ -38,29 +52,26 @@ public async Task ProduceAsync(TMessage message, MessageOptions? producerOptions
 
             if (producerOptions?.ApplicationProperties is not null)
             {
-                foreach (var kvp in producerOptions.ApplicationProperties)
-                {
-                    if (!serviceBusMessage.ApplicationProperties.ContainsKey(kvp.Key))
-                        serviceBusMessage.ApplicationProperties.Add(kvp.Key, kvp.Value);
-                }
+                producerOptions?.ApplicationProperties?
+                    .Where(kvp => !serviceBusMessage.ApplicationProperties.ContainsKey(kvp.Key))
+                    .ToList()
+                    .ForEach(kvp => serviceBusMessage.ApplicationProperties.Add(kvp.Key, kvp.Value));
             }
 
             if (producerOptions?.Delay is not null)
             {
                 serviceBusMessage.ScheduledEnqueueTime = DateTimeOffset.UtcNow.Add(producerOptions.Delay.Value);
             }
 
-            await _sender.SendMessageAsync(serviceBusMessage, cancellationToken);
-
-            _logger.LogInformation("Message {MessageType} published successfully!", message.GetType().Name);
+            return _sender.SendMessageAsync(serviceBusMessage, cancellationToken);
         }
 
         public Task ProduceAsync(TMessage message, TimeSpan delay, CancellationToken cancellationToken)
         {
             return ProduceAsync(message, new MessageOptions(delay, null), cancellationToken);
         }
 
-        public Task ProduceAsync(TMessage message, IDictionary<string, string> applicationProperties, CancellationToken cancellationToken)
+        public Task ProduceAsync(TMessage message, IDictionary<string, object> applicationProperties, CancellationToken cancellationToken)
         {
             return ProduceAsync(message, new MessageOptions(null, applicationProperties), cancellationToken);
         }