[Host.RabbitMq] RbbitMq server disconnection on unhandled routing keys messages

Signed-off-by: Tomasz Maruszak <maruszaktomasz@gmail.com>

Author: zarusz

src/SlimMessageBus.Host.RabbitMQ/Config/Delegates.cs

@@ -30,3 +30,13 @@
 /// </summary>
 /// <param name="option"></param>
 public delegate void RabbitMqMessageConfirmAction(RabbitMqMessageConfirmOptions option);
+
+/// <summary>
+/// Represents the method that handles a RabbitMQ message that includes an routing key that is obsolete or non relevent from the applicatin perspective
+/// Provides access to the message, its properties, and a confirmation action.
+/// </summary>
+/// <remarks>Use this delegate to process messages received from RabbitMQ queues where the routing key is not
+/// relevant or not provided. The handler is responsible for invoking the confirmation action to ensure proper message
+/// acknowledgment.</remarks>
+/// <param name="transportMessage">The event arguments containing the delivered RabbitMQ message and related metadata.</param>
+public delegate RabbitMqMessageConfirmOptions RabbitMqMessageUnrecognizedRoutingKeyHandler(BasicDeliverEventArgs transportMessage);

src/SlimMessageBus.Host.RabbitMQ/Consumers/RabbitMqConsumer.cs

@@ -16,14 +16,16 @@ public class RabbitMqConsumer : AbstractRabbitMqConsumer, IRabbitMqConsumer
     private readonly IMessageProcessor<BasicDeliverEventArgs> _messageProcessor;
     private readonly RoutingKeyMatcherService<IMessageProcessor<BasicDeliverEventArgs>> _routingKeyMatcher;
 
+    private readonly RabbitMqMessageUnrecognizedRoutingKeyHandler _messageUnrecognizedRoutingKeyHandler;
+
     protected override RabbitMqMessageAcknowledgementMode AcknowledgementMode => _acknowledgementMode;
 
     public RabbitMqConsumer(
         ILoggerFactory loggerFactory,
         IRabbitMqChannel channel,
         string queueName,
         IList<ConsumerSettings> consumers,
-        MessageBusBase messageBus,
+        MessageBusBase<RabbitMqMessageBusSettings> messageBus,
         MessageProvider<BasicDeliverEventArgs> messageProvider,
         IHeaderValueConverter headerValueConverter)
         : base(loggerFactory.CreateLogger<RabbitMqConsumer>(),
@@ -33,6 +35,7 @@ public RabbitMqConsumer(
                queueName,
                headerValueConverter)
     {
+        _messageUnrecognizedRoutingKeyHandler = messageBus.ProviderSettings.MessageUnrecognizedRoutingKeyHandler;
         _acknowledgementMode = consumers.Select(x => x.GetOrDefault(RabbitMqProperties.MessageAcknowledgementMode, messageBus.Settings)).FirstOrDefault(x => x != null)
             ?? RabbitMqMessageAcknowledgementMode.ConfirmAfterMessageProcessingWhenNoManualConfirmMade; // be default choose the safer acknowledgement mode
 
@@ -164,6 +167,8 @@ protected override async Task<Exception> OnMessageReceived(Dictionary<string, ob
         else
         {
             Logger.LogDebug("Exchange {Exchange} - Queue {Queue}: No message processor found for routing key {RoutingKey}", transportMessage.Exchange, Path, transportMessage.RoutingKey);
+            var confirmAction = _messageUnrecognizedRoutingKeyHandler(transportMessage);
+            ConfirmMessage(transportMessage, confirmAction, messageHeaders);
         }
 
         // error handling happens in the message processor

src/SlimMessageBus.Host.RabbitMQ/RabbitMqMessageBusSettings.cs

@@ -25,5 +25,11 @@ public string ConnectionString
     /// See the <see cref="DefaultHeaderValueConverter"/>.
     /// </summary>
     public IHeaderValueConverter HeaderValueConverter { get; set; } = new DefaultHeaderValueConverter();
+
+    /// <summary>
+    /// Allows to handle messages that arrive with an unrecognized routing key and decide what to do with them.
+    /// By default the message is Acknowledged.
+    /// </summary>
+    public RabbitMqMessageUnrecognizedRoutingKeyHandler MessageUnrecognizedRoutingKeyHandler { get; set; } = (_) => RabbitMqMessageConfirmOptions.Ack;
 }