Moved message consumption logic to the MessageHandlingService from QueueService. Added documentation for wildcard extension classes.

Author: Anton Vorontsov

src/RabbitMQ.Client.Core.DependencyInjection/Extensions/TreeNode.cs

@@ -0,0 +1,33 @@
+using System.Collections.Generic;
+
+namespace RabbitMQ.Client.Core.DependencyInjection.Extensions
+{
+    /// <summary>
+    /// A model that represents nodes supposed for building a tree (trie) structure for a pattern (wildcard) matching.
+    /// </summary>
+    public class TreeNode
+    {
+        /// <summary>
+        /// Part of the routing key.
+        /// </summary>
+        /// <remarks>
+        /// Routing key split into parts by the dot.
+        /// </remarks>
+        public string KeyPartition { get; set; }
+
+        /// <summary>
+        /// Parent node.
+        /// </summary>
+        public TreeNode Parent { get; set; }
+
+        /// <summary>
+        /// Child nodes.
+        /// </summary>
+        public List<TreeNode> Nodes { get; } = new List<TreeNode>();
+        
+        /// <summary>
+        /// Flag is the node "last" - a single word that comes without any other parts.
+        /// </summary>
+        public bool IsLastNode { get; set; }
+    }
+}

…ependencyInjection/WildcardExtensions.cs …jection/Extensions/WildcardExtensions.cssrc/RabbitMQ.Client.Core.DependencyInjection/WildcardExtensions.cs renamed to src/RabbitMQ.Client.Core.DependencyInjection/Extensions/WildcardExtensions.cs src/RabbitMQ.Client.Core.DependencyInjection/WildcardExtensions.cs renamed to src/RabbitMQ.Client.Core.DependencyInjection/Extensions/WildcardExtensions.cs

@@ -1,19 +1,35 @@
 using System.Collections.Generic;
 using System.Linq;
 
