Added BindAsync and UnbindAsync methods to IRabbitMqMessageConsumer

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, "");

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<IModel, BasicDeliverEventArgs, Task> callback);
     }
 }

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

@@ -0,0 +1,4 @@
+namespace Volo.Abp.RabbitMQ
+{
+
+}

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<QueueBindCommand> 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<RabbitMqMessageConsumer>.Instance;
 
+            QueueBindCommands = new ConcurrentQueue<QueueBindCommand>();
             Callbacks = new ConcurrentBag<Func<IModel, BasicDeliverEventArgs, Task>>();
 
             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<IModel, BasicDeliverEventArgs, Task> 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
+        }
     }
 }