From b415ed63797927e59febe5b54798feb6349cc3b1 Mon Sep 17 00:00:00 2001 From: Halil ibrahim Kalkan Date: Mon, 17 Dec 2018 15:32:41 +0300 Subject: [PATCH] Implemented reconnect for RabbitMqMessageConsumer --- .../RabbitMq/RabbitMqDistributedEventBus.cs | 7 +- .../RabbitMqDistributedEventBusOptions.cs | 2 + .../Volo.Abp.RabbitMQ.csproj | 1 + .../Volo/Abp/RabbitMQ/AbpRabbitMqModule.cs | 4 +- .../IRabbitMqMessageConsumerFactory.cs | 9 + .../Abp/RabbitMQ/RabbitMqMessageConsumer.cs | 171 ++++++++++++------ .../RabbitMqMessageConsumerFactory.cs | 26 +-- 7 files changed, 151 insertions(+), 69 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBus.cs index a717cb8d5b..02a9823d84 100644 --- a/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBus.cs +++ b/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(IDistributedEventHandler 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, diff --git a/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBusOptions.cs b/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBusOptions.cs index fe277f8599..dbcf916b0c 100644 --- a/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBusOptions.cs +++ b/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; } diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo.Abp.RabbitMQ.csproj b/framework/src/Volo.Abp.RabbitMQ/Volo.Abp.RabbitMQ.csproj index 5173519237..7194a29ee3 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo.Abp.RabbitMQ.csproj +++ b/framework/src/Volo.Abp.RabbitMQ/Volo.Abp.RabbitMQ.csproj @@ -16,6 +16,7 @@ + diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqModule.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqModule.cs index ffaa0c00a7..b030e2f5c2 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqModule.cs +++ b/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 { diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumerFactory.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumerFactory.cs index e2a5125f68..6e5688b2e1 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumerFactory.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumerFactory.cs @@ -2,6 +2,15 @@ { public interface IRabbitMqMessageConsumerFactory { + /// + /// Creates a new . + /// Avoid to create too many consumers since they are + /// not disposed until end of the application. + /// + /// + /// + /// + /// IRabbitMqMessageConsumer Create( ExchangeDeclareConfiguration exchange, QueueDeclareConfiguration queue, diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs index be5358ed45..4c749fe365 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs @@ -1,32 +1,56 @@ -using System; -using System.Collections.Concurrent; -using System.Threading.Tasks; +using JetBrains.Annotations; +using Microsoft.Extensions.Logging; +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 Logger { get; set; } + protected IConnectionPool ConnectionPool { get; } - protected ExchangeDeclareConfiguration Exchange { get; } - protected QueueDeclareConfiguration Queue { get; } - protected string ConnectionName { get; } - protected IModel ConsumerChannel { get; set; } + + protected AbpTimer Timer { get; } + + protected ExchangeDeclareConfiguration Exchange { get; private set; } + + protected QueueDeclareConfiguration Queue { get; private set; } + + protected string ConnectionName { get; private set; } + protected ConcurrentBag> Callbacks { get; } + protected IModel Channel { get; private set; } + public RabbitMqMessageConsumer( IConnectionPool connectionPool, - ExchangeDeclareConfiguration exchange, - QueueDeclareConfiguration queue, - string connectionName = null) + AbpTimer timer) { ConnectionPool = connectionPool; - Exchange = exchange; - Queue = queue; - ConnectionName = connectionName; + Timer = timer; + Logger = NullLogger.Instance; + Callbacks = new ConcurrentBag>(); - 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 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 - .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) => + if (Channel == null || Channel.IsOpen == false) { + 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 callback(channel, basicDeliverEventArgs); - } + await HandleIncomingMessage(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( - queue: Queue.QueueName, - autoAck: false, - consumer: consumer - ); + channel.BasicAck(basicDeliverEventArgs.DeliveryTag, multiple: false); + } + catch (Exception ex) + { + Logger.LogException(ex); + } + } - //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! + protected virtual void DisposeChannel() + { + if (Channel == null) + { + return; + } - channel.CallbackException += (sender, ea) => + try + { + Channel.Dispose(); + } + catch (Exception ex) { - ConsumerChannel.Dispose(); - ConsumerChannel = CreateConsumerChannel(); - }; + Logger.LogException(ex, LogLevel.Warning); + } + } - return channel; + public virtual void Dispose() + { + DisposeChannel(); } } } diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumerFactory.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumerFactory.cs index 1787633ac3..b4abd19860 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumerFactory.cs +++ b/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( - ConnectionPool, - exchange, - queue, - connectionName - ); + var consumer = ServiceScope.ServiceProvider.GetRequiredService(); + consumer.Initialize(exchange, queue, connectionName); + return consumer; + } + + public void Dispose() + { + ServiceScope?.Dispose(); } } } \ No newline at end of file