fix: use message receiver io processor to correctly stop message pump during circuit breaker

docs/preview/02-Features/06-general-messaging.md

@@ -24,21 +24,30 @@ To interact with the message processing system within your custom message handle
 using Arcus.Messaging.Abstractions.ServiceBus.MessageHandling;
 using Arcus.Messaging.Pumps.Abstractions.Resiliency;
 
-public class OrderMessageHandler : IAzureServiceBusMessageHandler<Order>
+public class OrderMessageHandler : CircuitBreakerServiceBusMessageHandler<Order>
 {
     private readonly IMessagePumpCircuitBreaker _circuitBreaker;
 
-    public OrderMessageHandler(IMessagePumpCircuitBreaker circuitBreaker)
+    public OrderMessageHandler(IMessagePumpCircuitBreaker circuitBreaker, ILogger<...> logger) : base(circuitBreaker, logger)
     {
         _circuitBreaker = circuitBreaker;
     }
 
-    public async Task ProcessMessageAsync(Order message, AzureServiceBusMessageContext messageContext, ...)
+    public override async Task ProcessMessageAsync(
+        Order message,
+        AzureServiceBusMessageContext messageContext,
+        MessagePumpCircuitBreakerOptions options,
+        ...)
     {
         // Determine whether your dependent system is healthy...
-
-        // If not, call the circuit breaker, processing will be halted temporarily.
-        await _circuitBreaker.PauseMessageProcessingAsync(messageContext.JobId);
+        if (!IsDependentSystemHealthy())
+        {
+            throw new ...Exception("My dependency system is temporarily unavailable, please halt message processing for now");
+        }
+        else
+        {
+            // Process your message...
+        }
     }
 }
 ```
@@ -47,24 +56,26 @@ The message pump will by default act in the following pattern:
 * Circuit breaker calls `Pause`
   * Message pump stops processing messages for a period of time (circuit is OPEN).
 * Message pump tries processing a single message (circuit is HALF-OPEN).
-  * Dependency still unhealthy? => Tries again later (circuit is OPEN)
-  * Dependency healthy? => Message pump starts receiving message in full again (circuit is CLOSED).
+  * Dependency still unhealthy? => circuit breaker pauses again (circuit is OPEN)
+  * Dependency healthy? => circuit breaker resumes, message pump starts receiving message in full again (circuit is CLOSED).
 
 Both the recovery period after the circuit is open and the interval between messages when the circuit is half-open is configurable when calling the circuit breaker. These time periods are related to your dependent system and could change by the type of transient connection failure.
 
 ```csharp
