adding AbstractDispatchingMessageHandler to process messages in parallel

SimpleRabbit.NetCore.Dispatcher/AbstractDispatchingMessageHandler.cs

@@ -0,0 +1,95 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+
+namespace SimpleRabbit.NetCore.Dispatcher
+{
+    public abstract class AbstractDispatchingMessageHandler<T> : IMessageHandler, IChannelOptions
+    {
+        private readonly ILogger<AbstractDispatchingMessageHandler<T>> _logger;
+        private readonly Dictionary<string, List<ModelDetails<T>>> _queues = new Dictionary<string, List<ModelDetails<T>>>();
+        private readonly object _semaphore = new object();
+
+        private IModel _model;
+        private Action _onError;
+
+        protected AbstractDispatchingMessageHandler(ILogger<AbstractDispatchingMessageHandler<T>> logger)
+        {
+            _logger = logger;
+        }
+
+        public abstract bool CanProcess(string tag);
+        protected abstract void ProcessMessage(T msg);
+        public abstract T Get(BasicDeliverEventArgs args);
+        public abstract string GetKey(T msg);
+
+        public bool Process(BasicDeliverEventArgs args)
+        {
+            var msg = Get(args);
+            var key = GetKey(msg);
+            lock (_semaphore)
+            {
+                if (!_queues.TryGetValue(key, out var queue))
+                {
+                    queue = new List<ModelDetails<T>>();
+                    _queues.Add(key, queue);
+
+                    Task.Run(() => ProcessQueue(queue, key));
+                }
+                queue.Add(new ModelDetails<T>
+                {
+                    Args = args,
+                    Message = msg
+                });
+            }
+
+            return false;
+        }
+
+        private void ProcessQueue(List<ModelDetails<T>> queue, string key)
+        {
+            try
+            {
+                while (true)
+                {
+                    ModelDetails<T> details;
+                    lock (_semaphore)
+                    {
+                        if (queue.Count == 0)
+                        {
+                            _queues.Remove(key);
+                            return;
+                        }
+
+                        details = queue[0];
+                    }
+
+                    ProcessMessage(details.Message);
+
+                    lock (_semaphore)
+                    {
+                        queue.Remove(details);
+                    }
+
+                    _model?.BasicAck(details.Args.DeliveryTag, false);
+                }
+            }
+            catch (Exception e)
+            {
+                _logger.LogError(e, "An error occured while processing a message queue");
+                queue.Clear();
+                _queues.Remove(key);
+                _onError?.Invoke();
+            }
+        }
+
+        public void SetChannelOptions(IModel model, Action onError)
+        {
+            _model = model;
+            _onError = onError;
+        }
+    }
+}

SimpleRabbit.NetCore.Dispatcher/Model/ModelDetails.cs

@@ -0,0 +1,10 @@
+using RabbitMQ.Client.Events;
+
+namespace SimpleRabbit.NetCore.Dispatcher
+{
+    public class ModelDetails<T>
+    {
+        public T Message;
+        public BasicDeliverEventArgs Args;
+    }
+}

SimpleRabbit.NetCore.Dispatcher/SimpleRabbit.NetCore.Dispatcher.csproj

@@ -0,0 +1,11 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp2.1</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\SimpleRabbit.NetCore\SimpleRabbit.NetCore.csproj" />
+  </ItemGroup>
+
+</Project>

SimpleRabbit.NetCore.sln

@@ -5,7 +5,9 @@ VisualStudioVersion = 15.0.27703.2035
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "SimpleRabbit.NetCore", "SimpleRabbit.NetCore\SimpleRabbit.NetCore.csproj", "{005D39F6-9FA3-4321-8F50-1B5967EBADBE}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SimpleRabbit.NetCore.Service", "SimpleRabbit.NetCore.Service\SimpleRabbit.NetCore.Service.csproj", "{48551957-A3C9-4FF0-BDAE-66AA85103D48}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "SimpleRabbit.NetCore.Service", "SimpleRabbit.NetCore.Service\SimpleRabbit.NetCore.Service.csproj", "{48551957-A3C9-4FF0-BDAE-66AA85103D48}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SimpleRabbit.NetCore.Dispatcher", "SimpleRabbit.NetCore.Dispatcher\SimpleRabbit.NetCore.Dispatcher.csproj", "{B0F9E58D-8E5B-48DF-A905-DB3FE8952FF6}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -21,6 +23,10 @@ Global
 		{48551957-A3C9-4FF0-BDAE-66AA85103D48}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{48551957-A3C9-4FF0-BDAE-66AA85103D48}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{48551957-A3C9-4FF0-BDAE-66AA85103D48}.Release|Any CPU.Build.0 = Release|Any CPU
+		{B0F9E58D-8E5B-48DF-A905-DB3FE8952FF6}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{B0F9E58D-8E5B-48DF-A905-DB3FE8952FF6}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{B0F9E58D-8E5B-48DF-A905-DB3FE8952FF6}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{B0F9E58D-8E5B-48DF-A905-DB3FE8952FF6}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE