Extract RabbitMqMessageConsumer.

framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueConfiguration.cs

@@ -3,7 +3,7 @@ using Volo.Abp.RabbitMQ;
 
 namespace Volo.Abp.BackgroundJobs.RabbitMQ
 {
-    public class JobQueueConfiguration : QueueConfiguration
+    public class JobQueueConfiguration : QueueDeclareConfiguration
     {
         public Type JobArgsType { get; }
 

framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBus.cs

@@ -10,7 +10,6 @@ using Volo.Abp.RabbitMQ;
 using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
 using Volo.Abp.Collections;
-using Volo.Abp.EventBus.Local;
 using Volo.Abp.Threading;
 
 namespace Volo.Abp.EventBus.Distributed.RabbitMq
@@ -29,27 +28,44 @@ namespace Volo.Abp.EventBus.Distributed.RabbitMq
         protected IRabbitMqSerializer Serializer { get; }
         protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; } //TODO: Accessing to the List<IEventHandlerFactory> may not be thread-safe!
         protected ConcurrentDictionary<string, Type> EventTypes { get; }
-        protected IModel ConsumerChannel;
         protected IHybridServiceScopeFactory ServiceScopeFactory { get; }
+        protected IRabbitMqMessageConsumerFactory MessageConsumerFactory { get; }
+        protected IRabbitMqMessageConsumer Consumer { get; }
 
         public RabbitMqDistributedEventBus(
             IOptions<RabbitMqDistributedEventBusOptions> options,
             IConnectionPool connectionPool,
             IRabbitMqSerializer serializer,
             IHybridServiceScopeFactory serviceScopeFactory, 
-            IOptions<DistributedEventBusOptions> distributedEventBusOptions)
+            IOptions<DistributedEventBusOptions> distributedEventBusOptions,
+            IRabbitMqMessageConsumerFactory messageConsumerFactory)
         {
             ConnectionPool = connectionPool;
             Serializer = serializer;
             ServiceScopeFactory = serviceScopeFactory;
+            MessageConsumerFactory = messageConsumerFactory;
             DistributedEventBusOptions = distributedEventBusOptions.Value;
             RabbitMqDistributedEventBusOptions = options.Value;
             
             HandlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>();
             EventTypes = new ConcurrentDictionary<string, Type>();
 
-            ConsumerChannel = CreateConsumerChannel();
             Subscribe(DistributedEventBusOptions.Handlers);
+
+            Consumer = MessageConsumerFactory.Create(
+                new ExchangeDeclareConfiguration(
+                    RabbitMqDistributedEventBusOptions.ExchangeName, 
+                    type: "direct"
+                    ),
+                new QueueDeclareConfiguration(
+                    RabbitMqDistributedEventBusOptions.ClientName,
+                    durable: true,
+                    exclusive: false,
+                    autoDelete: false
+                    )
+            );
+
+            Consumer.OnMessageReceived(ProcessEventAsync);
         }
 
         protected virtual void Subscribe(ITypeList<IEventHandler> handlers)
@@ -73,41 +89,6 @@ namespace Volo.Abp.EventBus.Distributed.RabbitMq
             }
         }
 
