Message error handler

Author: EtherZa

.autover/changes/5bdfbea6-1f68-4d45-b380-4b705ed18fe2.json

@@ -4,7 +4,7 @@
       "Name": "AWS.Messaging",
       "Type": "Minor",
       "ChangelogMessages": [
-        "Added subscriber middleware"
+        "Added subscriber middleware with optional error handler to override result or retry execution."
       ]
     }
   ]

src/AWS.Messaging/Configuration/MessageBusBuilder.cs

@@ -2,6 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 using System.Collections.Concurrent;
+using System.Diagnostics.CodeAnalysis;
+using System.Linq.Expressions;
+using System.Reflection;
 using AWS.Messaging.Configuration.Internal;
 using AWS.Messaging.Publishers;
 using AWS.Messaging.Publishers.EventBridge;
@@ -12,13 +15,11 @@
 using AWS.Messaging.Services.Backoff;
 using AWS.Messaging.Services.Backoff.Policies;
 using AWS.Messaging.Telemetry;
+using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
-using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
-using System.Diagnostics.CodeAnalysis;
-using System.Reflection;
 
 namespace AWS.Messaging.Configuration;
 
@@ -104,13 +105,20 @@ private IMessageBusBuilder AddPublisher([DynamicallyAccessedMembers(DynamicallyA
         return this;
     }
 
-    private IMessageBusBuilder AddMessageHandler([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] Type handlerType, Type messageType, Func<MessageEnvelope> envelopeFactory, MethodInfo middlewareInvokeAsyncMethodInfo, string? messageTypeIdentifier = null)
+    private IMessageBusBuilder AddMessageHandler([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] Type handlerType, Type messageType, Func<MessageEnvelope> envelopeFactory, HandlerInvokerDelegate handlerInvokerDelegate, string? messageTypeIdentifier = null)
     {
-        var subscriberMapping = new SubscriberMapping(handlerType, messageType, envelopeFactory, middlewareInvokeAsyncMethodInfo, messageTypeIdentifier);
+        var subscriberMapping = new SubscriberMapping(handlerType, messageType, envelopeFactory, handlerInvokerDelegate, messageTypeIdentifier);
         _messageConfiguration.SubscriberMappings.Add(subscriberMapping);
         return this;
     }
 
+    public IMessageBusBuilder AddMessageErrorHandler<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] T>(ServiceLifetime serviceLifetime = ServiceLifetime.Singleton)
+        where T: IMessageErrorHandler
+    {
+        AddAdditionalService(new ServiceDescriptor(typeof(IMessageErrorHandler), typeof(T), serviceLifetime));
+        return this;
+    }
+
     public IMessageBusBuilder AddMiddleware<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] TMiddleware>(ServiceLifetime serviceLifetime = ServiceLifetime.Singleton)
         where TMiddleware : class, IMiddleware
     {
@@ -226,24 +234,20 @@ 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)
                     ?? throw new InvalidAppSettingsConfigurationException($"Unable to find the provided message type '{messageHandler.MessageType}'.");
                 var handlerType = GetTypeFromAssemblies(callingAssembly, messageHandler.HandlerType)
                     ?? throw new InvalidAppSettingsConfigurationException($"Unable to find the provided message handler type '{messageHandler.HandlerType}'.");
 
+                var messageEnvelopeType = typeof(MessageEnvelope<>).MakeGenericType(messageType);
+
                 // This func 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
                 // func not being Native AOT compatible is okay.
                 MessageEnvelope envelopeFactory()
                 {
-                    var messageEnvelopeType = typeof(MessageEnvelope<>).MakeGenericType(messageType);
                     var envelope = Activator.CreateInstance(messageEnvelopeType);
                     if (envelope == null || envelope is not MessageEnvelope)
                     {
@@ -253,12 +257,8 @@ MessageEnvelope envelopeFactory()
                     return (MessageEnvelope)envelope;
                 }
 
-                // 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);
+                var handlerInoker = BuildHandlerInvoker(messageType, messageEnvelopeType);
+                AddMessageHandler(handlerType, messageType, envelopeFactory, handlerInoker, messageHandler.MessageTypeIdentifier);
             }
         }
 
@@ -303,6 +303,29 @@ MessageEnvelope envelopeFactory()
         }
 
         return this;
