FEAT: Adding new logic to work with middlewares

Author: Felipe Mattioli Dos Santos (MSC Technology Italia)

src/AzureServiceBusFlow.Sample/AzureServiceBusFlow.Sample.csproj

@@ -14,9 +14,9 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Scalar.AspNetCore" Version="2.8.10" />
+    <PackageReference Include="Scalar.AspNetCore" Version="2.9.0" />
     <PackageReference Include="Swashbuckle.AspNetCore" Version="9.0.6" />
-    <PackageReference Include="Mattioli.Configurations" Version="1.59.0" />
+    <PackageReference Include="Mattioli.Configurations" Version="1.61.0" />
   </ItemGroup>
 
   <ItemGroup>

src/AzureServiceBusFlow.Sample/Commands/ExampleCommandHandler.cs

@@ -8,7 +8,6 @@ public class CommandExemple1Handler : IMessageHandler<ExampleCommand1>
     {
         public Task HandleAsync(ExampleCommand1 message, ServiceBusReceivedMessage rawMessage, CancellationToken cancellationToken)
         {
-            throw new Exception("Erro");
             return Task.CompletedTask;
         }
     }

src/AzureServiceBusFlow.Sample/Middlewares/AsbSampleMiddleware.cs

@@ -0,0 +1,28 @@
+using Azure.Messaging.ServiceBus;
+using AzureServiceBusFlow.Abstractions;
+using AzureServiceBusFlow.Middlewar;
+
+namespace AzureServiceBusFlow.Sample.Middlewares
+{
+    /// <summary>
+    /// Sample middleware that adds a custom property
+    /// to every message published to Azure Service Bus.
+    /// </summary>
+    public class AsbSampleMiddleware : IProducerMiddleware
+    {
+        public async Task InvokeAsync(ServiceBusMessage message, Func<Task> next)
+        {
+            // Add a custom property if it doesn't exist
+            if (!message.ApplicationProperties.ContainsKey("SampleMiddleware"))
+            {
+                message.ApplicationProperties["SampleMiddleware"] = "Executed";
+            }
+
+            // Add a UTC timestamp for debugging or tracking
+            message.ApplicationProperties["ProcessedAtUtc"] = DateTime.UtcNow.ToString("O");
+
+            // Continue the pipeline (invoke the next middleware or the actual send)
+            await next();
+        }
+    }
+}

src/AzureServiceBusFlow.Sample/Program.cs

@@ -2,7 +2,7 @@
 using AzureServiceBusFlow.Models;
 using AzureServiceBusFlow.Sample.Commands;
 using AzureServiceBusFlow.Sample.Events;
-
+using AzureServiceBusFlow.Sample.Middlewares;
 using Mattioli.Configurations.Transformers;
 
 using Scalar.AspNetCore;
