Implement Middleware for Handlers

samples/src/JustSaying.Sample.Restaurant.OrderingApi/BusService.cs

@@ -27,8 +27,8 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         {
             _logger.LogInformation("Ordering API subscriber running");
 
-            await _bus.StartAsync(stoppingToken);
             await _publisher.StartAsync(stoppingToken);
+            await _bus.StartAsync(stoppingToken);
         }
     }
 }

src/JustSaying.Extensions.DependencyInjection.Microsoft/ServiceProviderResolver.cs

@@ -80,7 +80,6 @@ public IHandlerAsync<T> ResolveHandler<T>(HandlerResolutionContext context)
         }
 
         /// <inheritdoc />
-        public T ResolveService<T>()
-            => ServiceProvider.GetRequiredService<T>();
+        public T ResolveService<T>() => ServiceProvider.GetRequiredService<T>();
     }
 }

src/JustSaying/AwsTools/MessageHandling/Dispatch/IMessageDispatcher.cs

@@ -1,7 +1,6 @@
 using System.Threading;
 using System.Threading.Tasks;
 using JustSaying.Messaging.Channels.Context;
-using JustSaying.Messaging.Interrogation;
 
 namespace JustSaying.AwsTools.MessageHandling.Dispatch
 {

src/JustSaying/AwsTools/MessageHandling/Dispatch/MessageDispatcher.cs

@@ -11,6 +11,7 @@
 using JustSaying.Messaging.MessageHandling;
 using JustSaying.Messaging.MessageProcessingStrategies;
 using JustSaying.Messaging.MessageSerialization;
+using JustSaying.Messaging.Middleware;
 using JustSaying.Messaging.Monitoring;
 using Microsoft.Extensions.Logging;
 using Message = JustSaying.Models.Message;
@@ -21,7 +22,7 @@ public class MessageDispatcher : IMessageDispatcher
     {
         private readonly IMessageSerializationRegister _serializationRegister;
         private readonly IMessageMonitor _messagingMonitor;
-        private readonly HandlerMap _handlerMap;
+        private readonly MiddlewareMap _middlewareMap;
         private readonly IMessageBackoffStrategy _messageBackoffStrategy;
         private readonly IMessageContextAccessor _messageContextAccessor;
 
@@ -30,14 +31,14 @@ public class MessageDispatcher : IMessageDispatcher
         public MessageDispatcher(
             IMessageSerializationRegister serializationRegister,
             IMessageMonitor messagingMonitor,
-            HandlerMap handlerMap,
+            MiddlewareMap middlewareMap,
             ILoggerFactory loggerFactory,
             IMessageBackoffStrategy messageBackoffStrategy,
             IMessageContextAccessor messageContextAccessor)
         {
             _serializationRegister = serializationRegister;
             _messagingMonitor = messagingMonitor;
-            _handlerMap = handlerMap;
+            _middlewareMap = middlewareMap;
             _logger = loggerFactory.CreateLogger("JustSaying");
             _messageBackoffStrategy = messageBackoffStrategy;
             _messageContextAccessor = messageContextAccessor;
@@ -70,7 +71,7 @@ public class MessageDispatcher : IMessageDispatcher
                     _messageContextAccessor.MessageContext =
                         new MessageContext(messageContext.Message, messageContext.QueueUri, attributes);
 
-                    handlingSucceeded = await CallMessageHandler(messageContext.QueueName, typedMessage)
+                    handlingSucceeded = await CallMessageHandler(messageContext.QueueName, typedMessage, cancellationToken)
                         .ConfigureAwait(false);
                 }
 
@@ -79,8 +80,9 @@ public class MessageDispatcher : IMessageDispatcher
                     await messageContext.DeleteMessageFromQueueAsync(cancellationToken).ConfigureAwait(false);
                 }
             }
+
 #pragma warning disable CA1031
-            catch (Exception ex)
+            catch (Exception ex) when(!(ex is OperationCanceledException))
 #pragma warning restore CA1031
             {
                 _logger.LogError(
@@ -157,36 +159,38 @@ public class MessageDispatcher : IMessageDispatcher
             }
         }
 
-        private async Task<bool> CallMessageHandler(string queueName, Message message)
+        private async Task<bool> CallMessageHandler(string queueName, Message message, CancellationToken cancellationToken)
         {
             var messageType = message.GetType();
 
-            var handler = _handlerMap.Get(queueName, messageType);
+            var middleware = _middlewareMap.Get(queueName, messageType);
 
-            if (handler == null)
+            if (middleware == null)
             {
                 _logger.LogError(
-                    "Failed to dispatch. Handler for message of type '{MessageTypeName}' not found in handler map.",
+                    "Failed to dispatch. Middleware for message of type '{MessageTypeName}' not found in middleware map.",
                     message.GetType().FullName);
                 return false;
             }
 
             var watch = System.Diagnostics.Stopwatch.StartNew();
 
-            using (_messagingMonitor.MeasureHandler())
+            using (_messagingMonitor.MeasureDispatch())
             {
-                bool handlerSucceeded = false;
+                bool dispatchSuccessful = false;
                 try
                 {
-                    handlerSucceeded = await handler(message).ConfigureAwait(false);
+                    var context = new HandleMessageContext(message, messageType, queueName);
+                    dispatchSuccessful = await middleware.RunAsync(context, null, cancellationToken)
+                        .ConfigureAwait(false);
                 }
                 finally
                 {
                     watch.Stop();
 
                     var logMessage =
                         "{Status} handling message with Id '{MessageId}' of type {MessageType} in {TimeToHandle}ms.";
-                    if (handlerSucceeded)
+                    if (dispatchSuccessful)
                     {
                         _logger.LogInformation(logMessage,
                             "Succeeded",
@@ -204,7 +208,7 @@ private async Task<bool> CallMessageHandler(string queueName, Message message)
                     }
                 }
 
-                return handlerSucceeded;
+                return dispatchSuccessful;
             }
         }