Subscriber middleware

EtherZa · EtherZa · commit 8f91ed8292a7 · 2025-08-21T16:15:51.000+08:00
diff --git a/.autover/changes/5bdfbea6-1f68-4d45-b380-4b705ed18fe2.json b/.autover/changes/5bdfbea6-1f68-4d45-b380-4b705ed18fe2.json
@@ -0,0 +1,11 @@
+{
+  "Projects": [
+    {
+      "Name": "AWS.Messaging",
+      "Type": "Minor",
+      "ChangelogMessages": [
+        "Added subscriber middleware"
+      ]
+    }
+  ]
+}
diff --git a/AWS.Messaging.lutconfig b/AWS.Messaging.lutconfig
@@ -0,0 +1,6 @@
+<LUTConfig Version="1.0">
+  <Repository />
+  <ParallelBuilds>true</ParallelBuilds>
+  <ParallelTestRuns>true</ParallelTestRuns>
+  <TestCaseTimeout>180000</TestCaseTimeout>
+</LUTConfig>
diff --git a/sampleapps/SubscriberService/Middleware/SampleMiddleware.cs b/sampleapps/SubscriberService/Middleware/SampleMiddleware.cs
@@ -0,0 +1,16 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+using AWS.Messaging;
+
+namespace SubscriberService.Middleware;
+
+public class SampleMiddleware : IMiddleware
+{
+    public Task<MessageProcessStatus> InvokeAsync<T>(MessageEnvelope<T> messageEnvelope, RequestDelegate next, CancellationToken token = default)
+    {
+        // This middleware does not do anything, but exists to demonstrate how to implement a middleware
+
+        return next();
+    }
+}
diff --git a/sampleapps/SubscriberService/Program.cs b/sampleapps/SubscriberService/Program.cs
@@ -11,6 +11,7 @@
 using OpenTelemetry.Resources;
 using OpenTelemetry.Trace;
 using SubscriberService.MessageHandlers;
+using SubscriberService.Middleware;
 using SubscriberService.Models;
 
 await Host.CreateDefaultBuilder(args)
@@ -34,6 +35,8 @@ await Host.CreateDefaultBuilder(args)
             if (string.IsNullOrEmpty(mpfQueueUrl))
                 throw new InvalidOperationException("Missing required configuration parameter 'AWS:Resources:MPFQueueUrl'.");
 
+            builder.AddMiddleware<SampleMiddleware>();
+
             builder.AddSQSPoller(mpfQueueUrl);
             builder.AddMessageHandler<ChatMessageHandler, ChatMessage>("chatMessage");
 
diff --git a/src/AWS.Messaging/Configuration/IMessageBusBuilder.cs b/src/AWS.Messaging/Configuration/IMessageBusBuilder.cs
@@ -53,6 +53,17 @@ public interface IMessageBusBuilder
     IMessageBusBuilder AddMessageHandler<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] THandler, TMessage>(string? messageTypeIdentifier = null)
         where THandler : IMessageHandler<TMessage>;
 
+    /// <summary>
+    /// Adds a middleware to the subscriber message bus pipeline.
+    /// </summary>
+    /// <remarks>
+    /// Middleware will be executed in the order in which it is added.
+    /// </remarks>
+    /// <typeparam name="TMiddleware">The type that implements <see cref="IMiddleware"/></typeparam>
+    /// <param name="serviceLifetime">The lifetime of the middleware.</param>
+    IMessageBusBuilder AddMiddleware<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] TMiddleware>(ServiceLifetime serviceLifetime = ServiceLifetime.Singleton)
+        where TMiddleware : class, IMiddleware;
+
     /// <summary>
     /// Adds an SQS queue to poll for messages.
     /// </summary>
diff --git a/src/AWS.Messaging/Configuration/IMessageConfiguration.cs b/src/AWS.Messaging/Configuration/IMessageConfiguration.cs
@@ -46,6 +46,11 @@ public interface IMessageConfiguration
     /// <returns>The <see cref="SubscriberMapping"/> containing routing info for the specified message type.</returns>
     SubscriberMapping? GetSubscriberMapping(string messageTypeIdentifier);
 