+
+        // 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
+        // func not being Native AOT compatible is okay.
+        static HandlerInvokerDelegate BuildHandlerInvoker(Type messageType, Type messageEnvelopeType)
+        {
+            var invokerParam = Expression.Parameter(typeof(HandlerInvoker), "invoker");
+            var envelopeParam = Expression.Parameter(typeof(MessageEnvelope), "envelope");
+            var mappingParam = Expression.Parameter(typeof(SubscriberMapping), "mapping");
+            var tokenParam = Expression.Parameter(typeof(CancellationToken), "token");
+
+            // invoker.InvokeAsync<T>( (MessageEnvelope<T>) envelope, mapping, token )
+            var genericMethodDef = typeof(HandlerInvoker)
+                .GetMethods(BindingFlags.Public | BindingFlags.Instance)
+                .First(m => m.Name == nameof(HandlerInvoker.InvokeAsync) && m.IsGenericMethodDefinition)
+                .GetGenericMethodDefinition();
+
+            var closedMethod = genericMethodDef.MakeGenericMethod(messageType);
+            var typedEnvelope = Expression.Convert(envelopeParam, messageEnvelopeType);
+            var call = Expression.Call(invokerParam, closedMethod, typedEnvelope, mappingParam, tokenParam);
+            var lambda = Expression.Lambda<HandlerInvokerDelegate>(call, invokerParam, envelopeParam, mappingParam, tokenParam);
+            return lambda.Compile();
+        }
     }
 
     [RequiresUnreferencedCode("This method requires loading types dynamically as defined in the configuration system.")]

src/AWS.Messaging/Configuration/SubscriberMapping.cs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 using System.Diagnostics.CodeAnalysis;
-using System.Reflection;
+using AWS.Messaging.Services;
 
 namespace AWS.Messaging.Configuration;
 
@@ -19,7 +19,7 @@ public class SubscriberMapping
     public Type MessageType { get; }
 
     /// <inheritdoc/>
-    public MethodInfo MiddlewareInvokeAsyncMethodInfo { get; }
+    public HandlerInvokerDelegate HandlerInvoker { get; }
 
     /// <inheritdoc/>
     public string MessageTypeIdentifier { get; }
@@ -36,10 +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="handlerInvoker">Delegate to invoke handler 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, MethodInfo middlewareInvokeAsyncMethodInfo, string? messageTypeIdentifier = null)
+    internal SubscriberMapping([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.All)] Type handlerType, Type messageType, Func<MessageEnvelope> envelopeFactory, HandlerInvokerDelegate handlerInvoker, string? messageTypeIdentifier = null)
     {
         HandlerType = handlerType;
         MessageType = messageType;
@@ -49,7 +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;
+        HandlerInvoker = handlerInvoker;
     }
 
     /// <summary>
@@ -67,19 +67,13 @@ static MessageEnvelope<TMessage> envelopeFactory()
             return new MessageEnvelope<TMessage>();
         }
 
-        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));
-    }
+        static Task<MessageProcessStatus> handlerInvoker(HandlerInvoker invoker, MessageEnvelope messageEnvelope, SubscriberMapping subscriberMapping, CancellationToken token = default)
+        {
+            return invoker.InvokeAsync((MessageEnvelope<TMessage>)messageEnvelope, subscriberMapping, token);
+        }
 
-    private static class MiddlewareMethod
-    {
-        // resolved once, globally
-        public static readonly MethodInfo OpenGenericInvokeAsyncMethod = typeof(IMiddleware)
-            .GetMethods()
-            .First(m => m.Name == nameof(IMiddleware.InvokeAsync) && m.IsGenericMethod);
+        return new SubscriberMapping(typeof(THandler), typeof(TMessage), envelopeFactory, handlerInvoker, messageTypeIdentifier);
     }
 }
+
+public delegate Task<MessageProcessStatus> HandlerInvokerDelegate(HandlerInvoker invoker, MessageEnvelope messageEnvelope, SubscriberMapping subscriberMapping, CancellationToken token = default);

src/AWS.Messaging/IMessageErrorHandler.cs

@@ -0,0 +1,35 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+namespace AWS.Messaging;
+
+public interface IMessageErrorHandler
+{
+    /// <summary>
+    /// Handles errors that occur during message processing.
+    /// </summary>
+    /// <param name="messageEnvelope">The message being processed.</param>
+    /// <param name="exception"><see cref="Exception"/> raised while processing message.</param>
+    /// <param name="attempts">Number of attempts made at processing this message</param>
+    /// <param name="token"><see cref="CancellationToken"/></param>
+    /// <returns><see cref="MessageErrorHandlerResponse"/></returns>
+    public ValueTask<MessageErrorHandlerResponse> OnHandleError<T>(MessageEnvelope<T> messageEnvelope, Exception exception, int attempts, CancellationToken token);
+}
+
+public enum MessageErrorHandlerResponse
+{
+    /// <summary>
+    /// Failed response.
+    /// </summary>
+    Failed,
+
+    /// <summary>
+    /// Retry the message processing in the same process.
+    /// </summary>
+    Retry,
+
+    /// <summary>
+    /// Success response.
+    /// </summary>
+    Success
+}