-        private IModel CreateConsumerChannel()
-        {
-            //TODO: Support multiple connection (and consumer)?
-            var channel = ConnectionPool.Get().CreateModel();
-            channel.ExchangeDeclare(
-                exchange: RabbitMqDistributedEventBusOptions.ExchangeName,
-                type: "direct"
-            );
-
-            channel.QueueDeclare(
-                queue: RabbitMqDistributedEventBusOptions.ClientName,
-                durable: true,
-                exclusive: false,
-                autoDelete: false,
-                arguments: null
-            );
-
-            var consumer = new EventingBasicConsumer(channel);
-            consumer.Received += async (model, ea) => { await ProcessEventAsync(channel, ea); };
-
-            channel.BasicConsume(
-                queue: RabbitMqDistributedEventBusOptions.ClientName,
-                autoAck: false,
-                consumer: consumer
-            );
-
-            channel.CallbackException += (sender, ea) =>
-            {
-                ConsumerChannel.Dispose();
-                ConsumerChannel = CreateConsumerChannel();
-            };
-
-            return channel;
-        }
-
         private async Task ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea)
         {
             var eventName = ea.RoutingKey;

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ExchangeDeclareConfiguration.cs

@@ -0,0 +1,30 @@
+using System.Collections.Generic;
+
+namespace Volo.Abp.RabbitMQ
+{
+    public class ExchangeDeclareConfiguration
+    {
+        public string ExchangeName { get; }
+
+        public string Type { get; }
+
+        public bool Durable { get; set; }
+
+        public bool AutoDelete { get; set; }
+
+        public IDictionary<string, object> Arguments { get; }
+
+        public ExchangeDeclareConfiguration(
+            string exchangeName, 
+            string type, 
+            bool durable = false, 
+            bool autoDelete = false)
+        {
+            ExchangeName = exchangeName;
+            Type = type;
+            Durable = durable;
+            AutoDelete = autoDelete;
+            Arguments = new Dictionary<string, object>();
+        }
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumer.cs

@@ -0,0 +1,12 @@
+using System;
+using System.Threading.Tasks;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+
+namespace Volo.Abp.RabbitMQ
+{
+    public interface IRabbitMqMessageConsumer
+    {
+        void OnMessageReceived(Func<IModel, BasicDeliverEventArgs, Task> callback);
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumerFactory.cs

@@ -0,0 +1,11 @@
+namespace Volo.Abp.RabbitMQ
+{
+    public interface IRabbitMqMessageConsumerFactory
+    {
+        IRabbitMqMessageConsumer Create(
+            ExchangeDeclareConfiguration exchange,
+            QueueDeclareConfiguration queue,
+            string connectionName = null
+        );
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueConfiguration.cs → framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs

@@ -4,7 +4,7 @@ using RabbitMQ.Client;
 
 namespace Volo.Abp.RabbitMQ
 {
-    public class QueueConfiguration
+    public class QueueDeclareConfiguration
     {
         [NotNull]
         public string QueueName { get; }
@@ -17,7 +17,7 @@ namespace Volo.Abp.RabbitMQ
 
         public IDictionary<string, object> Arguments { get; }
 
-        public QueueConfiguration(
+        public QueueDeclareConfiguration(
             [NotNull] string queueName, 
             bool durable = true, 
             bool exclusive = false, 

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs

@@ -0,0 +1,86 @@
+using System;
+using System.Collections.Concurrent;
+using System.Threading.Tasks;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+
+namespace Volo.Abp.RabbitMQ
+{
+    public class RabbitMqMessageConsumer : IRabbitMqMessageConsumer
+    {
+        protected IConnectionPool ConnectionPool { get; }
+        protected ExchangeDeclareConfiguration Exchange { get; }
+        protected QueueDeclareConfiguration Queue { get; }
+        protected string ConnectionName { get; }
+        protected IModel ConsumerChannel { get; set; }
+        protected ConcurrentBag<Func<IModel, BasicDeliverEventArgs, Task>> Callbacks { get; }
+
+        public RabbitMqMessageConsumer(
+            IConnectionPool connectionPool,
+            ExchangeDeclareConfiguration exchange,
+            QueueDeclareConfiguration queue,
+            string connectionName = null)
+        {
+            ConnectionPool = connectionPool;
+            Exchange = exchange;
+            Queue = queue;
+            ConnectionName = connectionName;
+            Callbacks = new ConcurrentBag<Func<IModel, BasicDeliverEventArgs, Task>>();
+            ConsumerChannel = CreateConsumerChannel();
+        }
+
+        public void OnMessageReceived(Func<IModel, BasicDeliverEventArgs, Task> callback)
+        {
+            Callbacks.Add(callback);
+        }
+
+        private IModel CreateConsumerChannel()
+        {
+            var channel = ConnectionPool
+                .Get(ConnectionName)
+                .CreateModel();
+
+            channel.ExchangeDeclare(
+                exchange: Exchange.ExchangeName,
+                type: Exchange.Type
+            );
+
+            channel.QueueDeclare(
+                queue: Queue.QueueName,
+                durable: Queue.Durable,
+                exclusive: Queue.Exclusive,
+                autoDelete: Queue.AutoDelete,
+                arguments: Queue.Arguments
+            );
+
+            var consumer = new EventingBasicConsumer(channel);
+            consumer.Received += async (model, basicDeliverEventArgs) =>
+            {
+                {
+                    foreach (var callback in Callbacks)
+                    {
+                        await callback(channel, basicDeliverEventArgs);
+                    }
+                }
+            };
+
+            channel.BasicConsume(
+                queue: Queue.QueueName,
+                autoAck: false,
+                consumer: consumer
+            );
+
+            //TODO: Instead of creating a new customer immediately without exception handling,
+            //create a timer that constantly checks connection health and re-connect if needed.
+            //As similar. Do not connect on constructor!
+
+            channel.CallbackException += (sender, ea) =>
+            {
+                ConsumerChannel.Dispose();
+                ConsumerChannel = CreateConsumerChannel();
+            };
+
+            return channel;
+        }
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumerFactory.cs

@@ -0,0 +1,27 @@
+using Volo.Abp.DependencyInjection;
+
+namespace Volo.Abp.RabbitMQ
+{
+    public class RabbitMqMessageConsumerFactory : IRabbitMqMessageConsumerFactory, ISingletonDependency
+    {
+        protected IConnectionPool ConnectionPool { get; }
+
+        public RabbitMqMessageConsumerFactory(IConnectionPool connectionPool)
+        {
+            ConnectionPool = connectionPool;
+        }
+
+        public IRabbitMqMessageConsumer Create(
+            ExchangeDeclareConfiguration exchange,
+            QueueDeclareConfiguration queue,
+            string connectionName = null)
+        {
+            return new RabbitMqMessageConsumer(
+                ConnectionPool,
+                exchange,
+                queue,
+                connectionName
+            );
+        }
+    }
+}