+    /// <summary>
+    /// Maps the middleware types to be used in order of execution.
+    /// </summary>
+    IList<SubscriberMiddleware> SubscriberMiddleware { get; }
+
     /// <summary>
     /// List of configurations for subscriber to poll for messages from an AWS service endpoint.
     /// </summary>
diff --git a/src/AWS.Messaging/Configuration/MessageBusBuilder.cs b/src/AWS.Messaging/Configuration/MessageBusBuilder.cs
@@ -99,16 +99,26 @@ private IMessageBusBuilder AddPublisher([DynamicallyAccessedMembers(DynamicallyA
     public IMessageBusBuilder AddMessageHandler<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] THandler, TMessage>(string? messageTypeIdentifier = null)
         where THandler : IMessageHandler<TMessage>
     {
-        return AddMessageHandler(typeof(THandler), typeof(TMessage), () => new MessageEnvelope<TMessage>(), messageTypeIdentifier);
+        var subscriberMapping = SubscriberMapping.Create<THandler, TMessage>(messageTypeIdentifier);
+        _messageConfiguration.SubscriberMappings.Add(subscriberMapping);
+        return this;
     }
 
-    private IMessageBusBuilder AddMessageHandler([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] Type handlerType, Type messageType, Func<MessageEnvelope> envelopeFactory, string? messageTypeIdentifier = null)
+    private IMessageBusBuilder AddMessageHandler([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] Type handlerType, Type messageType, Func<MessageEnvelope> envelopeFactory, MethodInfo middlewareInvokeAsyncMethodInfo, string? messageTypeIdentifier = null)
     {
-        var subscriberMapping = new SubscriberMapping(handlerType, messageType, envelopeFactory, messageTypeIdentifier);
+        var subscriberMapping = new SubscriberMapping(handlerType, messageType, envelopeFactory, middlewareInvokeAsyncMethodInfo, messageTypeIdentifier);
         _messageConfiguration.SubscriberMappings.Add(subscriberMapping);
         return this;
     }
 
+    public IMessageBusBuilder AddMiddleware<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] TMiddleware>(ServiceLifetime serviceLifetime = ServiceLifetime.Singleton)
+        where TMiddleware : class, IMiddleware
+    {
+        var subscriberMiddleware = SubscriberMiddleware.Create<TMiddleware>(serviceLifetime);
+        _messageConfiguration.SubscriberMiddleware.Add(subscriberMiddleware);
+        return this;
+    }
+
     /// <inheritdoc/>
     public IMessageBusBuilder AddSQSPoller(string queueUrl, Action<SQSMessagePollerOptions>? options = null)
     {
@@ -216,6 +226,11 @@ public IMessageBusBuilder LoadConfigurationFromSettings(IConfiguration configura
 
         if (settings.MessageHandlers != null)
         {
+            // This is not Native AOT compatible but the method in general is marked
+            // as not being Native AOT compatible due to loading dynamic types. So this
+            // not being Native AOT compatible is okay.
+            var middlewareInvokeMethod = typeof(IMiddleware).GetMethod(nameof(IMiddleware.InvokeAsync))!;
+
             foreach (var messageHandler in settings.MessageHandlers)
             {
                 var messageType = GetTypeFromAssemblies(callingAssembly, messageHandler.MessageType)
@@ -238,7 +253,12 @@ MessageEnvelope envelopeFactory()
                     return (MessageEnvelope)envelope;
                 }
 
-                AddMessageHandler(handlerType, messageType, envelopeFactory, messageHandler.MessageTypeIdentifier);
+                // MakeGenericMethod is not Native AOT compatible but the method in general is marked
+                // as not being Native AOT compatible due to loading dynamic types. So this
+                // method not being Native AOT compatible is okay.
+                var middlewareInvokeAsyncMethodInfo = middlewareInvokeMethod.MakeGenericMethod(messageType);
+
+                AddMessageHandler(handlerType, messageType, envelopeFactory, middlewareInvokeAsyncMethodInfo, messageHandler.MessageTypeIdentifier);
             }
         }
 
@@ -378,6 +398,11 @@ internal void Build()
             {
                 _serviceCollection.TryAddScoped(subscriberMapping.HandlerType);
             }
+
+            foreach (var subscriberMiddleware in _messageConfiguration.SubscriberMiddleware)
+            {
+                _serviceCollection.TryAdd(new ServiceDescriptor(subscriberMiddleware.Type, subscriberMiddleware.Type, subscriberMiddleware.ServiceLifetime));
+            }
         }
 
         if (_messageConfiguration.MessagePollerConfigurations.Any())
diff --git a/src/AWS.Messaging/Configuration/MessageConfiguration.cs b/src/AWS.Messaging/Configuration/MessageConfiguration.cs
@@ -39,6 +39,9 @@ public class MessageConfiguration : IMessageConfiguration
         return subscriberMapping;
     }
 
