Added non-cycling message handlers.

Author: antonyvorontsov

README.md

@@ -152,6 +152,55 @@ public class CustomAsyncMessageHandler : IAsyncMessageHandler
 }
 ```
 
+But you can not use `IQueueService` inside those message handlers otherwise you will be faced with cycling dependency problem. But sometimes you may need to send something in other queue (e.g. queue with responses) from one message handler or another. For that purposes use non-cyclinc handlers.
+
+```csharp
+public class CustomMessageHandler : INonCyclicMessageHandler
+{
+	readonly ILogger<CustomMessageHandler> _logger;
+	public CustomMessageHandler(ILogger<CustomMessageHandler> logger)
+	{
+		_logger = logger;
+	}
+
+	public void Handle(string message, string routingKey, IQueueService queueService)
+	{
+		_logger.LogInformation("Got a message.");
+		var response = new { Message = message };
+		queueService.Send(response, "exchange.name", "routing.key");
+	}
+}
+```
+
+Or the same but async.
+
+```csharp
+public class CustomAsyncMessageHandler : IAsyncNonCyclicMessageHandler
+{
+	readonly ILogger<CustomAsyncMessageHandler> _logger;
+
+	public CustomAsyncMessageHandler(ILogger<CustomAsyncMessageHandler> logger)
+	{
+		_logger = logger;
+	}
+
+	public async Task Handle(string message, string routingKey, IQueueService queueService)
+	{
+		_logger.LogInformation("Doing something async.");
+		var response = new { Message = message };
+		await queueService.SendAsync(response, "exchange.name", "routing.key");
+	}
+}
+```
+
+And you have to register those classes the same way you did with simple handlers.
+```csharp
+services.AddRabbitMqClient(rabbitMqSection)
+	.AddExchange("exchange.name", exchangeSection)
+	.AddNonCyclicMessageHandlerSingleton<CustomMessageHandler>("routing.key")
+	.AddAsyncNonCyclicMessageHandlerSingleton<CustomAsyncMessageHandler>("other.routing.key");
+```
+
 You can find example projects in the repository too.
 
 ### appsettings.json configuration

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

@@ -0,0 +1,17 @@
+using System.Threading.Tasks;
+
+namespace RabbitMQ.Client.Core.DependencyInjection
+{
+    /// <summary>
+    /// Interface of a non-cycling async message handler.
+    /// </summary>
+    public interface IAsyncNonCyclicMessageHandler
+    {
+        /// <summary>
+        /// Handle message from a queue.
+        /// </summary>
+        /// <param name="message">Json message.</param>
+        /// <param name="routingKey">Routing key.</param>
+        Task Handle(string message, string routingKey, IQueueService queueService);
+    }
+}

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

@@ -0,0 +1,16 @@
+namespace RabbitMQ.Client.Core.DependencyInjection
+{
+    /// <summary>
+    /// Interface of a non-cycling message handler.
+    /// </summary>
+    public interface INonCyclicMessageHandler
+    {
+        /// <summary>
+        /// Handle message from a queue.
+        /// </summary>
+        /// <param name="message">Json message.</param>
+        /// <param name="routingKey">Routing key.</param>
+        /// <param name="queueService">Queue service.</param>
+        void Handle(string message, string routingKey, IQueueService queueService);
+    }
+}

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

@@ -32,6 +32,8 @@ internal class QueueService : IQueueService
 
         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 IDictionary<Type, List<string>> _routingKeys;
         readonly IEnumerable<RabbitMqExchange> _exchanges;
         readonly ILogger<QueueService> _logger;
@@ -46,6 +48,8 @@ internal class QueueService : IQueueService
         public QueueService(
             IEnumerable<IMessageHandler> messageHandlers,
             IEnumerable<IAsyncMessageHandler> asyncMessageHandlers,
+            IEnumerable<INonCyclicMessageHandler> nonCyclicHandlers,
+            IEnumerable<IAsyncNonCyclicMessageHandler> asyncnonCyclicHandlers,
             IEnumerable<RabbitMqExchange> exchanges,
             IEnumerable<MessageHandlerRouter> routers,
             ILoggerFactory loggerFactory,
@@ -58,7 +62,9 @@ public QueueService(
 
             _routingKeys = TransformMessageHandlerRouters(routers);
             _messageHandlers = TransformMessageHandlersCollection(messageHandlers);
-            _asyncMessageHandlers = TransformAsyncMessageHandlersCollection(asyncMessageHandlers);
+            _asyncMessageHandlers = TransformMessageHandlersCollection(asyncMessageHandlers);
+            _nonCyclicHandlers = TransformMessageHandlersCollection(nonCyclicHandlers);
+            _asyncNonCyclicHandlers = TransformMessageHandlersCollection(asyncnonCyclicHandlers);
 
             _logger = loggerFactory.CreateLogger<QueueService>();
 
@@ -278,9 +284,9 @@ IDictionary<Type, List<string>> TransformMessageHandlerRouters(IEnumerable<Messa
             return dictionary;
         }
 
-        IDictionary<string, IList<IMessageHandler>> TransformMessageHandlersCollection(IEnumerable<IMessageHandler> messageHandlers)
+        IDictionary<string, IList<T>> TransformMessageHandlersCollection<T>(IEnumerable<T> messageHandlers)
         {
-            var dictionary = new Dictionary<string, IList<IMessageHandler>>();
+            var dictionary = new Dictionary<string, IList<T>>();
             foreach (var handler in messageHandlers)
             {
                 var type = handler.GetType();
@@ -292,27 +298,7 @@ IDictionary<string, IList<IMessageHandler>> TransformMessageHandlersCollection(I
                             dictionary[routingKey].Add(handler);
                     }
                     else
-                        dictionary.Add(routingKey, new List<IMessageHandler>() { handler });
-                }
-            }
-            return dictionary;
-        }
-
-        IDictionary<string, IList<IAsyncMessageHandler>> TransformAsyncMessageHandlersCollection(IEnumerable<IAsyncMessageHandler> messageHandlers)
-        {
-            var dictionary = new Dictionary<string, IList<IAsyncMessageHandler>>();
-            foreach (var handler in messageHandlers)
-            {
-                var type = handler.GetType();
-                foreach (var routingKey in _routingKeys[type])
-                {
-                    if (dictionary.ContainsKey(routingKey))
-                    {
-                        if (!dictionary[routingKey].Any(x => x.GetType() == handler.GetType()))
-                            dictionary[routingKey].Add(handler);
-                    }
-                    else
-                        dictionary.Add(routingKey, new List<IAsyncMessageHandler>() { handler });
+                        dictionary.Add(routingKey, new List<T>() { handler });
                 }
             }
             return dictionary;
@@ -345,6 +331,22 @@ void StartClient()
                             _logger.LogDebug($"The message has been processed by message handler {handler?.GetType().Name}.");
                         }
                     }
+                    if (_asyncNonCyclicHandlers.ContainsKey(@event.RoutingKey))
+                    {
+                        var tasks = new List<Task>();
+                        foreach (var handler in _asyncNonCyclicHandlers[@event.RoutingKey])
+                            tasks.Add(RunAsyncNonCyclicHandler(handler, message, @event.RoutingKey));
+                        Task.WaitAll(tasks.ToArray());
+                    }
+                    if (_nonCyclicHandlers.ContainsKey(@event.RoutingKey))
+                    {
+                        foreach (var handler in _nonCyclicHandlers[@event.RoutingKey])
+                        {
+                            _logger.LogDebug($"Starting processing the message by non-cyclic message handler {handler?.GetType().Name}.");
+                            handler.Handle(message, @event.RoutingKey, this);
+                            _logger.LogDebug($"The message has been processed by non-cyclic message handler {handler?.GetType().Name}.");
+                        }
+                    }
                     _logger.LogInformation($"Success message with deliveryTag {@event.DeliveryTag}.");
                     Channel.BasicAck(@event.DeliveryTag, false);
                 }
@@ -396,6 +398,13 @@ async Task RunAsyncHandler(IAsyncMessageHandler handler, string message, string
             _logger.LogDebug($"The message has been processed by async message handler {handler?.GetType().Name}.");
         }
 
+        async Task RunAsyncNonCyclicHandler(IAsyncNonCyclicMessageHandler handler, string message, string routingKey)
+        {
+            _logger.LogDebug($"Starting processing the message by async non-cyclic message handler {handler?.GetType().Name}.");
+            await handler.Handle(message, routingKey, this);
+            _logger.LogDebug($"The message has been processed by async non-cyclic message handler {handler?.GetType().Name}.");
+        }
+
         void StartDeadLetterExchange(string exchangeName) =>
             _channel.ExchangeDeclare(
                 exchange: exchangeName,

src/RabbitMQ.Client.Core.DependencyInjection/RabbitMQ.Client.Core.DependencyInjection.csproj

@@ -3,7 +3,7 @@
   <PropertyGroup>
     <TargetFramework>netcoreapp2.2</TargetFramework>
     <LangVersion>latest</LangVersion>
-    <Version>1.2.3</Version>
+    <Version>1.3.0</Version>
     <PackageTags>RabbitMQ</PackageTags>
     <PackageLicenseFile>LICENSE.txt</PackageLicenseFile>
     <RepositoryUrl>https://github.com/AntonyVorontsov/RabbitMQ.Client.Core.DependencyInjection</RepositoryUrl>

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

@@ -142,8 +142,7 @@ public static IServiceCollection AddMessageHandlerSingleton<T>(this IServiceColl
             services.Add(new ServiceDescriptor(typeof(MessageHandlerRouter), router));
             return services;
         }
-
-
+        
         /// <summary>
         /// Add transient async message handler.
         /// </summary>
@@ -207,5 +206,133 @@ public static IServiceCollection AddAsyncMessageHandlerSingleton<T>(this IServic
             services.Add(new ServiceDescriptor(typeof(MessageHandlerRouter), router));
             return services;
         }
+
+        /// <summary>
+        /// Add transient non-cyclic message handler.
+        /// </summary>
+        /// <typeparam name="T">Message handler type.</typeparam>
+        /// <param name="services">Service collection.</param>
+        /// <param name="routingKey">Routing key.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddNonCyclicMessageHandlerTransient<T>(this IServiceCollection services, string routingKey)
+            where T : class, INonCyclicMessageHandler
+        {
+            services.AddTransient<INonCyclicMessageHandler, T>();
+            var router = new MessageHandlerRouter { Type = typeof(T), RoutingKeys = new[] { routingKey }.ToList() };
+            services.Add(new ServiceDescriptor(typeof(MessageHandlerRouter), router));
+            return services;
+        }
+
+        /// <summary>
+        /// Add transient non-cyclic message handler.
+        /// </summary>
+        /// <typeparam name="T">Message handler type.</typeparam>
+        /// <param name="services">Service collection.</param>
+        /// <param name="routingKeys">Routing keys.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddNonCyclicMessageHandlerTransient<T>(this IServiceCollection services, IEnumerable<string> routingKeys)
+            where T : class, INonCyclicMessageHandler
+        {
+            services.AddTransient<INonCyclicMessageHandler, T>();
+            var router = new MessageHandlerRouter { Type = typeof(T), RoutingKeys = routingKeys.ToList() };
+            services.Add(new ServiceDescriptor(typeof(MessageHandlerRouter), router));
+            return services;
+        }
+
+        /// <summary>
+        /// Add singleton non-cyclic message handler.
+        /// </summary>
+        /// <typeparam name="T">Message handler type.</typeparam>
+        /// <param name="services">Service collection.</param>
+        /// <param name="routingKey">Routing key.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddNonCyclicMessageHandlerSingleton<T>(this IServiceCollection services, string routingKey)
+            where T : class, INonCyclicMessageHandler
+        {
+            services.AddSingleton<INonCyclicMessageHandler, T>();
+            var router = new MessageHandlerRouter { Type = typeof(T), RoutingKeys = new[] { routingKey }.ToList() };
+            services.Add(new ServiceDescriptor(typeof(MessageHandlerRouter), router));
+            return services;
+        }
+
+        /// <summary>
+        /// Add singleton non-cyclic message handler.
+        /// </summary>
+        /// <typeparam name="T">Message handler type.</typeparam>
+        /// <param name="services">Service collection.</param>
+        /// <param name="routingKeys">Routing keys.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddNonCyclicMessageHandlerSingleton<T>(this IServiceCollection services, IEnumerable<string> routingKeys)
+            where T : class, INonCyclicMessageHandler
+        {
+            services.AddSingleton<INonCyclicMessageHandler, T>();
+            var router = new MessageHandlerRouter { Type = typeof(T), RoutingKeys = routingKeys.ToList() };
+            services.Add(new ServiceDescriptor(typeof(MessageHandlerRouter), router));
+            return services;
+        }
+
+        /// <summary>
+        /// Add transient async non-cyclic message handler.
+        /// </summary>
+        /// <typeparam name="T">Message handler type.</typeparam>
+        /// <param name="services">Service collection.</param>
+        /// <param name="routingKey">Routing key.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddAsyncNonCyclicMessageHandlerTransient<T>(this IServiceCollection services, string routingKey)
+            where T : class, IAsyncNonCyclicMessageHandler
+        {
+            services.AddTransient<IAsyncNonCyclicMessageHandler, T>();
+            var router = new MessageHandlerRouter { Type = typeof(T), RoutingKeys = new[] { routingKey }.ToList() };
+            services.Add(new ServiceDescriptor(typeof(MessageHandlerRouter), router));
+            return services;
+        }
+
+        /// <summary>
+        /// Add transient async non-cyclic message handler.
+        /// </summary>
+        /// <typeparam name="T">Message handler type.</typeparam>
+        /// <param name="services">Service collection.</param>
+        /// <param name="routingKeys">Routing keys.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddAsyncNonCyclicMessageHandlerTransient<T>(this IServiceCollection services, IEnumerable<string> routingKeys)
+            where T : class, IAsyncNonCyclicMessageHandler
+        {
+            services.AddTransient<IAsyncNonCyclicMessageHandler, T>();
+            var router = new MessageHandlerRouter { Type = typeof(T), RoutingKeys = routingKeys.ToList() };
+            services.Add(new ServiceDescriptor(typeof(MessageHandlerRouter), router));
+            return services;
+        }
+
+        /// <summary>
+        /// Add singleton async non-cyclic message handler.
+        /// </summary>
+        /// <typeparam name="T">Message handler type.</typeparam>
+        /// <param name="services">Service collection.</param>
+        /// <param name="routingKey">Routing key.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddAsyncNonCyclicMessageHandlerSingleton<T>(this IServiceCollection services, string routingKey)
+            where T : class, IAsyncNonCyclicMessageHandler
+        {
+            services.AddSingleton<IAsyncNonCyclicMessageHandler, T>();
+            var router = new MessageHandlerRouter { Type = typeof(T), RoutingKeys = new[] { routingKey }.ToList() };
+            services.Add(new ServiceDescriptor(typeof(MessageHandlerRouter), router));
+            return services;
+        }
+
+        /// <summary>
+        /// Add singleton async non-cyclic message handler.
+        /// </summary>
+        /// <typeparam name="T">Message handler type.</typeparam>
+        /// <param name="services">Service collection.</param>
+        /// <param name="routingKeys">Routing keys.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddAsyncNonCyclicMessageHandlerSingleton<T>(this IServiceCollection services, IEnumerable<string> routingKeys)
+            where T : class, IAsyncNonCyclicMessageHandler
+        {
+            services.AddSingleton<IAsyncNonCyclicMessageHandler, T>();
+            var router = new MessageHandlerRouter { Type = typeof(T), RoutingKeys = routingKeys.ToList() };
+            services.Add(new ServiceDescriptor(typeof(MessageHandlerRouter), router));
+            return services;
+        }
     }
 }