Feature - provide specific Azure Service Bus message operations during message handling (#119)

* Feature - provide specific Azure Service Bus message operations during message handling

* pr-sug: add single not-matching message handler to the message pump

* pr-sug: add sending of the order message

* pr-sug: don't autocomplete in dead letter scenario

* pr-sug: extra logging + more strict build pipelines

* pr-sug: use public service type

* pr-fix: use correct dead letter queue

* pr-add: integration test with abandoning the message

* pr-fix: use correct logger property

* pr-add: integration test for abandoning mesasges

* pr-docs: update docs with influence features

* pr-test: update with finishing touches on the comments and clarity

* pr-fix: not autocomplete on abandon

* pr-fix: update with static delivery count

* Update docs/preview/features/message-pumps/service-bus.md

Co-authored-by: Tom Kerkhove <[email protected]>

* Update docs/preview/features/message-pumps/service-bus.md

Co-authored-by: Tom Kerkhove <[email protected]>

* Update src/Arcus.Messaging.Pumps.ServiceBus/AzureServiceBusMessagePump.cs

Co-authored-by: Tom Kerkhove <[email protected]>

* Update src/Arcus.Messaging.Pumps.ServiceBus/AzureServiceBusMessagePump.cs

Co-authored-by: Tom Kerkhove <[email protected]>

* Update src/Arcus.Messaging.Pumps.ServiceBus/MessageHandling/AzureServiceBusFallbackMessageHandler.cs

Co-authored-by: Tom Kerkhove <[email protected]>

* Update src/Arcus.Messaging.Pumps.ServiceBus/MessageHandling/AzureServiceBusFallbackMessageHandler.cs

Co-authored-by: Tom Kerkhove <[email protected]>

* Update src/Arcus.Messaging.Pumps.ServiceBus/MessageHandling/AzureServiceBusFallbackMessageHandler.cs

Co-authored-by: Tom Kerkhove <[email protected]>

* Update src/Arcus.Messaging.Pumps.ServiceBus/MessageHandling/AzureServiceBusFallbackMessageHandler.cs

Co-authored-by: Tom Kerkhove <[email protected]>

* Update src/Arcus.Messaging.Pumps.ServiceBus/MessageHandling/AzureServiceBusFallbackMessageHandler.cs

Co-authored-by: Tom Kerkhove <[email protected]>

* pr-sug: add xml docs for 'Service' prop

* pr-sug: update logs with extra information and lower cases

Co-authored-by: Tom Kerkhove <[email protected]>

Author: stijnmoreels

build/ci-build.yml

@@ -135,7 +135,7 @@ stages:
 
   - stage: SelfContainingIntegrationTests
     displayName: Self-Containing Integration Tests
-    dependsOn: Build
+    dependsOn: DockerIntegrationTests
     condition: succeeded()
     variables:
       - name: 'Arcus.Health.Port.Queue'

build/nuget-release.yml

@@ -120,7 +120,7 @@ stages:
 
   - stage: SelfContainingIntegrationTests
     displayName: Self-Containing Integration Tests
-    dependsOn: Build
+    dependsOn: DockerIntegrationTests
     condition: succeeded()
     variables:
       - name: 'Arcus.Health.Port.Queue'

docs/preview/features/message-pumps/customization.md

@@ -8,6 +8,10 @@ layout: default
 While the message processing is handled by the `IMessageHandler<>` implementations, the message pump controls in what format the message is received.
 We allow several customizations while implementing your own message pump.
 
+- [Control custom deserialization](#control-custom-deserialization)
+- [Filter messages based on message context](#filter-messages-based-on-message-context)
+- [Fallback message handling](#fallback-message-handling)
+
 ## Control custom deserialization
 
 When inheriting from an `...MessagePump` type, there's a way to control how the incoming raw message is being deserialized.

docs/preview/features/message-pumps/service-bus.md

@@ -14,7 +14,6 @@ Azure Service Bus Message Pump will perform all the plumbing that is required fo
 - Provide telemetry
 
 As a user, the only thing you have to do is **focus on processing messages, not how to get them**.
-
 You can do this by creating a message handler which implements from `IAzureServiceBusMessageHandler<TMessage>` (or `IMessageHandler<TMessage, MessageContext>`).
 
 Here is an example of a message handler that expects messages of type `Order`:
@@ -57,6 +56,13 @@ public class OrdersMessageHandler : IMessageHandler<Order>
 }
 ```
 
+Other topics:
+- [Configuration](#configuration)
+- [Customized configuration](#customized-configuration)
+- [Fallback message handling](#fallback-message-handling)
+- [Influence handling of Service Bus message in a message handler](#influence-handling-of-Service-Bus-message-in-message-handler)
+- [Correlation](#correlation)
+
 ## Configuration
 
 Once the message handler is created, you can very easily configure it:
@@ -194,6 +200,89 @@ public void ConfigureServices(IServiceCollection services)
 }
 ```
 
+## Influence handling of Service Bus message in message handler
+
+When an Azure Service Bus message is received (either via regular message handlers or fallback message handlers), we allow specific Azure Service Bus operations during the message handling.
+Currently we support [**Dead letter**](https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dead-letter-queues) and [*Abandon**](https://docs.microsoft.com/en-us/dotnet/api/microsoft.servicebus.messaging.messagereceiver.abandon?view=azure-dotnet).
+
+### During (regular) message handling
+
+To have access to the Azure Service Bus operations, you have to implement the `abstract` `AzureServiceBusMessageHandler<T>` class. 
+Behind the screens it implements the `IMessageHandler<>` interface, so you can register this the same way as your other regular message handlers.
+
+This base class provides several protected methods to call the Azure Service Bus operations:
+- `.DeadLetterMessageAsync`
+- `.AbandonMessageAsync`
+
+Example:
+
+```csharp
+public class AbandonsUnknownOrderMessageHandler : AzureServiceBusMessageHandler<Order>
+{
+    public AbandonsUnknownOrderMessageHandler(ILogger<AbandonsUnknownOrderMessageHandler> logger)
+        : base(logger)
+    {
+    }
+
+    public override async Task ProcessMessageAsync(Order order, AzureServiceBusMessageContext context, ...)
+    {
+        if (order.Id < 1)
+        {
+            await AbandonMessageAsync();
+        }
+        else
+        {
+            Logger.LogInformation("Received valid order");
+        }
+    }
+}
+```
+
+The registration happens the same as any other regular message handler:
+
+```csharp
+public void ConfigureServices(IServiceCollection services)
+{
+    services.WithServiceBusMessageHandler<AbandonUnknownOrderMessageHandler, Order>();
+}
+```
+
+### During fallback message handling
+
+To have access to the Azure Service Bus operations, you have to implement the abstract `AzureServiceBusFallbackMessageHandler` class.
+Behind the scenes it implements the `IServiceBusFallbackMessageHandler`, so you can register this the same way as any other fallback message handler.
+
+This base class provides several protected methods to call the Azure Service Bus operations:
+- `.DeadLetterAsync`
+- `.AbandonAsync`
+
+Example:
+
+```csharp
+public class DeadLetterFallbackMessageHandler : AzureServiceBusFallbackMessageHandler
+{
+    public DeadLetterFallbackMessageHandler(ILogger<DeadLetterFallbackMessageHandler> logger)
+        : base(logger)
+    {
+    }
+
+    public override async Task ProcessMessageAsync(Message message, AzureServiceBusMessageContext context, ...)
+    {
+        Logger.LogInformation("Message is not handled by any message handler, will dead letter");
+        await DeadLetterAsync(message);
+    }
+}
+```
+
+The registration happens the same way as any other fallback message handler:
+
+```csharp
+public void ConfigureServices(IServiceCollection services)
+{
+    services.WithServiceBusFallbackMessageHandler<DeadLetterFallbackMessageHandler>();
+}
+```
+
 ## Correlation
 
 To retrieve the correlation information of Azure Service Bus messages, we provide an extension that wraps all correlation information.
@@ -220,4 +309,4 @@ We provide templates to get started easily:
 - Azure Service Bus Queue Worker Template ([docs](https://templates.arcus-azure.net/features/servicebus-queue-worker-template))
 - Azure Service Bus Topic Worker Template ([docs](https://templates.arcus-azure.net/features/servicebus-topic-worker-template))
 
-[&larr; back](/)
+[&larr; back](/)

src/Arcus.Messaging.Pumps.Abstractions/MessageHandling/IServiceCollectionExtensions.cs

@@ -26,7 +26,7 @@ public static IServiceCollection WithMessageHandler<TMessageHandler, TMessage>(t
         {
             Guard.NotNull(services, nameof(services));
 
-            services.AddSingleton<IMessageHandler<TMessage, MessageContext>, TMessageHandler>();
+            services.AddTransient<IMessageHandler<TMessage, MessageContext>, TMessageHandler>();
 
             return services;
         }
@@ -48,7 +48,7 @@ public static IServiceCollection WithMessageHandler<TMessageHandler, TMessage>(
             Guard.NotNull(services, nameof(services));
             Guard.NotNull(implementationFactory, nameof(implementationFactory));
 
-            services.AddSingleton<IMessageHandler<TMessage, MessageContext>, TMessageHandler>(implementationFactory);
+            services.AddTransient<IMessageHandler<TMessage, MessageContext>, TMessageHandler>(implementationFactory);
 
             return services;
         }
@@ -68,7 +68,7 @@ public static IServiceCollection WithMessageHandler<TMessageHandler, TMessage, T
         {
             Guard.NotNull(services, nameof(services));
 
-            services.AddSingleton<IMessageHandler<TMessage, TMessageContext>, TMessageHandler>();
+            services.AddTransient<IMessageHandler<TMessage, TMessageContext>, TMessageHandler>();
 
             return services;
         }
@@ -92,7 +92,7 @@ public static IServiceCollection WithMessageHandler<TMessageHandler, TMessage, T
             Guard.NotNull(services, nameof(services));
             Guard.NotNull(implementationFactory, nameof(implementationFactory));
 
-            services.AddSingleton<IMessageHandler<TMessage, TMessageContext>, TMessageHandler>(implementationFactory);
+            services.AddTransient<IMessageHandler<TMessage, TMessageContext>, TMessageHandler>(implementationFactory);
 
             return services;
         }
@@ -186,7 +186,7 @@ public static IServiceCollection WithMessageHandler<TMessageHandler, TMessage, T
             Guard.NotNull(messageContextFilter,  nameof(messageContextFilter));
             Guard.NotNull(implementationFactory, nameof(implementationFactory));
 
-            return services.AddSingleton<IMessageHandler<TMessage, TMessageContext>, MessageHandlerRegistration<TMessage, TMessageContext>>(
+            return services.AddTransient<IMessageHandler<TMessage, TMessageContext>, MessageHandlerRegistration<TMessage, TMessageContext>>(
                 serviceProvider => new MessageHandlerRegistration<TMessage, TMessageContext>(
                     messageContextFilter, implementationFactory(serviceProvider)));
         }

src/Arcus.Messaging.Pumps.Abstractions/MessageHandling/MessageHandler.cs

@@ -18,7 +18,6 @@ namespace Arcus.Messaging.Pumps.Abstractions.MessageHandling
     /// </summary>
     public class MessageHandler
     {
-        private readonly object _service;
         private readonly ILogger _logger;
 
         private MessageHandler(Type serviceType, object service, ILogger logger)
@@ -29,19 +28,24 @@ private MessageHandler(Type serviceType, object service, ILogger logger)
             Guard.For<ArgumentException>(
                 () => serviceType.GenericTypeArguments.Length != 2, 
                 $"Message handler type '{serviceType.Name}' has not the expected 2 generic type arguments");
-            
-            _service = service;
+
             _logger = logger;
 
+            Service = service;
             ServiceType = serviceType;
             MessageType = ServiceType.GenericTypeArguments[0];
             MessageContextType = ServiceType.GenericTypeArguments[1];
         }
 
+        /// <summary>
+        /// Gets the instance of the message handler that this abstracted message handler represents.
+        /// </summary>
+        public object Service { get; }
+
         /// <summary>
         /// Gets the type of the message handler that this abstracted message handler represents.
         /// </summary>
-        internal Type ServiceType { get; }
+        public Type ServiceType { get; }
 
         /// <summary>
         /// Gets the type of the message that this abstracted message handler can process.
@@ -104,13 +108,13 @@ public bool CanProcessMessage<TMessageContext>(TMessageContext messageContext) w
                     "Message context type '{ActualMessageContextType}' matches registered message handler's {MessageHandlerType} context type {ExpectedMessageContextType}",
                     actualMessageContextType.Name, ServiceType.Name, expectedMessageContextType.Name);
 
-                if (_service.GetType().Name == typeof(MessageHandlerRegistration<,>).Name)
+                if (Service.GetType().Name == typeof(MessageHandlerRegistration<,>).Name)
                 {
                     _logger.LogTrace(
                         "Determining whether the message context predicate registered with the message handler {MessageHandlerType} holds...",
                          ServiceType.Name);
 
-                    var canProcessMessage = (bool) _service.InvokeMethod(
+                    var canProcessMessage = (bool) Service.InvokeMethod(
                         "CanProcessMessage",
                         BindingFlags.Instance | BindingFlags.NonPublic,
                         messageContext);
@@ -169,13 +173,13 @@ public async Task ProcessMessageAsync<TMessageContext>(
             try
             {
                 var processMessageAsync =
-                        (Task)_service.InvokeMethod(
+                        (Task)Service.InvokeMethod(
                             methodName, BindingFlags.Instance | BindingFlags.Public, message, messageContext, correlationInfo, cancellationToken);
 
                 if (processMessageAsync is null)
                 {
                     throw new InvalidOperationException(
-                        $"The '{typeof(IMessageHandler<,>).Name}' implementation '{_service.GetType().Name}' returned 'null' while calling the '{methodName}' method");
+                        $"The '{typeof(IMessageHandler<,>).Name}' implementation '{Service.GetType().Name}' returned 'null' while calling the '{methodName}' method");
                 }
 
                 await processMessageAsync;
@@ -190,7 +194,7 @@ public async Task ProcessMessageAsync<TMessageContext>(
                     methodName, ServiceType.Name, methodName, ServiceType.Name);
 
                 throw new AmbiguousMatchException(
-                    $"Ambiguous match found of '{methodName}' methods in the '{_service.GetType().Name}'. ", exception);
+                    $"Ambiguous match found of '{methodName}' methods in the '{Service.GetType().Name}'. ", exception);
             }
         }
     }

src/Arcus.Messaging.Pumps.Abstractions/MessagePump.cs

@@ -90,6 +90,17 @@ protected virtual Task HandleReceiveExceptionAsync(Exception receiveException)
             return Task.CompletedTask;
         }
 
+        /// <summary>
+        /// Pre-process the message by setting the necessary values the <see cref="IMessageHandler{TMessage,TMessageContext}"/> implementation.
+        /// </summary>
+        /// <param name="messageHandler">The message handler to be used to process the message.</param>
+        /// <param name="messageContext">The message context of the message that will be handled.</param>
+        protected virtual Task PreProcessMessageAsync<TMessageContext>(MessageHandler messageHandler, TMessageContext messageContext) 
+            where TMessageContext : MessageContext
+        {
+            return Task.CompletedTask;
+        }
+
         /// <summary>
         ///     Handle a new message that was received
         /// </summary>
@@ -173,7 +184,7 @@ private async Task<bool> TryProcessMessageAsync<TMessageContext>(
             CancellationToken cancellationToken)
             where TMessageContext : MessageContext
         {
-            IEnumerable<MessageHandler> handlers = _messageHandlers.Value;
+            IEnumerable<MessageHandler> handlers = MessageHandler.SubtractFrom(ServiceProvider, Logger);
             if (!handlers.Any() && _fallbackMessageHandler is null)
             {
                 throw new InvalidOperationException(
@@ -215,6 +226,7 @@ private async Task<bool> ProcessMessageAsync<TMessageContext>(
                         "Successful parsing from abstracted message to concrete message handler type did unexpectedly result in a 'null' parsing result");
                 }
 
+                await PreProcessMessageAsync(handler, messageContext);
                 await handler.ProcessMessageAsync(result, messageContext, correlationInfo, cancellationToken);
                 return true;
             }