+    /// <inheritdoc/>
+    public IList<SubscriberMiddleware> SubscriberMiddleware { get; } = new List<SubscriberMiddleware>();
+
     /// <inheritdoc/>
     public IList<IMessagePollerConfiguration> MessagePollerConfigurations { get; set; } = new List<IMessagePollerConfiguration>();
 
diff --git a/src/AWS.Messaging/Configuration/SubscriberMapping.cs b/src/AWS.Messaging/Configuration/SubscriberMapping.cs
@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 using System.Diagnostics.CodeAnalysis;
+using System.Reflection;
 
 namespace AWS.Messaging.Configuration;
 
@@ -17,6 +18,9 @@ public class SubscriberMapping
     /// <inheritdoc/>
     public Type MessageType { get; }
 
+    /// <inheritdoc/>
+    public MethodInfo MiddlewareInvokeAsyncMethodInfo { get; }
+
     /// <inheritdoc/>
     public string MessageTypeIdentifier { get; }
 
@@ -32,9 +36,10 @@ public class SubscriberMapping
     /// <param name="handlerType">The type that implements <see cref="IMessageHandler{T}"/></param>
     /// <param name="messageType">The type that will be message data will deserialized into</param>
     /// <param name="envelopeFactory">Func for creating <see cref="MessageEnvelope{messageType}"/></param>
+    /// <param name="middlewareInvokeAsyncMethodInfo"><see cref="MethodInfo"/> to invoke middleware of <see cref="MessageType"/>.</param>
     /// <param name="messageTypeIdentifier">Optional message type identifier. If not set the full name of the <see cref="MessageType"/> is used.</param>
 
-    internal SubscriberMapping([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] Type handlerType, Type messageType, Func<MessageEnvelope> envelopeFactory, string? messageTypeIdentifier = null)
+    internal SubscriberMapping([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] Type handlerType, Type messageType, Func<MessageEnvelope> envelopeFactory, MethodInfo middlewareInvokeAsyncMethodInfo, string? messageTypeIdentifier = null)
     {
         HandlerType = handlerType;
         MessageType = messageType;
@@ -44,6 +49,7 @@ internal SubscriberMapping([DynamicallyAccessedMembers(DynamicallyAccessedMember
             messageType.FullName ?? throw new InvalidMessageTypeException("Unable to retrieve the Full Name of the provided Message Type.");
 
         MessageEnvelopeFactory = envelopeFactory;
+        MiddlewareInvokeAsyncMethodInfo = middlewareInvokeAsyncMethodInfo;
     }
 
     /// <summary>
@@ -61,6 +67,19 @@ static MessageEnvelope<TMessage> envelopeFactory()
             return new MessageEnvelope<TMessage>();
         }
 
-        return new SubscriberMapping(typeof(THandler), typeof(TMessage), envelopeFactory, messageTypeIdentifier);
+        return new SubscriberMapping(typeof(THandler), typeof(TMessage), envelopeFactory, MiddlewareMethodCache<TMessage>.InvokeAsyncMethod, messageTypeIdentifier);
+    }
+
+    private static class MiddlewareMethodCache<T>
+    {
+        public static readonly MethodInfo InvokeAsyncMethod = MiddlewareMethod.OpenGenericInvokeAsyncMethod.MakeGenericMethod(typeof(T));
+    }
+
+    private static class MiddlewareMethod
+    {
+        // resolved once, globally
+        public static readonly MethodInfo OpenGenericInvokeAsyncMethod = typeof(IMiddleware)
+            .GetMethods()
+            .First(m => m.Name == nameof(IMiddleware.InvokeAsync) && m.IsGenericMethod);
     }
 }