-await _circuitBreaker.PauseMessageProcessingAsync(
-    messageContext.JobId,
-    options =>
-    {
-        // Sets the time period the circuit breaker should wait before retrying to receive messages.
-        // A.k.a. the time period the circuit is closed (default: 30 seconds).
-        options.MessageRecoveryPeriod = TimeSpan.FromSeconds(15);
-
-        // Sets the time period the circuit breaker should wait between each message after the circuit was closed, during recovery.
-        // A.k.a. the time interval to receive messages during which the circuit is half-open (default: 10 seconds).
-        options.MessageIntervalDuringRecovery = TimeSpan.FromSeconds(1.5);
-    });
+ public override async Task ProcessMessageAsync(
+        Order message,
+        AzureServiceBusMessageContext messageContext,
+        MessagePumpCircuitBreakerOptions options,
+        ...)
+{
+    // Sets the time period the circuit breaker should wait before retrying to receive messages.
+    // A.k.a. the time period the circuit is closed (default: 30 seconds).
+    options.MessageRecoveryPeriod = TimeSpan.FromSeconds(15);
+
+    // Sets the time period the circuit breaker should wait between each message after the circuit was closed, during recovery.
+    // A.k.a. the time interval to receive messages during which the circuit is half-open (default: 10 seconds).
+    options.MessageIntervalDuringRecovery = TimeSpan.FromSeconds(1.5);
+}
 ```
 
 ### Pause message processing for a fixed period of time 

src/Arcus.Messaging.Abstractions.ServiceBus/Extensions/ProcessMessageEventArgsExtensions.cs

@@ -19,6 +19,7 @@ public static class ProcessMessageEventArgsExtensions
         /// <exception cref="TypeNotFoundException">Thrown when the no Azure Service Bus receiver could be found on the <paramref name="args"/>.</exception>
         /// <exception cref="InvalidOperationException">Thrown when no value could be found for the Azure Service Bus receiver on the <paramref name="args"/>.</exception>
         /// <exception cref="InvalidCastException">Thrown when the value for the Azure Service Bus receiver on the <paramref name="args"/> wasn't the expected type.</exception>
+        [Obsolete("Service Bus receiver is used internally instead, no need to go via the processor")]
         public static ServiceBusReceiver GetServiceBusReceiver(this ProcessMessageEventArgs args)
         {
             Guard.NotNull(args, nameof(args), "Requires an event args instance to retrieve the original Service Bus message receiver");

src/Arcus.Messaging.Abstractions.ServiceBus/MessageHandling/AzureServiceBusMessageRouter.cs

@@ -269,7 +269,20 @@ private async Task TryRouteMessageWithPotentialFallbackAsync(
             try
             {
                 MessageHandler[] messageHandlers = GetRegisteredMessageHandlers(serviceProvider).ToArray();
-                EnsureAnyMessageHandlerAvailable(messageHandlers, messageContext);
+                if (messageHandlers.Length <= 0)
+                {
+                    if (messageReceiver != null)
+                    {
+                        Logger.LogError("Failed to process Azure Service Bus message '{MessageId}' in pump '{JobId}' as no message handler was matched against the message and no fallback message handlers was configured, dead-lettering message!", message.MessageId, messageContext.JobId);
+                        await messageReceiver.DeadLetterMessageAsync(message); 
+                    }
+
+                    throw new InvalidOperationException(
+                        $"Azure Service Bus message router cannot correctly process the message in the '{nameof(AzureServiceBusMessageContext)}' "
+                        + "because no 'IAzureServiceBusMessageHandler<>' was registered in the dependency injection container. "
+                        + $"Make sure you call the correct 'WithServiceBusMessageHandler' extension on the {nameof(IServiceCollection)} "
+                        + "during the registration of the Azure Service Bus message pump or message router to register a message handler");
+                }
 
                 Encoding encoding = messageContext.GetMessageEncodingProperty(Logger);
                 string messageBody = encoding.GetString(message.Body.ToArray());
@@ -293,13 +306,43 @@ private async Task TryRouteMessageWithPotentialFallbackAsync(
                     }
                 }
 
-                if (!hasGoneThroughMessageHandler)
+                ServiceBusFallbackMessageHandler[] serviceBusFallbackHandlers = 
+                    GetAvailableFallbackMessageHandlersByContext<ServiceBusReceivedMessage, AzureServiceBusMessageContext>(messageContext);
+
+                FallbackMessageHandler<string, MessageContext>[] generalFallbackHandlers = 
+                    GetAvailableFallbackMessageHandlersByContext<string, MessageContext>(messageContext);
+
+                bool fallbackAvailable = serviceBusFallbackHandlers.Length > 0 || generalFallbackHandlers.Length > 0;
+
+                if (hasGoneThroughMessageHandler && !fallbackAvailable)
+                {
+                    if (messageReceiver != null)
+                    {
+                        Logger.LogError("Failed to process Azure Service Bus message '{MessageId}' in pump '{JobId}' as the matched message handler did not successfully processed the message and no fallback message handlers configured, abandoning message!", message.MessageId, messageContext.JobId);
+                        await messageReceiver.AbandonMessageAsync(message); 
+                    }
+                }
+                else if (!hasGoneThroughMessageHandler && !fallbackAvailable)
+                {
+                    if (messageReceiver != null)
+                    {
+                        Logger.LogError("Failed to process Azure Service Bus message '{MessageId}' in pump '{JobId}' as no message handler was matched against the message and no fallback message handlers was configured, dead-lettering message!", message.MessageId, messageContext.JobId);
+                        await messageReceiver.DeadLetterMessageAsync(message); 
+                    }
+                }
+                else
                 {
-                    EnsureFallbackMessageHandlerAvailable(messageContext); 
+                    bool isProcessedByGeneralFallback = await TryFallbackProcessMessageAsync(messageBody, messageContext, correlationInfo, cancellationToken);
+                    if (isProcessedByGeneralFallback)
+                    {
+                        return;
+                    }
+                    else
+                    {
+                        await TryServiceBusFallbackMessageAsync(messageReceiver, message, messageContext, correlationInfo, cancellationToken);
+                    }
                 }
 
-                await TryFallbackProcessMessageAsync(messageBody, messageContext, correlationInfo, cancellationToken);
-                await TryServiceBusFallbackMessageAsync(messageReceiver, message, messageContext, correlationInfo, cancellationToken);
             }
             catch (Exception exception)
             {
@@ -308,42 +351,6 @@ private async Task TryRouteMessageWithPotentialFallbackAsync(
             }
         }
 
-        private void EnsureAnyMessageHandlerAvailable(MessageHandler[] messageHandlers, AzureServiceBusMessageContext messageContext)
-        {
-            ServiceBusFallbackMessageHandler[] serviceBusFallbackHandlers = 
-                GetAvailableFallbackMessageHandlersByContext<ServiceBusReceivedMessage, AzureServiceBusMessageContext>(messageContext);
-
-            FallbackMessageHandler<string, MessageContext>[] generalFallbackHandlers = 
-                GetAvailableFallbackMessageHandlersByContext<string, MessageContext>(messageContext);
-
-            if (messageHandlers.Length <= 0 && serviceBusFallbackHandlers.Length <= 0 && generalFallbackHandlers.Length <= 0)
-            {
-                throw new InvalidOperationException(
-                    $"Azure Service Bus message router cannot correctly process the message in the '{nameof(AzureServiceBusMessageContext)}' "
-                    + "because no 'IAzureServiceBusMessageHandler<>' was registered in the dependency injection container. "
-                    + $"Make sure you call the correct 'WithServiceBusMessageHandler' extension on the {nameof(IServiceCollection)} "
-                    + "during the registration of the Azure Service Bus message pump or message router to register a message handler");
-            }
-        }
-
-        private void EnsureFallbackMessageHandlerAvailable(AzureServiceBusMessageContext messageContext)
-        {
-            ServiceBusFallbackMessageHandler[] serviceBusFallbackHandlers = 
-                GetAvailableFallbackMessageHandlersByContext<ServiceBusReceivedMessage, AzureServiceBusMessageContext>(messageContext);
-
-            FallbackMessageHandler<string, MessageContext>[] generalFallbackHandlers = 
-                GetAvailableFallbackMessageHandlersByContext<string, MessageContext>(messageContext);
-
-            if (serviceBusFallbackHandlers.Length <= 0 && generalFallbackHandlers.Length <= 0)
-            {
-                throw new InvalidOperationException(
-                    $"Azure Service Bus message router cannot correctly process the message in the '{nameof(AzureServiceBusMessageContext)}' "
-                    + "because none of the registered 'IAzureServiceBusMessageHandler<,>' implementations in the dependency injection container matches the incoming message type and context; "
-                    + $"and no '{nameof(IFallbackMessageHandler)}' or '{nameof(IAzureServiceBusFallbackMessageHandler)}' was registered to fall back to."
-                    + $"Make sure you call the correct '.WithServiceBusMessageHandler' extension on the {nameof(IServiceCollection)} during the registration of the message pump or message router to register a message handler");
-            }
-        }
-
         /// <summary>
         /// Sets the Azure Service Bus properties on registered <see cref="IAzureServiceBusMessageHandler{TMessage}"/>s
         /// that implements the <see cref="AzureServiceBusMessageHandler{TMessage}"/> for calling specific Service Bus operations during the message processing.
@@ -402,12 +409,6 @@ protected async Task TryServiceBusFallbackMessageAsync(
 
             ServiceBusFallbackMessageHandler[] fallbackHandlers = 
                 GetAvailableFallbackMessageHandlersByContext<ServiceBusReceivedMessage, AzureServiceBusMessageContext>(messageContext);
-
-            if (fallbackHandlers.Length <= 0)
-            {
-                Logger.LogTrace("No Azure Service Bus message handlers found within message context (JobId: {JobId})", messageContext.JobId);
-                return;
-            }
 
             foreach (ServiceBusFallbackMessageHandler handler in fallbackHandlers)
             {
@@ -431,11 +432,17 @@ protected async Task TryServiceBusFallbackMessageAsync(
                 if (result)
                 {
                     Logger.LogTrace("Fallback message handler '{FallbackMessageHandlerType}' has processed the message", fallbackMessageHandlerTypeName);
-                    break;
+                    return;
                 }
 
                 Logger.LogTrace("Fallback message handler '{FallbackMessageHandlerType}' was not able to process the message", fallbackMessageHandlerTypeName);
             }
+
+            if (messageReceiver != null)
+            {
+                Logger.LogWarning("No fallback message handler processed the Azure Service Bus message '{MessageId}' in pump '{JobId}', abandoning message!", message.MessageId, messageContext.JobId);
+                await messageReceiver.AbandonMessageAsync(message);
+            }
         }
     }
 }

src/Arcus.Messaging.Abstractions/MessageHandling/FallbackMessageHandler.cs

@@ -91,8 +91,16 @@ public async Task<bool> ProcessMessageAsync(
 
             if (CanProcessMessageBasedOnContext(messageContext))
             {
-                await MessageHandlerInstance.ProcessMessageAsync(message, messageContext, correlationInfo, cancellationToken);
-                return true;
+                try
+                {
+                    await MessageHandlerInstance.ProcessMessageAsync(message, messageContext, correlationInfo, cancellationToken);
+                    return true;
+                }
+                catch (Exception exception)
+                {
+                    _logger.LogError(exception, "Fallback message handler '{MessageHandlerType}' failed to handle '{MessageType}' message due to an exception: {Message}", MessageHandlerType.Name, typeof(TMessage).Name, exception.Message);
+                    return false;   
+                }
             }
 
             _logger.LogTrace("Fallback message handler '{FallbackMessageHandlerType}' cannot handle message because it was not registered in the correct message context (JobId: {JobId})", MessageHandlerType.Name, _jobId);

src/Arcus.Messaging.EventHubs.Core/Arcus.Messaging.EventHubs.Core.csproj

@@ -28,8 +28,8 @@
 
   <ItemGroup>
     <PackageReference Include="Arcus.Security.Core" Version="[2.0.0,3.0.0)" />
-    <PackageReference Include="Azure.Messaging.EventHubs.Processor" Version="5.7.1" />
-    <PackageReference Include="Microsoft.Extensions.Azure" Version="1.2.0" />
+    <PackageReference Include="Azure.Messaging.EventHubs.Processor" Version="5.11.3" />
+    <PackageReference Include="Microsoft.Extensions.Azure" Version="1.7.4" />
   </ItemGroup>
 
   <ItemGroup>

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

@@ -11,6 +11,27 @@
 
 namespace Arcus.Messaging.Pumps.Abstractions
 {
+    /// <summary>
+    /// Represents the available states in which the <see cref="MessagePump"/> is presently in within the circuit breaker context
+    /// </summary>
+    public enum MessagePumpCircuitState
+    {
+        /// <summary>
+        /// The message pump is able to receive messages.
+        /// </summary>
+        Closed,
+
+        /// <summary>
+        /// The message pump is under inspection if it can receive messages.
+        /// </summary>
+        HalfOpen,
+
+        /// <summary>
+        /// The message pump is unable to receive messages.
+        /// </summary>
+        Open
+    }
+
     /// <summary>
     /// Represents the foundation for building message pumps.
     /// </summary>
@@ -44,7 +65,12 @@ protected MessagePump(IConfiguration configuration, IServiceProvider serviceProv
         /// <summary>
         /// Gets the boolean flag that indicates whether the message pump is started and receiving messages.
         /// </summary>
-        public bool IsStarted { get; private set; }
+        public bool IsStarted { get; protected set; }
+
+        /// <summary>
+        /// Gets the current state of the message pump within the circuit breaker context.
+        /// </summary>
+        public MessagePumpCircuitState CircuitState { get; private set; } = MessagePumpCircuitState.Closed;
 
         /// <summary>
         /// Gets hte ID of the client being used to connect to the messaging service.
@@ -105,6 +131,7 @@ public override async Task StopAsync(CancellationToken cancellationToken)
         /// </returns>
         public virtual Task<MessageProcessingResult> TryProcessProcessSingleMessageAsync(MessagePumpCircuitBreakerOptions options)
         {
+            CircuitState = MessagePumpCircuitState.HalfOpen;
             return Task.FromResult(MessageProcessingResult.Success);
         }
 
@@ -115,6 +142,8 @@ public virtual Task<MessageProcessingResult> TryProcessProcessSingleMessageAsync
         public virtual Task StartProcessingMessagesAsync(CancellationToken cancellationToken)
         {
             IsStarted = true;
+            CircuitState = MessagePumpCircuitState.Closed;
+
             return Task.CompletedTask;
         }
 
@@ -125,6 +154,8 @@ public virtual Task StartProcessingMessagesAsync(CancellationToken cancellationT
         public virtual Task StopProcessingMessagesAsync(CancellationToken cancellationToken)
         {
             IsStarted = false;
+            CircuitState = MessagePumpCircuitState.Open;
+
             return Task.CompletedTask;
         }