FEAT: Adding new configuration to use consumer middlewares follow scope DI

Author: Felipe Mattioli

src/AzureServiceBusFlow/Builders/ServiceBusConfigurationBuilder.cs

@@ -51,7 +51,7 @@ public ServiceBusConfigurationBuilder UseGlobalConsumerMiddleware<TMiddleware>()
                 s.ServiceType == typeof(IConsumerMiddleware) &&
                 s.ImplementationType == typeof(TMiddleware)))
             {
-                _services.AddSingleton<IConsumerMiddleware, TMiddleware>();
+                _services.AddScoped<IConsumerMiddleware, TMiddleware>();
             }
 
             return this;

src/AzureServiceBusFlow/Builders/ServiceBusConsumerConfigurationBuilder.cs

@@ -96,22 +96,18 @@ public void Build()
                                            s.ImplementationType == middlewareType)
                                            select middlewareType)
             {
-                _services.AddKeyedSingleton(typeof(IConsumerMiddleware), _consumerMiddlewareKey,  middlewareType);
+                _services.AddKeyedScoped(typeof(IConsumerMiddleware), _consumerMiddlewareKey,  middlewareType);
             }
 
             _services.AddSingleton<IHostedService>(sp =>
             {
                 var logger = sp.GetRequiredService<ILogger<ServiceBusConsumerHostedService>>();
-                var localConsumerMiddlewares = sp.GetKeyedServices<IConsumerMiddleware>(_consumerMiddlewareKey) ?? [];
-                var globalConsumerMiddlewares = sp.GetServices<IConsumerMiddleware>() ?? [];
-
-                var consumerMiddlewares = globalConsumerMiddlewares.Union(localConsumerMiddlewares);
 
                 if (!string.IsNullOrWhiteSpace(_queueName))
                 {
                     return new ServiceBusConsumerHostedService(
                         (rawMessage, rootProvider, cancellationToken) =>
-                            MessageConsumingHandler(rawMessage, rootProvider, consumerMiddlewares, logger, cancellationToken),
+                            MessageConsumingHandler(rawMessage, rootProvider, _consumerMiddlewareKey, logger, cancellationToken),
                         sp,
                         logger,
                         _azureServiceBusConfiguration,
@@ -120,7 +116,7 @@ public void Build()
 
                 return new ServiceBusConsumerHostedService(
                     (rawMessage, rootProvider, cancellationToken) =>
-                        MessageConsumingHandler(rawMessage, rootProvider, consumerMiddlewares, logger, cancellationToken),
+                        MessageConsumingHandler(rawMessage, rootProvider, _consumerMiddlewareKey, logger, cancellationToken),
                     sp,
                     logger,
                     _azureServiceBusConfiguration,
@@ -129,23 +125,38 @@ public void Build()
             });
         }
 
-        private async Task MessageConsumingHandler(ServiceBusReceivedMessage rawMessage, IServiceProvider rootProvider, IEnumerable<IConsumerMiddleware> middlewares, ILogger<ServiceBusConsumerHostedService> logger, CancellationToken cancellationToken)
+        private async Task MessageConsumingHandler(
+            ServiceBusReceivedMessage rawMessage,
+            IServiceProvider rootProvider,
+            object consumerMiddlewareKey,
+            ILogger<ServiceBusConsumerHostedService> logger,
+            CancellationToken cancellationToken)
         {
+            using var scope = rootProvider.CreateScope();
+            var sp = scope.ServiceProvider;
+
+            var global = sp.GetServices<IConsumerMiddleware>() ?? [];
+            var local = sp.GetKeyedServices<IConsumerMiddleware>(consumerMiddlewareKey) ?? [];
+
+            var middlewares = global
+                .Concat(local)
+                .GroupBy(m => m.GetType())
+                .Select(g => g.First())
+                .ToList();
+
             Func<Task> finalStep = async () =>
             {
-                await ProcessHandlersAsync(rawMessage, rootProvider, logger, cancellationToken);
+                await ProcessHandlersAsync(rawMessage, sp, logger, cancellationToken);
             };
 
-            if (middlewares != null && middlewares.Any())
+            if (middlewares.Count > 0)
             {
                 Func<Task> next = finalStep;
 
-                foreach (var middleware in middlewares.Reverse())
+                foreach (var middleware in middlewares.AsEnumerable().Reverse())
                 {
-                    var current = middleware;
                     var prevNext = next;
-
-                    next = () => current.InvokeAsync(rawMessage, prevNext, cancellationToken);
+                    next = () => middleware.InvokeAsync(rawMessage, prevNext, cancellationToken);
                 }
 
                 await next();