diff --git a/src/AWS.Messaging/Configuration/SubscriberMiddleware.cs b/src/AWS.Messaging/Configuration/SubscriberMiddleware.cs
@@ -0,0 +1,47 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+using System.Diagnostics.CodeAnalysis;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace AWS.Messaging.Configuration;
+
+/// <summary>
+/// Tracks the <see cref="IMiddleware"/> to be processed by the <see cref="Services.IHandlerInvoker"/> implementation and its <see cref="ServiceLifetime"/>.
+/// </summary>
+public class SubscriberMiddleware
+{
+    /// <summary>
+    /// Constructs an instance of <see cref="SubscriberMiddleware"/>
+    /// </summary>
+    /// <param name="type">The type that implements <see cref="IMiddleware"/>.</param>
+    /// <param name="serviceLifetime">The lifetime of the middleware.</param>
+    internal SubscriberMiddleware([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] Type type, ServiceLifetime serviceLifetime)
+    {
+        Type = type;
+        ServiceLifetime = serviceLifetime;
+    }
+
+    /// <summary>
+    /// Type that implements <see cref="IMiddleware"/>.
+    /// </summary>
+    [DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)]
+    public Type Type { get; }
+
+    /// <summary>
+    /// Service lifetime of the middleware.
+    /// </summary>
+    public ServiceLifetime ServiceLifetime { get; }
+
+    /// <summary>
+    /// Creates a SubscriberMiddleware from the generic parameters for the middleware.
+    /// </summary>
+    /// <typeparam name="TMiddleware">The type that implements <see cref="IMiddleware"/></typeparam>
+    /// <param name="serviceLifetime">The lifetime of the middleware.</param>
+    /// <returns><see cref="SubscriberMapping"/></returns>
+    public static SubscriberMiddleware Create<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] TMiddleware>(ServiceLifetime serviceLifetime = ServiceLifetime.Singleton)
+        where TMiddleware : class, IMiddleware
+    {
+        return new SubscriberMiddleware(typeof(TMiddleware), serviceLifetime);
+    }
+}
diff --git a/src/AWS.Messaging/IMiddleware.cs b/src/AWS.Messaging/IMiddleware.cs
@@ -0,0 +1,30 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+namespace AWS.Messaging;
+
+/// <summary>
+/// This interface is implemented by the users of this library for each layer of middleware that should be processed.
+/// </summary>
+public interface IMiddleware
+{
+    /// <summary>
+    /// Processes a message through the middleware pipeline, invoking the next middleware or the message handler.
+    /// </summary>
+    /// <param name="messageEnvelope">The message read from the message source wrapped around a message envelope containing message metadata.</param>
+    /// <param name="next">Delegate to execute the next layer of middleware. When no further middleware remains, the delegate will execute the message handler.</param>
+    /// <param name="token">The optional cancellation token.</param>
+    /// <returns>
+    /// The status of the processed message. For example whether the message was successfully processed.
+    /// Default implementations should return the result returned from the next delegate.
+    /// </returns>
+    Task<MessageProcessStatus> InvokeAsync<T>(MessageEnvelope<T> messageEnvelope, RequestDelegate next, CancellationToken token = default);
+}
+
+/// <summary>
+/// The delegate used to invoke the next middleware layer or the message handler.
+/// </summary>
+/// <returns>
+/// The status of the processed message. For example whether the message was successfully processed.
+/// </returns>
+public delegate Task<MessageProcessStatus> RequestDelegate();
diff --git a/src/AWS.Messaging/Services/HandlerInvoker.cs b/src/AWS.Messaging/Services/HandlerInvoker.cs
@@ -16,6 +16,7 @@ public class HandlerInvoker : IHandlerInvoker
     private readonly IServiceProvider _serviceProvider;
     private readonly ILogger<HandlerInvoker> _logger;
     private readonly ITelemetryFactory _telemetryFactory;
