Implemented reconnect for RabbitMqMessageConsumer

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

@@ -62,7 +62,8 @@ namespace Volo.Abp.EventBus.Distributed.RabbitMq
                     durable: true,
                     exclusive: false,
                     autoDelete: false
-                    )
+                    ),
+                RabbitMqDistributedEventBusOptions.ConnectionName
             );
 
             Consumer.OnMessageReceived(ProcessEventAsync);
@@ -101,8 +102,6 @@ namespace Volo.Abp.EventBus.Distributed.RabbitMq
             var eventData = Serializer.Deserialize(ea.Body, eventType);
 
             await TriggerHandlersAsync(eventType, eventData);
-
-            channel.BasicAck(ea.DeliveryTag, multiple: false);
         }
 
         public IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler) where TEvent : class
@@ -120,7 +119,7 @@ namespace Volo.Abp.EventBus.Distributed.RabbitMq
             {
                 var eventName = EventNameAttribute.GetNameOrDefault(eventType);
 
-                using (var channel = ConnectionPool.Get().CreateModel()) //TODO: Connection name per event!
+                using (var channel = ConnectionPool.Get(RabbitMqDistributedEventBusOptions.ConnectionName).CreateModel())
                 {
                     channel.QueueBind(
                         queue: RabbitMqDistributedEventBusOptions.ClientName,

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

@@ -2,6 +2,8 @@
 {
     public class RabbitMqDistributedEventBusOptions
     {
+        public string ConnectionName { get; set; }
+
         public string ClientName { get; set; }
 
         public string ExchangeName { get; set; }

framework/src/Volo.Abp.RabbitMQ/Volo.Abp.RabbitMQ.csproj

@@ -16,6 +16,7 @@
   <ItemGroup>
     <PackageReference Include="RabbitMQ.Client" Version="5.1.0" />
     <ProjectReference Include="..\Volo.Abp.Json\Volo.Abp.Json.csproj" />
+    <ProjectReference Include="..\Volo.Abp.Threading\Volo.Abp.Threading.csproj" />
   </ItemGroup>
 
 </Project>

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

@@ -1,11 +1,13 @@
 using Microsoft.Extensions.DependencyInjection;
 using Volo.Abp.Json;
 using Volo.Abp.Modularity;
+using Volo.Abp.Threading;
 
 namespace Volo.Abp.RabbitMQ
 {
     [DependsOn(
-        typeof(AbpJsonModule)
+        typeof(AbpJsonModule),
+        typeof(AbpThreadingModule)
         )]
     public class AbpRabbitMqModule : AbpModule
     {

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

@@ -2,6 +2,15 @@
 {
     public interface IRabbitMqMessageConsumerFactory
     {
+        /// <summary>
+        /// Creates a new <see cref="IRabbitMqMessageConsumer"/>.
+        /// Avoid to create too many consumers since they are
+        /// not disposed until end of the application.
+        /// </summary>
+        /// <param name="exchange"></param>
+        /// <param name="queue"></param>
+        /// <param name="connectionName"></param>
+        /// <returns></returns>
         IRabbitMqMessageConsumer Create(
             ExchangeDeclareConfiguration exchange,
             QueueDeclareConfiguration queue,

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

@@ -1,32 +1,56 @@
-using System;
+using JetBrains.Annotations;
-using System.Collections.Concurrent;
+using Microsoft.Extensions.Logging;
-using System.Threading.Tasks;
+using Microsoft.Extensions.Logging.Abstractions;
 using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
+using System;
+using System.Collections.Concurrent;
+using System.Threading.Tasks;
+using Volo.Abp.DependencyInjection;
+using Volo.Abp.Threading;
 
 namespace Volo.Abp.RabbitMQ
 {
-    public class RabbitMqMessageConsumer : IRabbitMqMessageConsumer
+    public class RabbitMqMessageConsumer : IRabbitMqMessageConsumer, ITransientDependency, IDisposable
     {
+        public ILogger<RabbitMqMessageConsumer> Logger { get; set; }
+
         protected IConnectionPool ConnectionPool { get; }
-        protected ExchangeDeclareConfiguration Exchange { get; }
+
-        protected QueueDeclareConfiguration Queue { get; }
+        protected AbpTimer Timer { get; }
-        protected string ConnectionName { get; }
+
-        protected IModel ConsumerChannel { get; set; }
+        protected ExchangeDeclareConfiguration Exchange { get; private set; }
+
+        protected QueueDeclareConfiguration Queue { get; private set; }
+
+        protected string ConnectionName { get; private set; }
+
         protected ConcurrentBag<Func<IModel, BasicDeliverEventArgs, Task>> Callbacks { get; }
 
+        protected IModel Channel { get; private set; }
+
         public RabbitMqMessageConsumer(
             IConnectionPool connectionPool,
-            ExchangeDeclareConfiguration exchange,
+            AbpTimer timer)
-            QueueDeclareConfiguration queue,
-            string connectionName = null)
         {
             ConnectionPool = connectionPool;
-            Exchange = exchange;
+            Timer = timer;
-            Queue = queue;
+            Logger = NullLogger<RabbitMqMessageConsumer>.Instance;
-            ConnectionName = connectionName;
+
             Callbacks = new ConcurrentBag<Func<IModel, BasicDeliverEventArgs, Task>>();
-            ConsumerChannel = CreateConsumerChannel();
+
+            Timer.Period = 5000; //5 sec.
+            Timer.Elapsed += Timer_Elapsed;
+        }
+
+        public void Initialize(
+            [NotNull] ExchangeDeclareConfiguration exchange,
+            [NotNull] QueueDeclareConfiguration queue,
+            string connectionName = null)
+        {
+            Exchange = Check.NotNull(exchange, nameof(exchange));
+            Queue = Check.NotNull(queue, nameof(queue));
+            ConnectionName = connectionName;
         }
 
         public void OnMessageReceived(Func<IModel, BasicDeliverEventArgs, Task> callback)
@@ -34,53 +58,94 @@ namespace Volo.Abp.RabbitMQ
             Callbacks.Add(callback);
         }
 
-        private IModel CreateConsumerChannel()
+        protected virtual void Timer_Elapsed(object sender, EventArgs e)
         {
-            var channel = ConnectionPool
+            if (Channel == null || Channel.IsOpen == false)
-                .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) =>
             {
+                TryCreateChannel();
+            }
+        }
+
+        private void TryCreateChannel()
+        {
+            DisposeChannel();
+
+            try
+            {
+                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 HandleIncomingMessage(channel, basicDeliverEventArgs);
-                    {
+                };
-                        await callback(channel, basicDeliverEventArgs);
+
-                    }
+                channel.BasicConsume(
+                    queue: Queue.QueueName,
+                    autoAck: false,
+                    consumer: consumer
+                );
+
+                Channel = channel;
+            }
+            catch (Exception ex)
+            {
+                Logger.LogException(ex, LogLevel.Warning);
+            }
+        }
+
+        protected virtual async Task HandleIncomingMessage(IModel channel, BasicDeliverEventArgs basicDeliverEventArgs)
+        {
+            try
+            {
+                foreach (var callback in Callbacks)
+                {
+                    await callback(channel, basicDeliverEventArgs);
                 }
-            };
 
-            channel.BasicConsume(
+                channel.BasicAck(basicDeliverEventArgs.DeliveryTag, multiple: false);
-                queue: Queue.QueueName,
+            }
-                autoAck: false,
+            catch (Exception ex)
-                consumer: consumer
+            {
-            );
+                Logger.LogException(ex);
+            }
+        }
 
-            //TODO: Instead of creating a new customer immediately without exception handling,
+        protected virtual void DisposeChannel()
-            //create a timer that constantly checks connection health and re-connect if needed.
+        {
-            //As similar. Do not connect on constructor!
+            if (Channel == null)
+            {
+                return;
+            }
 
-            channel.CallbackException += (sender, ea) =>
+            try
+            {
+                Channel.Dispose();
+            }
+            catch (Exception ex)
             {
-                ConsumerChannel.Dispose();
+                Logger.LogException(ex, LogLevel.Warning);
-                ConsumerChannel = CreateConsumerChannel();
+            }
-            };
+        }
 
-            return channel;
+        public virtual void Dispose()
+        {
+            DisposeChannel();
         }
     }
 }

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

@@ -1,14 +1,16 @@
-using Volo.Abp.DependencyInjection;
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using Volo.Abp.DependencyInjection;
 
 namespace Volo.Abp.RabbitMQ
 {
-    public class RabbitMqMessageConsumerFactory : IRabbitMqMessageConsumerFactory, ISingletonDependency
+    public class RabbitMqMessageConsumerFactory : IRabbitMqMessageConsumerFactory, ISingletonDependency, IDisposable
     {
-        protected IConnectionPool ConnectionPool { get; }
+        protected IServiceScope ServiceScope { get; }
 
-        public RabbitMqMessageConsumerFactory(IConnectionPool connectionPool)
+        public RabbitMqMessageConsumerFactory(IServiceScopeFactory serviceScopeFactory)
         {
-            ConnectionPool = connectionPool;
+            ServiceScope = serviceScopeFactory.CreateScope();
         }
 
         public IRabbitMqMessageConsumer Create(
@@ -16,12 +18,14 @@ namespace Volo.Abp.RabbitMQ
             QueueDeclareConfiguration queue,
             string connectionName = null)
         {
-            return new RabbitMqMessageConsumer(
+            var consumer = ServiceScope.ServiceProvider.GetRequiredService<RabbitMqMessageConsumer>();
-                ConnectionPool,
+            consumer.Initialize(exchange, queue, connectionName);
-                exchange,
+            return consumer;
-                queue,
+        }
-                connectionName
+
-            );
+        public void Dispose()
+        {
+            ServiceScope?.Dispose();
         }
     }
 }