From 1ed1c994c7d01a56e8ecda55eb521aca9403974f Mon Sep 17 00:00:00 2001 From: Halil ibrahim Kalkan Date: Mon, 17 Dec 2018 15:59:04 +0300 Subject: [PATCH] Added BindAsync and UnbindAsync methods to IRabbitMqMessageConsumer --- .../RabbitMq/RabbitMqDistributedEventBus.cs | 11 +-- .../Abp/RabbitMQ/IRabbitMqMessageConsumer.cs | 4 + .../Volo/Abp/RabbitMQ/QueueBindType.cs | 4 + .../Abp/RabbitMQ/RabbitMqMessageConsumer.cs | 85 +++++++++++++++++++ 4 files changed, 95 insertions(+), 9 deletions(-) create mode 100644 framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueBindType.cs 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 02a9823d84..2007a69bf5 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 @@ -119,14 +119,7 @@ namespace Volo.Abp.EventBus.Distributed.RabbitMq { var eventName = EventNameAttribute.GetNameOrDefault(eventType); - using (var channel = ConnectionPool.Get(RabbitMqDistributedEventBusOptions.ConnectionName).CreateModel()) - { - channel.QueueBind( - queue: RabbitMqDistributedEventBusOptions.ClientName, - exchange: RabbitMqDistributedEventBusOptions.ExchangeName, - routingKey: eventName - ); - } + Consumer.BindAsync(eventName); } return new EventHandlerFactoryUnregistrar(this, eventType, factory); @@ -191,7 +184,7 @@ namespace Volo.Abp.EventBus.Distributed.RabbitMq var eventName = EventNameAttribute.GetNameOrDefault(eventType); var body = Serializer.Serialize(eventData); - using (var channel = ConnectionPool.Get().CreateModel()) //TODO: Connection name per event! + using (var channel = ConnectionPool.Get(RabbitMqDistributedEventBusOptions.ConnectionName).CreateModel()) { //TODO: Other properties like durable? channel.ExchangeDeclare(RabbitMqDistributedEventBusOptions.ExchangeName, ""); diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumer.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumer.cs index 097c5cec35..d600677ec7 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumer.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqMessageConsumer.cs @@ -7,6 +7,10 @@ namespace Volo.Abp.RabbitMQ { public interface IRabbitMqMessageConsumer { + Task BindAsync(string routingKey); + + Task UnbindAsync(string routingKey); + void OnMessageReceived(Func callback); } } \ No newline at end of file diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueBindType.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueBindType.cs new file mode 100644 index 0000000000..2ff06260a6 --- /dev/null +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueBindType.cs @@ -0,0 +1,4 @@ +namespace Volo.Abp.RabbitMQ +{ + +} \ No newline at end of file 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 4c749fe365..25ba3a2aa4 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs @@ -29,6 +29,10 @@ namespace Volo.Abp.RabbitMQ protected IModel Channel { get; private set; } + protected ConcurrentQueue QueueBindCommands { get; } + + protected object ChannelSendSyncLock { get; } = new object(); + public RabbitMqMessageConsumer( IConnectionPool connectionPool, AbpTimer timer) @@ -37,6 +41,7 @@ namespace Volo.Abp.RabbitMQ Timer = timer; Logger = NullLogger.Instance; + QueueBindCommands = new ConcurrentQueue(); Callbacks = new ConcurrentBag>(); Timer.Period = 5000; //5 sec. @@ -51,6 +56,66 @@ namespace Volo.Abp.RabbitMQ Exchange = Check.NotNull(exchange, nameof(exchange)); Queue = Check.NotNull(queue, nameof(queue)); ConnectionName = connectionName; + Timer.Start(); + } + + public async Task BindAsync(string routingKey) + { + QueueBindCommands.Enqueue(new QueueBindCommand(QueueBindType.Bind, routingKey)); + await TrySendQueueBindCommandsAsync(); + } + + public async Task UnbindAsync(string routingKey) + { + QueueBindCommands.Enqueue(new QueueBindCommand(QueueBindType.Unbind, routingKey)); + await TrySendQueueBindCommandsAsync(); + } + + protected Task TrySendQueueBindCommandsAsync() + { + try + { + while (!QueueBindCommands.IsEmpty) + { + if (Channel == null || Channel.IsClosed) + { + return Task.CompletedTask; + } + + lock (ChannelSendSyncLock) + { + QueueBindCommands.TryPeek(out var command); + + switch (command.Type) + { + case QueueBindType.Bind: + Channel.QueueBind( + queue: Queue.QueueName, + exchange: Exchange.ExchangeName, + routingKey: command.RoutingKey + ); + break; + case QueueBindType.Unbind: + Channel.QueueUnbind( + queue: Queue.QueueName, + exchange: Exchange.ExchangeName, + routingKey: command.RoutingKey + ); + break; + default: + throw new AbpException($"Unknown {nameof(QueueBindType)}: {command.Type}"); + } + + QueueBindCommands.TryDequeue(out command); + } + } + } + catch (Exception ex) + { + Logger.LogException(ex, LogLevel.Warning); + } + + return Task.CompletedTask; } public void OnMessageReceived(Func callback) @@ -63,6 +128,7 @@ namespace Volo.Abp.RabbitMQ if (Channel == null || Channel.IsOpen == false) { TryCreateChannel(); + AsyncHelper.RunSync(TrySendQueueBindCommandsAsync); } } @@ -147,5 +213,24 @@ namespace Volo.Abp.RabbitMQ { DisposeChannel(); } + + public class QueueBindCommand + { + public QueueBindType Type { get; } + + public string RoutingKey { get; } + + public QueueBindCommand(QueueBindType type, string routingKey) + { + Type = type; + RoutingKey = routingKey; + } + } + + public enum QueueBindType + { + Bind, + Unbind + } } }