+    private readonly IMessageConfiguration _messageConfiguration;
 
     /// <summary>
     /// Caches the <see cref="MethodInfo"/> of the <see cref="IMessageHandler{T}.HandleAsync(MessageEnvelope{T}, CancellationToken)"/>
@@ -29,14 +30,17 @@ public class HandlerInvoker : IHandlerInvoker
     /// <param name="serviceProvider">Service provider used to resolve handler objects</param>
     /// <param name="logger">Logger for debugging information</param>
     /// <param name="telemetryFactory">Factory for telemetry data</param>
+    /// <param name="messageConfiguration">Messaging configuration holding middleware configuration</param>
     public HandlerInvoker(
         IServiceProvider serviceProvider,
         ILogger<HandlerInvoker> logger,
-        ITelemetryFactory telemetryFactory)
+        ITelemetryFactory telemetryFactory,
+        IMessageConfiguration messageConfiguration)
     {
         _serviceProvider = serviceProvider;
         _logger = logger;
         _telemetryFactory = telemetryFactory;
+        _messageConfiguration = messageConfiguration;
     }
 
     /// <inheritdoc/>
@@ -83,15 +87,8 @@ public async Task<MessageProcessStatus> InvokeAsync(MessageEnvelope messageEnvel
 
                     try
                     {
-                        var task = method.Invoke(handler, new object[] { messageEnvelope, token }) as Task<MessageProcessStatus>;
-
-                        if (task == null)
-                        {
-                            _logger.LogError("Unexpected return type for the HandleAsync method on {HandlerType} while handling message ID {MessageEnvelopeId}. Expected {ExpectedType}", subscriberMapping.HandlerType, messageEnvelope.Id, nameof(Task<MessageProcessStatus>));
-                            throw new InvalidMessageHandlerSignatureException($"Unexpected return type for the HandleAsync method on {subscriberMapping.HandlerType} while handling message ID {messageEnvelope.Id}. Expected {nameof(Task<MessageProcessStatus>)}");
-                        }
-
-                        return await task;
+                        var middlewares = _messageConfiguration.SubscriberMiddleware.Select(subscriberMiddleware => (IMiddleware)scope.ServiceProvider.GetRequiredService(subscriberMiddleware.Type)!).ToList();
+                        return await ExecutePipelineAsync(messageEnvelope, subscriberMapping, middlewares, handler, method, subscriberMapping.MiddlewareInvokeAsyncMethodInfo, token).ConfigureAwait(false);
                     }
                     // Since we are invoking HandleAsync via reflection, we need to unwrap the TargetInvocationException
                     // containing application exceptions that happened inside the IMessageHandler
@@ -126,4 +123,44 @@ public async Task<MessageProcessStatus> InvokeAsync(MessageEnvelope messageEnvel
             }
         }
     }