-namespace RabbitMQ.Client.Core.DependencyInjection
+namespace RabbitMQ.Client.Core.DependencyInjection.Extensions
 {
+    /// <summary>
+    /// An extension class that contains functionality of pattern (wildcard) matching.
+    /// </summary>
+    /// <remarks>
+    /// Methods of that class allows finding route patterns by which message handlers are "listening" for messages.
+    /// </remarks>
     public static class WildcardExtensions
     {
-        private const string Separator = ".";
-        private const string SingleWordPattern = "*";
-        private const string MultipleWordsPattern = "#";
+        const string Separator = ".";
+        const string SingleWordPattern = "*";
+        const string MultipleWordsPattern = "#";
 
-        public static IEnumerable<TreeNode> ConstructTree(IEnumerable<string> routingKeyBindings)
+        /// <summary>
+        /// Construct tree (trie) structure of message handler route patterns.
+        /// </summary>
+        /// <param name="routePatterns">
+        /// Collection of message handler route patterns, which are used by them for a message "listening".
+        /// </param>
+        /// <returns>
+        /// Collection of tree nodes <see cref="TreeNode"/>.
+        /// Depending on routing key bindings that collection can be flat or treelike.
+        /// </returns>
+        public static IEnumerable<TreeNode> ConstructTree(IEnumerable<string> routePatterns)
         {
             var tree = new List<TreeNode>();
 
-            foreach (var binding in routingKeyBindings)
+            foreach (var binding in routePatterns)
             {
                 var keyParts = binding.Split(Separator);
                 TreeNode parentTreeNode = null;
@@ -49,9 +65,16 @@ public static IEnumerable<TreeNode> ConstructTree(IEnumerable<string> routingKey
             return tree;
         }
 
-        public static IEnumerable<string> GetMatchingRoutePatterns(IEnumerable<TreeNode> bindingsTree, string[] routingKeyParts, int depth = 0)
+        /// <summary>
+        /// Get route patterns that match the given routing key.
+        /// </summary>
+        /// <param name="tree">Collection (tree, trie) of nodes.</param>
+        /// <param name="routingKeyParts">Array of routing key parts split by dots.</param>
+        /// <param name="depth">Nesting level of the tree.</param>
+        /// <returns>Collection of route patterns that correspond to the given routing key.</returns>
+        public static IEnumerable<string> GetMatchingRoutePatterns(IEnumerable<TreeNode> tree, string[] routingKeyParts, int depth = 0)
         {
-            foreach (var node in bindingsTree)
+            foreach (var node in tree)
             {
                 var matchingPart = routingKeyParts[depth];
                 if (node.KeyPartition == MultipleWordsPattern)
@@ -91,15 +114,15 @@ public static IEnumerable<string> GetMatchingRoutePatterns(IEnumerable<TreeNode>
             }
         }
 
-        private static IEnumerable<string[]> CollectRoutingKeyTails(IReadOnlyCollection<string> routingKeyParts, int depthStart)
+        static IEnumerable<string[]> CollectRoutingKeyTails(IReadOnlyCollection<string> routingKeyParts, int depthStart)
         {
             for (var index = depthStart; index < routingKeyParts.Count; index++)
             {
                 yield return routingKeyParts.Skip(index).ToArray();
             }
         }
 
-        private static string CollectRoutingKeyInReverseOrder(TreeNode node, string routingKey = "")
+        static string CollectRoutingKeyInReverseOrder(TreeNode node, string routingKey = "")
         {
             routingKey = string.IsNullOrEmpty(routingKey) ? node.KeyPartition : $"{node.KeyPartition}.{routingKey}";
             return node.Parent != null ? CollectRoutingKeyInReverseOrder(node.Parent, routingKey) : routingKey;

src/RabbitMQ.Client.Core.DependencyInjection/IMessageHandlingService.cs

@@ -0,0 +1,17 @@
+using RabbitMQ.Client.Events;
+
+namespace RabbitMQ.Client.Core.DependencyInjection
+{
+    /// <summary>
+    /// An interface of the service that contains logic of handling message consumption events and passing them to the message handlers.
+    /// </summary>
+    public interface IMessageHandlingService
+    {
+        /// <summary>
+        /// Handle message consumption event.
+        /// </summary>
+        /// <param name="eventArgs">Arguments of message consumption event.</param>
+        /// <param name="queueService">An instance of the queue service <see cref="IQueueService"/>.</param>
+        void HandleMessage(BasicDeliverEventArgs @eventArgs, IQueueService queueService);
+    }
+}

src/RabbitMQ.Client.Core.DependencyInjection/MessageHandlerRouter.cs

@@ -6,7 +6,7 @@ namespace RabbitMQ.Client.Core.DependencyInjection
     /// <summary>
     /// Message handler router model.
     /// </summary>
-    internal class MessageHandlerRouter
+    public class MessageHandlerRouter
     {
         /// <summary>
         /// Message Handler Type

src/RabbitMQ.Client.Core.DependencyInjection/MessageHandlingService.cs

@@ -0,0 +1,185 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using RabbitMQ.Client.Events;
+
+namespace RabbitMQ.Client.Core.DependencyInjection
+{
+    /// <summary>
+    /// An implementation of the service that handles message consumption events.
+    /// </summary>
+    public class MessageHandlingService : IMessageHandlingService
+    {
+        const int ResendTimeout = 60;
+        
+        readonly IEnumerable<RabbitMqExchange> _exchanges;
+        readonly IDictionary<Type, List<string>> _messageHandlerRouters;
+        readonly IDictionary<string, IList<IMessageHandler>> _messageHandlers;
+        readonly IDictionary<string, IList<IAsyncMessageHandler>> _asyncMessageHandlers;
+        readonly IDictionary<string, IList<INonCyclicMessageHandler>> _nonCyclicHandlers;
+        readonly IDictionary<string, IList<IAsyncNonCyclicMessageHandler>> _asyncNonCyclicHandlers;
+        readonly ILogger<MessageHandlingService> _logger;
+        
+        public MessageHandlingService(
+            IEnumerable<RabbitMqExchange> exchanges,
+            IEnumerable<MessageHandlerRouter> routers,
+            IEnumerable<IMessageHandler> messageHandlers,
+            IEnumerable<IAsyncMessageHandler> asyncMessageHandlers,
+            IEnumerable<INonCyclicMessageHandler> nonCyclicHandlers,
+            IEnumerable<IAsyncNonCyclicMessageHandler> asyncNonCyclicHandlers,
+            ILogger<MessageHandlingService> logger)
+        {
+            _exchanges = exchanges;
+            _messageHandlerRouters = TransformMessageHandlerRouters(routers);
+            _messageHandlers = TransformMessageHandlersCollection(messageHandlers);
+            _asyncMessageHandlers = TransformMessageHandlersCollection(asyncMessageHandlers);
+            _nonCyclicHandlers = TransformMessageHandlersCollection(nonCyclicHandlers);
+            _asyncNonCyclicHandlers = TransformMessageHandlersCollection(asyncNonCyclicHandlers);
+            _logger = logger;
+        }
+        
+        /// <summary>
+        /// Handle message consumption event.
+        /// </summary>
+        /// <param name="eventArgs">Arguments of message consumption event.</param>
+        /// <param name="queueService">An instance of the queue service <see cref="IQueueService"/>.</param>
+        public void HandleMessage(BasicDeliverEventArgs eventArgs, IQueueService queueService)
+        {
+            var message = Encoding.UTF8.GetString(eventArgs.Body);
+
+                _logger.LogInformation($"New message was received with deliveryTag {eventArgs.DeliveryTag}.");
+                _logger.LogInformation(message);
+
+                try
+                {
+                    if (_asyncMessageHandlers.ContainsKey(eventArgs.RoutingKey))
+                    {
+                        var tasks = new List<Task>();
+                        foreach (var handler in _asyncMessageHandlers[eventArgs.RoutingKey])
+                        {
+                            tasks.Add(RunAsyncHandler(handler, message, eventArgs.RoutingKey));
+                        }
+                        Task.WaitAll(tasks.ToArray());
+                    }
+                    if (_messageHandlers.ContainsKey(eventArgs.RoutingKey))
+                    {
+                        foreach (var handler in _messageHandlers[eventArgs.RoutingKey])
+                        {
+                            _logger.LogDebug($"Starting processing the message by message handler {handler?.GetType().Name}.");
+                            handler.Handle(message, eventArgs.RoutingKey);
+                            _logger.LogDebug($"The message has been processed by message handler {handler?.GetType().Name}.");
+                        }
+                    }
+                    if (_asyncNonCyclicHandlers.ContainsKey(eventArgs.RoutingKey))
+                    {
+                        var tasks = new List<Task>();
+                        foreach (var handler in _asyncNonCyclicHandlers[eventArgs.RoutingKey])
+                        {
+                            tasks.Add(RunAsyncNonCyclicHandler(handler, message, eventArgs.RoutingKey, queueService));
+                        }
+                        Task.WaitAll(tasks.ToArray());
+                    }
+                    if (_nonCyclicHandlers.ContainsKey(eventArgs.RoutingKey))
+                    {
+                        foreach (var handler in _nonCyclicHandlers[eventArgs.RoutingKey])
+                        {
+                            _logger.LogDebug($"Starting processing the message by non-cyclic message handler {handler?.GetType().Name}.");
+                            handler.Handle(message, eventArgs.RoutingKey, queueService);
+                            _logger.LogDebug($"The message has been processed by non-cyclic message handler {handler?.GetType().Name}.");
+                        }
+                    }
+                    _logger.LogInformation($"Success message with deliveryTag {eventArgs.DeliveryTag}.");
+                    queueService.Channel.BasicAck(eventArgs.DeliveryTag, false);
+                }
+                catch (Exception exception)
+                {
+                    _logger.LogError(new EventId(), exception, $"An error occurred while processing received message with delivery tag {eventArgs.DeliveryTag}.");
+
+                    queueService.Channel.BasicAck(eventArgs.DeliveryTag, false);
+
+                    if (eventArgs.BasicProperties.Headers is null)
+                    {
+                        eventArgs.BasicProperties.Headers = new Dictionary<string, object>();
+                    }
+
+                    var exchange = _exchanges.FirstOrDefault(x => x.Name == eventArgs.Exchange);
+                    if (exchange is null)
+                    {
+                        _logger.LogError($"Could not detect exchange {eventArgs.Exchange} to detect the necessity of resending the failed message.");
+                        return;
+                    }
+
+                    if (exchange.Options.RequeueFailedMessages
+                        && !string.IsNullOrEmpty(exchange.Options.DeadLetterExchange)
+                        && !eventArgs.BasicProperties.Headers.ContainsKey("requeued"))
+                    {
+                        eventArgs.BasicProperties.Headers.Add("requeued", true);
+                        queueService.Send(eventArgs.Body, eventArgs.BasicProperties, eventArgs.Exchange, eventArgs.RoutingKey, ResendTimeout);
+                        _logger.LogInformation("The failed message has been requeued.");
+                    }
+                    else
+                    {
+                        _logger.LogInformation("The failed message would not be requeued.");
+                    }
+                }
+        }
+
+        async Task RunAsyncHandler(IAsyncMessageHandler handler, string message, string routingKey)
+        {
+            _logger.LogDebug($"Starting processing the message by async message handler {handler?.GetType().Name}.");
+            await handler.Handle(message, routingKey);
+            _logger.LogDebug($"The message has been processed by async message handler {handler?.GetType().Name}.");
+        }
+
+        async Task RunAsyncNonCyclicHandler(IAsyncNonCyclicMessageHandler handler, string message, string routingKey, IQueueService queueService)
+        {
+            _logger.LogDebug($"Starting processing the message by async non-cyclic message handler {handler?.GetType().Name}.");
+            await handler.Handle(message, routingKey, queueService);
+            _logger.LogDebug($"The message has been processed by async non-cyclic message handler {handler?.GetType().Name}.");
+        }
+
+        IDictionary<Type, List<string>> TransformMessageHandlerRouters(IEnumerable<MessageHandlerRouter> routers)
+        {
+            var dictionary = new Dictionary<Type, List<string>>();
+            foreach (var router in routers)
+            {
+                if (dictionary.ContainsKey(router.Type))
+                {
+                    dictionary[router.Type] = dictionary[router.Type].Union(router.RoutingKeys).ToList();
+                }
+                else
+                {
+                    dictionary.Add(router.Type, router.RoutingKeys);
+                }
+            }
+            return dictionary;
+        }
+        
+        IDictionary<string, IList<T>> TransformMessageHandlersCollection<T>(IEnumerable<T> messageHandlers)
+        {
+            var dictionary = new Dictionary<string, IList<T>>();
+            foreach (var handler in messageHandlers)
+            {
+                var type = handler.GetType();
+                foreach (var routingKey in _messageHandlerRouters[type])
+                {
+                    if (dictionary.ContainsKey(routingKey))
+                    {
+                        if (!dictionary[routingKey].Any(x => x.GetType() == handler.GetType()))
+                        {
+                            dictionary[routingKey].Add(handler);
+                        }
+                    }
+                    else
+                    {
+                        dictionary.Add(routingKey, new List<T> { handler });
+                    }
+                }
+            }
+            return dictionary;
+        }
+    }
+}