diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueConfiguration.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueConfiguration.cs index 91cd5b4a1b..959952de22 100644 --- a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueConfiguration.cs +++ b/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; } 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 99173ee333..a717cb8d5b 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 @@ -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> HandlerFactories { get; } //TODO: Accessing to the List may not be thread-safe! protected ConcurrentDictionary EventTypes { get; } - protected IModel ConsumerChannel; protected IHybridServiceScopeFactory ServiceScopeFactory { get; } + protected IRabbitMqMessageConsumerFactory MessageConsumerFactory { get; } + protected IRabbitMqMessageConsumer Consumer { get; } public RabbitMqDistributedEventBus( IOptions options, IConnectionPool connectionPool, IRabbitMqSerializer serializer, IHybridServiceScopeFactory serviceScopeFactory, - IOptions distributedEventBusOptions) + IOptions distributedEventBusOptions, + IRabbitMqMessageConsumerFactory messageConsumerFactory) { ConnectionPool = connectionPool; Serializer = serializer; ServiceScopeFactory = serviceScopeFactory; + MessageConsumerFactory = messageConsumerFactory; DistributedEventBusOptions = distributedEventBusOptions.Value; RabbitMqDistributedEventBusOptions = options.Value; HandlerFactories = new ConcurrentDictionary>(); EventTypes = new ConcurrentDictionary(); - 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 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; diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ExchangeDeclareConfiguration.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ExchangeDeclareConfiguration.cs new file mode 100644 index 0000000000..67f234aec4 --- /dev/null +++ b/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 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(); + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumer.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumer.cs new file mode 100644 index 0000000000..097c5cec35 --- /dev/null +++ b/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 callback); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumerFactory.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumerFactory.cs new file mode 100644 index 0000000000..e2a5125f68 --- /dev/null +++ b/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 + ); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueConfiguration.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs similarity index 92% rename from framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueConfiguration.cs rename to framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs index 3d74a6708e..8cc07b7bb9 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueConfiguration.cs +++ b/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 Arguments { get; } - public QueueConfiguration( + public QueueDeclareConfiguration( [NotNull] string queueName, bool durable = true, bool exclusive = false, diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs new file mode 100644 index 0000000000..be5358ed45 --- /dev/null +++ b/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> Callbacks { get; } + + public RabbitMqMessageConsumer( + IConnectionPool connectionPool, + ExchangeDeclareConfiguration exchange, + QueueDeclareConfiguration queue, + string connectionName = null) + { + ConnectionPool = connectionPool; + Exchange = exchange; + Queue = queue; + ConnectionName = connectionName; + Callbacks = new ConcurrentBag>(); + ConsumerChannel = CreateConsumerChannel(); + } + + public void OnMessageReceived(Func 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; + } + } +} diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumerFactory.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumerFactory.cs new file mode 100644 index 0000000000..1787633ac3 --- /dev/null +++ b/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 + ); + } + } +} \ No newline at end of file