+
+    private async Task<MessageProcessStatus> ExecutePipelineAsync(
+        MessageEnvelope messageEnvelope,
+        SubscriberMapping subscriberMapping,
+        IReadOnlyList<IMiddleware> middlewares,
+        object handler,
+        MethodInfo handlerMethod,
+        MethodInfo middlewareMethod,
+        CancellationToken token)
+    {
+        RequestDelegate next = () =>
+        {
+            if (handlerMethod.Invoke(handler, new object[] { messageEnvelope, token }) is not Task<MessageProcessStatus> task)
+            {
+                _logger.LogError("Unexpected return type for the HandleAsync method on {HandlerType} while handling message ID {MessageEnvelopeId}. Expected {ExpectedType}", subscriberMapping.HandlerType, messageEnvelope.Id, nameof(Task<MessageProcessStatus>));
+                throw new InvalidMessageHandlerSignatureException($"Unexpected return type for the HandleAsync method on {subscriberMapping.HandlerType} while handling message ID {messageEnvelope.Id}. Expected {nameof(Task<MessageProcessStatus>)}");
+            }
+
+            return task;
+        };
+
+        for (var i = middlewares.Count - 1; i >= 0; i--)
+        {
+            var capturedNext = next;
+            var middleware = middlewares[i];
+
+            next = () =>
+            {
+                if (middlewareMethod.Invoke(middleware, new object[] { messageEnvelope, capturedNext, token }) is not Task<MessageProcessStatus> task)
+                {
+                    _logger.LogError("Unexpected return type for the InvokeAsync method on {MiddlewareType} while handling message ID {MessageEnvelopeId}. Expected {ExpectedType}", middleware.GetType(), messageEnvelope.Id, nameof(Task<MessageProcessStatus>));
+                    throw new InvalidMessageHandlerSignatureException($"Unexpected return type for the InvokeAsync method on {middleware.GetType()} while handling message ID {messageEnvelope.Id}. Expected {nameof(Task<MessageProcessStatus>)}");
+                }
+
+                return task;
+            };
+        }
+
+        return await next().ConfigureAwait(false);
+    }
 }
diff --git a/test/AWS.Messaging.UnitTests/HandlerInvokerTests.cs b/test/AWS.Messaging.UnitTests/HandlerInvokerTests.cs
diff --git a/test/AWS.Messaging.UnitTests/Models/SubscriberMiddlewareModels.cs b/test/AWS.Messaging.UnitTests/Models/SubscriberMiddlewareModels.cs
diff --git a/test/AWS.Messaging.UnitTests/OpenTelemetryTests.cs b/test/AWS.Messaging.UnitTests/OpenTelemetryTests.cs

Original file line number	Diff line number	Diff line change
`@@ -99,16 +99,26 @@ private IMessageBusBuilder AddPublisher([DynamicallyAccessedMembers(DynamicallyA`
`99`	`99`	`public IMessageBusBuilder AddMessageHandler<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] THandler, TMessage>(string? messageTypeIdentifier = null)`
`100`	`100`	`where THandler : IMessageHandler<TMessage>`
`101`	`101`	`{`
`102`		`- return AddMessageHandler(typeof(THandler), typeof(TMessage), () => new MessageEnvelope<TMessage>(), messageTypeIdentifier);`
	`102`	`+ var subscriberMapping = SubscriberMapping.Create<THandler, TMessage>(messageTypeIdentifier);`
	`103`	`+ _messageConfiguration.SubscriberMappings.Add(subscriberMapping);`
	`104`	`+ return this;`
`103`	`105`	`}`
`104`	`106`
`105`		`- private IMessageBusBuilder AddMessageHandler([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] Type handlerType, Type messageType, Func<MessageEnvelope> envelopeFactory, string? messageTypeIdentifier = null)`
	`107`	`+ private IMessageBusBuilder AddMessageHandler([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] Type handlerType, Type messageType, Func<MessageEnvelope> envelopeFactory, MethodInfo middlewareInvokeAsyncMethodInfo, string? messageTypeIdentifier = null)`
`106`	`108`	`{`
`107`		`- var subscriberMapping = new SubscriberMapping(handlerType, messageType, envelopeFactory, messageTypeIdentifier);`
	`109`	`+ var subscriberMapping = new SubscriberMapping(handlerType, messageType, envelopeFactory, middlewareInvokeAsyncMethodInfo, messageTypeIdentifier);`
