FEAT: Adding new features

Author: fmattioli

src/AzureServiceBusFlow/Abstractions/ICommandProducer.cs

@@ -1,6 +1,10 @@
-namespace AzureServiceBusFlow.Abstractions;
+using AzureServiceBusFlow.Models;
+
+namespace AzureServiceBusFlow.Abstractions;
 
 public interface ICommandProducer<in TCommand> where TCommand : class, IServiceBusMessage
 {
     Task ProduceCommandAsync(TCommand command, CancellationToken cancellationToken);
+
+    Task ProduceCommandAsync(TCommand command, MessageOptions messageOptions, CancellationToken cancellationToken);
 }

src/AzureServiceBusFlow/Abstractions/IServiceBusProducer.cs

@@ -1,7 +1,14 @@
-namespace AzureServiceBusFlow.Abstractions
+using AzureServiceBusFlow.Models;
+
+namespace AzureServiceBusFlow.Abstractions
 {
     public interface IServiceBusProducer<in TMessage> where TMessage : class, IServiceBusMessage
     {
         Task ProduceAsync(TMessage message, CancellationToken cancellationToken);
+
+        Task ProduceAsync(
+            TMessage message,
+            MessageOptions producerOptions,
+            CancellationToken cancellationToken);
     }
 }

src/AzureServiceBusFlow/Models/MessageOptions.cs

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

src/AzureServiceBusFlow/Producers/CommandProducer.cs

@@ -1,4 +1,5 @@
 using AzureServiceBusFlow.Abstractions;
+using AzureServiceBusFlow.Models;
 
 namespace AzureServiceBusFlow.Producers
 {
@@ -11,5 +12,10 @@ public Task ProduceCommandAsync(TCommand command, CancellationToken cancellation
         {
             return _producer.ProduceAsync(command, cancellationToken);
         }
+
+        public Task ProduceCommandAsync(TCommand command, MessageOptions messageOptions, CancellationToken cancellationToken)
+        {
+            return _producer.ProduceAsync(command, messageOptions, cancellationToken);
+        }
     }
 }

src/AzureServiceBusFlow/Producers/ServiceBusProducer.cs

@@ -1,7 +1,6 @@
 using Azure.Messaging.ServiceBus;
 using AzureServiceBusFlow.Abstractions;
 using AzureServiceBusFlow.Models;
-
 using Microsoft.Extensions.Logging;
 using Newtonsoft.Json;
 
@@ -11,6 +10,7 @@ public class ServiceBusProducer<TMessage> : IServiceBusProducer<TMessage> where
     {
         private readonly ServiceBusSender _sender;
         private readonly ILogger _logger;
+
         public ServiceBusProducer(AzureServiceBusConfiguration azureServiceBusConfiguration, string queueOrTopicName, ILogger logger)
         {
             var client = new ServiceBusClient(azureServiceBusConfiguration.ConnectionString);
@@ -36,5 +36,35 @@ public async Task ProduceAsync(TMessage message, CancellationToken cancellationT
 
             _logger.LogInformation("Message {MessageType} published with successfully!", message.GetType().Name);
         }
+
+        public Task ProduceAsync(TMessage message, MessageOptions producerOptions, CancellationToken cancellationToken)
+        {
+            var json = JsonConvert.SerializeObject(message);
+            var serviceBusMessage = new ServiceBusMessage(json)
+            {
+                Subject = (message as IServiceBusMessage)?.RoutingKey ?? message.GetType().Name,
+                ApplicationProperties =
+                {
+                    { "MessageType", message.GetType().FullName },
+                    { "CreatedAt", (message as IServiceBusMessage)?.CreatedDate.ToString("O") ?? DateTime.UtcNow.ToString("O") }
+                }
+            };
+
+            if (producerOptions?.ApplicationProperties is not null)
+            {
+                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);
+            }
+
+            return _sender.SendMessageAsync(serviceBusMessage, cancellationToken);
+        }
+
     }
 }