Created models directory and moved models there. Created DI methods for message handlers that listen for routes and exchanges. Set up tests.

Author: Anton Vorontsov

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

@@ -1,5 +1,6 @@
 using System.Collections.Generic;
 using System.Linq;
+using RabbitMQ.Client.Core.DependencyInjection.Models;
 
 namespace RabbitMQ.Client.Core.DependencyInjection.Extensions
 {

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

@@ -5,6 +5,7 @@
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using RabbitMQ.Client.Core.DependencyInjection.Extensions;
+using RabbitMQ.Client.Core.DependencyInjection.Models;
 using RabbitMQ.Client.Events;
 
 namespace RabbitMQ.Client.Core.DependencyInjection
@@ -17,11 +18,7 @@ public class MessageHandlingService : IMessageHandlingService
         const int ResendTimeout = 60;
 
         readonly IEnumerable<RabbitMqExchange> _exchanges;
-        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 IEnumerable<TreeNode> _routeTree;
+        readonly IEnumerable<MessageHandlerContainer> _messageHandlerContainers;
         readonly ILogger<MessageHandlingService> _logger;
 
         public MessageHandlingService(
@@ -34,12 +31,12 @@ public MessageHandlingService(
             ILogger<MessageHandlingService> logger)
         {
             _exchanges = exchanges;
-            var routersDictionary = TransformMessageHandlerRoutersToDictionary(routers);
-            _messageHandlers = TransformMessageHandlersCollectionToDictionary(messageHandlers, routersDictionary);
-            _asyncMessageHandlers = TransformMessageHandlersCollectionToDictionary(asyncMessageHandlers, routersDictionary);
-            _nonCyclicHandlers = TransformMessageHandlersCollectionToDictionary(nonCyclicHandlers, routersDictionary);
-            _asyncNonCyclicHandlers = TransformMessageHandlersCollectionToDictionary(asyncNonCyclicHandlers, routersDictionary);
-            _routeTree = ConstructRoutesTree(routers);
+            _messageHandlerContainers = ConstructMessageHandlerContainers(
+                routers,
+                messageHandlers,
+                asyncMessageHandlers,
+                nonCyclicHandlers,
+                asyncNonCyclicHandlers);
             _logger = logger;
         }
 
@@ -57,8 +54,8 @@ public void HandleMessageReceivingEvent(BasicDeliverEventArgs eventArgs, IQueueS
 
             try
             {
-                var matchingRoutes = GetMatchingRoutePatterns(eventArgs.RoutingKey);
-                ProcessMessage(message, queueService, matchingRoutes);
+                var matchingRoutes = GetMatchingRoutePatterns(eventArgs.Exchange, eventArgs.RoutingKey);
+                ProcessMessage(eventArgs.Exchange, message, queueService, matchingRoutes);
                 queueService.Channel.BasicAck(eventArgs.DeliveryTag, false);
                 _logger.LogInformation($"Message processing finished successfully. Acknowledge has been sent with deliveryTag {eventArgs.DeliveryTag}.");
             }
@@ -95,21 +92,34 @@ public void HandleMessageReceivingEvent(BasicDeliverEventArgs eventArgs, IQueueS
             }
         }
 
-        IEnumerable<string> GetMatchingRoutePatterns(string routingKey)
+        IEnumerable<string> GetMatchingRoutePatterns(string exchange, string routingKey)
         {
+            var tree = _messageHandlerContainers.FirstOrDefault(x => x.Exchange == exchange)?.Tree 
+                ?? _messageHandlerContainers.FirstOrDefault(x => x.IsGeneral)?.Tree;
+            if (tree is null)
+            {
+                return Enumerable.Empty<string>();
+            }
+
             var routingKeyParts = routingKey.Split(".");
-            return WildcardExtensions.GetMatchingRoutePatterns(_routeTree, routingKeyParts).ToList();
+            return WildcardExtensions.GetMatchingRoutePatterns(tree, routingKeyParts).ToList();
         }
 
-        void ProcessMessage(string message, IQueueService queueService, IEnumerable<string> matchingRoutes)
+        void ProcessMessage(string exchange, string message, IQueueService queueService, IEnumerable<string> matchingRoutes)
         {
+            var container = _messageHandlerContainers.FirstOrDefault(x => x.Exchange == exchange) ?? _messageHandlerContainers.FirstOrDefault(x => x.IsGeneral);
+            if (container is null)
+            {
+                return;
+            }
+
             var executedHandlers = new List<Type>();
             foreach (var matchingRoute in matchingRoutes)
             {
-                if (_asyncMessageHandlers.ContainsKey(matchingRoute))
+                if (container.AsyncMessageHandlers.ContainsKey(matchingRoute))
                 {
                     var tasks = new List<Task>();
-                    foreach (var handler in _asyncMessageHandlers[matchingRoute])
+                    foreach (var handler in container.AsyncMessageHandlers[matchingRoute])
                     {
                         var handlerType = handler.GetType();
                         if (executedHandlers.Contains(handlerType))
@@ -122,9 +132,9 @@ void ProcessMessage(string message, IQueueService queueService, IEnumerable<stri
                     }
                     Task.WaitAll(tasks.ToArray());
                 }
-                if (_messageHandlers.ContainsKey(matchingRoute))
+                if (container.MessageHandlers.ContainsKey(matchingRoute))
                 {
-                    foreach (var handler in _messageHandlers[matchingRoute])
+                    foreach (var handler in container.MessageHandlers[matchingRoute])
                     {
                         var handlerType = handler.GetType();
                         if (executedHandlers.Contains(handlerType))
@@ -136,10 +146,10 @@ void ProcessMessage(string message, IQueueService queueService, IEnumerable<stri
                         RunMessageHandler(handler, message, matchingRoute);
                     }
                 }
-                if (_asyncNonCyclicHandlers.ContainsKey(matchingRoute))
+                if (container.AsyncNonCyclicHandlers.ContainsKey(matchingRoute))
                 {
                     var tasks = new List<Task>();
-                    foreach (var handler in _asyncNonCyclicHandlers[matchingRoute])
+                    foreach (var handler in container.AsyncNonCyclicHandlers[matchingRoute])
                     {
                         var handlerType = handler.GetType();
                         if (executedHandlers.Contains(handlerType))
@@ -152,9 +162,9 @@ void ProcessMessage(string message, IQueueService queueService, IEnumerable<stri
                     }
                     Task.WaitAll(tasks.ToArray());
                 }
-                if (_nonCyclicHandlers.ContainsKey(matchingRoute))
+                if (container.NonCyclicHandlers.ContainsKey(matchingRoute))
                 {
-                    foreach (var handler in _nonCyclicHandlers[matchingRoute])
+                    foreach (var handler in container.NonCyclicHandlers[matchingRoute])
                     {
                         var handlerType = handler.GetType();
                         if (executedHandlers.Contains(handlerType))
@@ -201,18 +211,74 @@ async Task RunAsyncNonCyclicMessageHandler(IAsyncNonCyclicMessageHandler handler
             _logger.LogDebug($"The message has been processed by async non-cyclic message handler {handler.GetType().Name}.");
         }
 
-        void ValidateHandler<T>(T messageHandler)
+        static void ValidateHandler<T>(T messageHandler)
         {
             if (messageHandler is null)
             {
                 throw new ArgumentNullException(nameof(messageHandler), "Message handler is null.");
             }
         }
 
-        static IEnumerable<TreeNode> ConstructRoutesTree(IEnumerable<MessageHandlerRouter> routers)
+        static IEnumerable<MessageHandlerContainer> ConstructMessageHandlerContainers(
+            IEnumerable<MessageHandlerRouter> routers,
+            IEnumerable<IMessageHandler> messageHandlers,
+            IEnumerable<IAsyncMessageHandler> asyncMessageHandlers,
+            IEnumerable<INonCyclicMessageHandler> nonCyclicHandlers,
+            IEnumerable<IAsyncNonCyclicMessageHandler> asyncNonCyclicHandlers)
         {
-            var routePatterns = routers.SelectMany(x => x.RoutePatterns).Distinct();
-            return WildcardExtensions.ConstructRoutesTree(routePatterns);
+            var containers = new List<MessageHandlerContainer>();
+            var generalRouters = routers.Where(x => x.IsGeneral).ToList();
+            if (generalRouters.Any())
+            {
+                var container = CreateContailer(
+                    null,
+                    generalRouters,
+                    messageHandlers,
+                    asyncMessageHandlers,
+                    nonCyclicHandlers,
+                    asyncNonCyclicHandlers);
+                containers.Add(container);
+            }
+            
+            var exchanges = routers.Where(x => !x.IsGeneral).Select(x => x.Exchange).Distinct().ToList();
+            foreach (var exchange in exchanges)
+            {
+                var exchangeRouters = routers.Where(x => x.Exchange == exchange).ToList();
+                var container = CreateContailer(
+                    exchange,
+                    exchangeRouters,
+                    messageHandlers,
+                    asyncMessageHandlers,
+                    nonCyclicHandlers,
+                    asyncNonCyclicHandlers);
+                containers.Add(container);
+            }
+            return containers;
+        }
+
+        static MessageHandlerContainer CreateContailer(
+            string exchange,
+            IEnumerable<MessageHandlerRouter> selectedRouters,
+            IEnumerable<IMessageHandler> messageHandlers,
+            IEnumerable<IAsyncMessageHandler> asyncMessageHandlers,
+            IEnumerable<INonCyclicMessageHandler> nonCyclicHandlers,
+            IEnumerable<IAsyncNonCyclicMessageHandler> asyncNonCyclicHandlers)
+        {
+            var routersDictionary = TransformMessageHandlerRoutersToDictionary(selectedRouters);
+            var boundMessageHandlers = messageHandlers.Where(x => routersDictionary.Keys.Contains(x.GetType()));
+            var boundAsyncMessageHandlers = asyncMessageHandlers.Where(x => routersDictionary.Keys.Contains(x.GetType()));
+            var boundNonCyclicMessageHandlers = nonCyclicHandlers.Where(x => routersDictionary.Keys.Contains(x.GetType()));
+            var boundAsyncNonCyclicMessageHandlers = asyncNonCyclicHandlers.Where(x => routersDictionary.Keys.Contains(x.GetType()));
+            var routePatterns = selectedRouters.SelectMany(x => x.RoutePatterns).Distinct().ToList();
+            return new MessageHandlerContainer
+            {
+                Exchange = exchange,
+                Tree = WildcardExtensions.ConstructRoutesTree(routePatterns),
+                MessageHandlers = TransformMessageHandlersCollectionToDictionary(boundMessageHandlers, routersDictionary),
+                AsyncMessageHandlers = TransformMessageHandlersCollectionToDictionary(boundAsyncMessageHandlers, routersDictionary),
+                NonCyclicHandlers = TransformMessageHandlersCollectionToDictionary(boundNonCyclicMessageHandlers, routersDictionary),
+                AsyncNonCyclicHandlers = TransformMessageHandlersCollectionToDictionary(boundAsyncNonCyclicMessageHandlers, routersDictionary)
+            };
         }
 
         static IDictionary<Type, List<string>> TransformMessageHandlerRoutersToDictionary(IEnumerable<MessageHandlerRouter> routers)

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

@@ -1,7 +1,7 @@
-using Microsoft.Extensions.DependencyInjection;
-using System;
+using System;
+using Microsoft.Extensions.DependencyInjection;
 
-namespace RabbitMQ.Client.Core.DependencyInjection
+namespace RabbitMQ.Client.Core.DependencyInjection.Models
 {
     /// <summary>
     /// A service extension for registration exchange singleton "services".

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

@@ -0,0 +1,49 @@
+using System.Collections.Generic;
+
+namespace RabbitMQ.Client.Core.DependencyInjection.Models
+{
+    /// <summary>
+    /// The service model that represents a container that contains a route patterns tree connected with an exchange and all types of message handlers.
+    /// If route patterns configured without an exchange they get in the "general" container.
+    /// </summary>
+    public class MessageHandlerContainer
+    {
+        /// <summary>
+        /// An exchange.
+        /// </summary>
+        /// <remarks>
+        /// Could be null.
+        /// </remarks>
+        public string Exchange { get; set; }
+        
+        /// <summary>
+        /// Route patterns tree (trie) structure.
+        /// </summary>
+        public IEnumerable<TreeNode> Tree { get; set; }
+
+        /// <summary>
+        /// Flag is the container general.
+        /// </summary>
+        public bool IsGeneral => string.IsNullOrEmpty(Exchange);
+
+        /// <summary>
+        /// Dictionary of route patterns and message handlers connected by them.
+        /// </summary>
+        public IDictionary<string, IList<IMessageHandler>> MessageHandlers { get; set; }
+        
+        /// <summary>
+        /// Dictionary of route patterns and async message handlers connected by them.
+        /// </summary>
+        public IDictionary<string, IList<IAsyncMessageHandler>> AsyncMessageHandlers { get; set; }
+        
+        /// <summary>
+        /// Dictionary of route patterns and non-cyclic message handlers connected by them.
+        /// </summary>
+        public IDictionary<string, IList<INonCyclicMessageHandler>> NonCyclicHandlers { get; set; }
+        
+        /// <summary>
+        /// Dictionary of route patterns and async non-cyclic message handlers connected by them.
+        /// </summary>
+        public IDictionary<string, IList<IAsyncNonCyclicMessageHandler>> AsyncNonCyclicHandlers { get; set; }
+    }
+}

…cyInjection/ExchangeServiceDescriptor.cs …tion/Models/ExchangeServiceDescriptor.cssrc/RabbitMQ.Client.Core.DependencyInjection/ExchangeServiceDescriptor.cs renamed to src/RabbitMQ.Client.Core.DependencyInjection/Models/ExchangeServiceDescriptor.cs src/RabbitMQ.Client.Core.DependencyInjection/ExchangeServiceDescriptor.cs renamed to src/RabbitMQ.Client.Core.DependencyInjection/Models/ExchangeServiceDescriptor.cs

@@ -0,0 +1,34 @@
+using System;
+using System.Collections.Generic;
+
+namespace RabbitMQ.Client.Core.DependencyInjection.Models
+{
+    /// <summary>
+    /// Message handler router model.
+    /// </summary>
+    public class MessageHandlerRouter
+    {
+        /// <summary>
+        /// Message Handler Type
+        /// </summary>
+        public Type Type { get; set; }
+        /// <summary>
+        /// Collection of route patterns (routing keys) that handler will be "listening".
+        /// </summary>
+        public List<string> RoutePatterns { get; set; } = new List<string>();
+
+        /// <summary>
+        /// An exchange which is being listened by the message handler by specified route patterns.
+        /// </summary>
+        /// <remarks>
+        /// Exchange can be null, and in that scenario message handler will be general
+        /// (it will listen all messages regardless of an exchange).
+        /// </remarks>
+        public string Exchange { get; set; }
+
+        /// <summary>
+        /// Flag is the message handler general.
+        /// </summary>
+        public bool IsGeneral => string.IsNullOrEmpty(Exchange);
+    }
+}

src/RabbitMQ.Client.Core.DependencyInjection/Models/MessageHandlerContainer.cs

@@ -1,6 +1,6 @@
 using RabbitMQ.Client.Core.DependencyInjection.Configuration;
 
-namespace RabbitMQ.Client.Core.DependencyInjection
+namespace RabbitMQ.Client.Core.DependencyInjection.Models
 {
     /// <summary>
     /// Exchange model.

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

@@ -1,6 +1,6 @@
 using System.Collections.Generic;
 
-namespace RabbitMQ.Client.Core.DependencyInjection.Extensions
+namespace RabbitMQ.Client.Core.DependencyInjection.Models
 {
     /// <summary>
     /// A model that represents nodes supposed for building a tree (trie) structure for a pattern (wildcard) matching.

….DependencyInjection/RabbitMqExchange.cs …encyInjection/Models/RabbitMqExchange.cssrc/RabbitMQ.Client.Core.DependencyInjection/RabbitMqExchange.cs renamed to src/RabbitMQ.Client.Core.DependencyInjection/Models/RabbitMqExchange.cs src/RabbitMQ.Client.Core.DependencyInjection/RabbitMqExchange.cs renamed to src/RabbitMQ.Client.Core.DependencyInjection/Models/RabbitMqExchange.cs

@@ -9,6 +9,7 @@
 using System.Net;
 using System.Text;
 using System.Threading.Tasks;
+using RabbitMQ.Client.Core.DependencyInjection.Models;
 
 namespace RabbitMQ.Client.Core.DependencyInjection
 {