`108`	`110`	`_messageConfiguration.SubscriberMappings.Add(subscriberMapping);`
`109`	`111`	`return this;`
`110`	`112`	`}`
`111`	`113`
	`114`	`+ public IMessageBusBuilder AddMiddleware<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] TMiddleware>(ServiceLifetime serviceLifetime = ServiceLifetime.Singleton)`
	`115`	`+ where TMiddleware : class, IMiddleware`
	`116`	`+ {`
	`117`	`+ var subscriberMiddleware = SubscriberMiddleware.Create<TMiddleware>(serviceLifetime);`
	`118`	`+ _messageConfiguration.SubscriberMiddleware.Add(subscriberMiddleware);`
	`119`	`+ return this;`
	`120`	`+ }`
	`121`	`+`
`112`	`122`	`/// <inheritdoc/>`
`113`	`123`	`public IMessageBusBuilder AddSQSPoller(string queueUrl, Action<SQSMessagePollerOptions>? options = null)`
`114`	`124`	`{`
`@@ -216,6 +226,11 @@ public IMessageBusBuilder LoadConfigurationFromSettings(IConfiguration configura`
`216`	`226`
`217`	`227`	`if (settings.MessageHandlers != null)`
`218`	`228`	`{`
	`229`	`+ // This is not Native AOT compatible but the method in general is marked`
	`230`	`+ // as not being Native AOT compatible due to loading dynamic types. So this`
	`231`	`+ // not being Native AOT compatible is okay.`
	`232`	`+ var middlewareInvokeMethod = typeof(IMiddleware).GetMethod(nameof(IMiddleware.InvokeAsync))!;`
	`233`	`+`
`219`	`234`	`foreach (var messageHandler in settings.MessageHandlers)`
`220`	`235`	`{`
`221`	`236`	`var messageType = GetTypeFromAssemblies(callingAssembly, messageHandler.MessageType)`
`@@ -238,7 +253,12 @@ MessageEnvelope envelopeFactory()`
`238`	`253`	`return (MessageEnvelope)envelope;`
`239`	`254`	`}`
`240`	`255`
`241`		`- AddMessageHandler(handlerType, messageType, envelopeFactory, messageHandler.MessageTypeIdentifier);`
	`256`	`+ // MakeGenericMethod is not Native AOT compatible but the method in general is marked`
	`257`	`+ // as not being Native AOT compatible due to loading dynamic types. So this`
	`258`	`+ // method not being Native AOT compatible is okay.`
	`259`	`+ var middlewareInvokeAsyncMethodInfo = middlewareInvokeMethod.MakeGenericMethod(messageType);`
	`260`	`+`
	`261`	`+ AddMessageHandler(handlerType, messageType, envelopeFactory, middlewareInvokeAsyncMethodInfo, messageHandler.MessageTypeIdentifier);`
`242`	`262`	`}`
`243`	`263`	`}`
`244`	`264`
`@@ -378,6 +398,11 @@ internal void Build()`
`378`	`398`	`{`
`379`	`399`	`_serviceCollection.TryAddScoped(subscriberMapping.HandlerType);`
`380`	`400`	`}`
	`401`	`+`
	`402`	`+ foreach (var subscriberMiddleware in _messageConfiguration.SubscriberMiddleware)`
	`403`	`+ {`
	`404`	`+ _serviceCollection.TryAdd(new ServiceDescriptor(subscriberMiddleware.Type, subscriberMiddleware.Type, subscriberMiddleware.ServiceLifetime));`
	`405`	`+ }`
`381`	`406`	`}`
`382`	`407`
`383`	`408`	`if (_messageConfiguration.MessagePollerConfigurations.Any())`
Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,9 @@ public class MessageConfiguration : IMessageConfiguration`
`39`	`39`	`return subscriberMapping;`
`40`	`40`	`}`
`41`	`41`
	`42`	`+ /// <inheritdoc/>`
	`43`	`+ public IList<SubscriberMiddleware> SubscriberMiddleware { get; } = new List<SubscriberMiddleware>();`
	`44`	`+`
`42`	`45`	`/// <inheritdoc/>`
`43`	`46`	`public IList<IMessagePollerConfiguration> MessagePollerConfigurations { get; set; } = new List<IMessagePollerConfiguration>();`
`44`	`47`