@@ -27,6 +27,7 @@
     .AddProducer<ExampleCommand1>(p => p
         .EnsureQueueExists("command-queue-one")
         .WithCommandProducer()
+        //.UseMiddleware<AsbSampleMiddleware>()
         .ToQueue("command-queue-one"))
     .AddProducer<ExampleCommand2>(p => p
         .EnsureQueueExists("command-queue-two")

src/AzureServiceBusFlow/AzureServiceBusFlow.csproj

@@ -16,8 +16,8 @@
   <ItemGroup>
     <PackageReference Include="Azure.Messaging.ServiceBus" Version="7.20.1" />
     <PackageReference Include="Microsoft.Azure.ServiceBus" Version="5.2.0" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.9" />
-    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.9" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.10" />
+    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.10" />
     <PackageReference Include="Polly" Version="8.6.4" />
   </ItemGroup>
 

src/AzureServiceBusFlow/Builders/ServiceBusProducerConfigurationBuilder.cs

@@ -1,4 +1,5 @@
 using AzureServiceBusFlow.Abstractions;
+using AzureServiceBusFlow.Middlewar;
 using AzureServiceBusFlow.Models;
 using AzureServiceBusFlow.Producers;
 using Microsoft.Azure.ServiceBus.Management;
@@ -8,13 +9,21 @@
 namespace AzureServiceBusFlow.Builders
 {
     public class ServiceBusProducerConfigurationBuilder<TMessage>(AzureServiceBusConfiguration azureServiceBusConfiguration, IServiceCollection services)
-        where TMessage : class, IServiceBusMessage
+    where TMessage : class, IServiceBusMessage
     {
         private readonly AzureServiceBusConfiguration _azureServiceBusConfiguration = azureServiceBusConfiguration;
         private readonly IServiceCollection _services = services;
 
         private string? _topicName;
         private string? _queueName;
+        private readonly List<Type> _middlewares = [];
+
+        public ServiceBusProducerConfigurationBuilder<TMessage> UseMiddleware<TMiddleware>()
+            where TMiddleware : IProducerMiddleware
+        {
+            _middlewares.Add(typeof(TMiddleware));
+            return this;
+        }
 
         public ServiceBusProducerConfigurationBuilder<TMessage> WithTopic(string topic)
         {
@@ -80,29 +89,26 @@ public ServiceBusProducerConfigurationBuilder<TMessage> WithQueue(string queue)
 
         internal void Build()
         {
-            if (!string.IsNullOrEmpty(_queueName))
-            {
-                _services.AddSingleton<IServiceBusProducer<TMessage>>(sp =>
-                {
-                    var logger = sp.GetRequiredService<ILogger<ServiceBusProducer<TMessage>>>();
-                    return new ServiceBusProducer<TMessage>(_azureServiceBusConfiguration, _queueName, logger);
-                });
+            if (string.IsNullOrEmpty(_queueName) && string.IsNullOrEmpty(_topicName))
+                throw new InvalidOperationException("Either topic or queue name must be specified.");
 
-                return;
-            }
-
-            if (!string.IsNullOrEmpty(_topicName))
+            foreach (var middlewareType in _middlewares)
             {
-                _services.AddSingleton<IServiceBusProducer<TMessage>>(sp =>
-                {
-                    var logger = sp.GetRequiredService<ILogger<ServiceBusProducer<TMessage>>>();
-                    return new ServiceBusProducer<TMessage>(_azureServiceBusConfiguration, _topicName, logger);
-                });
-
-                return;
+                _services.AddSingleton(typeof(IProducerMiddleware), middlewareType);
             }
 
-            throw new InvalidOperationException("Either topic or queue name must be specified.");
+            _services.AddSingleton<IServiceBusProducer<TMessage>>(sp =>
+            {
+                var logger = sp.GetRequiredService<ILogger<ServiceBusProducer<TMessage>>>();
+                var middlewares = sp.GetServices<IProducerMiddleware>();
+                var name = _queueName ?? _topicName!;
+                return new ServiceBusProducer<TMessage>(
+                    _azureServiceBusConfiguration,
+                    name,
+                    logger,
+                    middlewares);
+            });
         }
     }
+
 }

src/AzureServiceBusFlow/Middlewar/IProducerMiddleware.cs

@@ -0,0 +1,10 @@
+using Azure.Messaging.ServiceBus;
+
+namespace AzureServiceBusFlow.Middlewar
+{
+    public interface IProducerMiddleware
+    {
+        Task InvokeAsync(ServiceBusMessage message, Func<Task> next);
+    }
+
+}

src/AzureServiceBusFlow/Producers/ServiceBusProducer.cs

@@ -1,22 +1,29 @@
 using Azure.Messaging.ServiceBus;
 using AzureServiceBusFlow.Abstractions;
+using AzureServiceBusFlow.Middlewar;
 using AzureServiceBusFlow.Models;
 using Microsoft.Extensions.Logging;
 using Newtonsoft.Json;
 
 namespace AzureServiceBusFlow.Producers
 {
-    public class ServiceBusProducer<TMessage> : IServiceBusProducer<TMessage> where TMessage : class, IServiceBusMessage
+    public class ServiceBusProducer<TMessage> : IServiceBusProducer<TMessage>
+        where TMessage : class, IServiceBusMessage
     {
         private readonly ServiceBusSender _sender;
         private readonly ILogger _logger;
+        private readonly IEnumerable<IProducerMiddleware>? _middlewares;
 
-        public ServiceBusProducer(AzureServiceBusConfiguration azureServiceBusConfiguration, string queueOrTopicName, ILogger logger)
+        public ServiceBusProducer(
+            AzureServiceBusConfiguration azureServiceBusConfiguration,
+            string queueOrTopicName,
+            ILogger logger,
+            IEnumerable<IProducerMiddleware>? middlewares = null)
         {
             var client = new ServiceBusClient(azureServiceBusConfiguration.ConnectionString);
             _sender = client.CreateSender(queueOrTopicName);
-
             _logger = logger;
+            _middlewares = middlewares;
         }
 
         public async Task ProduceAsync(TMessage message, CancellationToken cancellationToken)
@@ -32,12 +39,31 @@ public async Task ProduceAsync(TMessage message, CancellationToken cancellationT
                 }
             };
 
-            await _sender.SendMessageAsync(serviceBusMessage, cancellationToken);
+            async Task finalStep()
+            {
+                await _sender.SendMessageAsync(serviceBusMessage, cancellationToken);
+                _logger.LogInformation("Message {MessageType} published successfully!", message.GetType().Name);
+            }
 
-            _logger.LogInformation("Message {MessageType} published with successfully!", message.GetType().Name);
+            // Run middlewares, if it exist
+            if (_middlewares != null && _middlewares.Any())
+            {
+                Func<Task> next = finalStep;
+                foreach (var middleware in _middlewares.Reverse())
+                {
+                    var current = middleware;
+                    var prevNext = next;
+                    next = () => current.InvokeAsync(serviceBusMessage, prevNext);
+                }
+                await next();
+            }
+            else
+            {
+                await finalStep();
+            }
         }
 
-        public Task ProduceAsync(TMessage message, MessageOptions producerOptions, CancellationToken cancellationToken)
+        public async Task ProduceAsync(TMessage message, MessageOptions producerOptions, CancellationToken cancellationToken)
         {
             var json = JsonConvert.SerializeObject(message);
             var serviceBusMessage = new ServiceBusMessage(json)
@@ -52,18 +78,41 @@ public Task ProduceAsync(TMessage message, MessageOptions producerOptions, Cance
 
             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));
+                foreach (var kvp in from kvp in producerOptions.ApplicationProperties
+                                    where !serviceBusMessage.ApplicationProperties.ContainsKey(kvp.Key)
+                                    select 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);
+            // Mesmo pipeline de middlewares para as versões com opções
+            Func<Task> finalStep = async () =>
+            {
+                await _sender.SendMessageAsync(serviceBusMessage, cancellationToken);
+                _logger.LogInformation("Message {MessageType} published successfully!", message.GetType().Name);
+            };
+
+            if (_middlewares != null && _middlewares.Any())
+            {
+                Func<Task> next = finalStep;
+                foreach (var middleware in _middlewares.Reverse())
+                {
+                    var current = middleware;
+                    var prevNext = next;
+                    next = () => current.InvokeAsync(serviceBusMessage, prevNext);
+                }
+                await next();
+            }
+            else
+            {
+                await finalStep();
+            }
         }
 
         public Task ProduceAsync(TMessage message, TimeSpan delay, CancellationToken cancellationToken)