Fixed #9: Allow subscribing to queues before consuming

Tapeti/Config/IConfig.cs

@@ -24,6 +24,12 @@ namespace Tapeti.Config
     }
 
 
+    public interface IDynamicQueue : IQueue
+    {
+        void SetName(string name);
+    }
+
+
     public interface IBinding
     {
         Type Controller { get; }

Tapeti/Connection/TapetiSubscriber.cs

@@ -9,17 +9,30 @@ namespace Tapeti.Connection
     public class TapetiSubscriber : ISubscriber
     {
         private readonly Func<TapetiWorker> workerFactory;
+        private readonly List<IQueue> queues;
+        private bool consuming;
 
 
-        public TapetiSubscriber(Func<TapetiWorker> workerFactory)
+        public TapetiSubscriber(Func<TapetiWorker> workerFactory, IEnumerable<IQueue> queues)
         {
             this.workerFactory = workerFactory;
+            this.queues = queues.ToList();
         }
 
 
-        public async Task BindQueues(IEnumerable<IQueue> queues)
+        public Task BindQueues()
+        {           
+            return Task.WhenAll(queues.Select(queue => workerFactory().Subscribe(queue)).ToList());
+        }
+
+
+        public Task Resume()
         {
-            await Task.WhenAll(queues.Select(queue => workerFactory().Subscribe(queue)).ToList());
+            if (consuming)
+                return Task.CompletedTask;
+
+            consuming = true;
+            return Task.WhenAll(queues.Select(queue => workerFactory().Consume(queue.Name, queue.Bindings)).ToList());
         }
     }
 }

Tapeti/Connection/TapetiWorker.cs

@@ -48,6 +48,9 @@ namespace Tapeti.Connection
 
         public Task Consume(string queueName, IEnumerable<IBinding> bindings)
         {
+            if (string.IsNullOrEmpty(queueName))
+                throw new ArgumentNullException(nameof(queueName));
+
             return taskQueue.Value.Add(async () =>
             {
                 (await GetChannel()).BasicConsume(queueName, false, new TapetiConsumer(this, queueName, dependencyResolver, bindings, messageMiddleware));
@@ -55,32 +58,29 @@ namespace Tapeti.Connection
         }
 
 
-        public async Task Subscribe(IQueue queue)
+        public Task Subscribe(IQueue queue)
         {
-            var queueName = await taskQueue.Value.Add(async () =>
+            return taskQueue.Value.Add(async () =>
             {
                 var channel = await GetChannel();
 
                 if (queue.Dynamic)
                 {
                     var dynamicQueue = channel.QueueDeclare();
+                    (queue as IDynamicQueue)?.SetName(dynamicQueue.QueueName);
 
                     foreach (var binding in queue.Bindings)
                     {
                         var routingKey = routingKeyStrategy.GetRoutingKey(binding.MessageClass);
-                        channel.QueueBind(dynamicQueue.QueueName, exchangeStrategy.GetExchange(binding.MessageClass), routingKey);
+                        var exchange = exchangeStrategy.GetExchange(binding.MessageClass);
 
+                        channel.QueueBind(dynamicQueue.QueueName, exchange, routingKey);
                         (binding as IDynamicQueueBinding)?.SetQueueName(dynamicQueue.QueueName);
                     }
-                    
-                    return dynamicQueue.QueueName;
                 }
-
+                else
-                channel.QueueDeclarePassive(queue.Name);
+                    channel.QueueDeclarePassive(queue.Name);
-                return queue.Name;
             }).Unwrap();
-
-            await Consume(queueName, queue.Bindings);
         }
 
 

Tapeti/ISubscriber.cs

@@ -1,6 +1,9 @@
-namespace Tapeti
+using System.Threading.Tasks;
+
+namespace Tapeti
 {
     public interface ISubscriber
     {
+        Task Resume();
     }
 }

Tapeti/TapetiConfig.cs

@@ -342,10 +342,10 @@ namespace Tapeti
         }
 
 
-        protected class Queue : IQueue
+        protected class Queue : IDynamicQueue
         {
             public bool Dynamic { get; }
-            public string Name { get; }
+            public string Name { get; set;  }
             public IEnumerable<IBinding> Bindings { get; }
 
 
@@ -355,6 +355,12 @@ namespace Tapeti
                 Name = queue.Name;
                 Bindings = bindings;
             }
+
+
+            public void SetName(string name)
+            {
+                Name = name;
+            }
         }
 
 

Tapeti/TapetiConnection.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Linq;
 using System.Threading.Tasks;
 using Tapeti.Config;
 using Tapeti.Connection;
@@ -25,10 +26,13 @@ namespace Tapeti
         }
 
 
-        public async Task<ISubscriber> Subscribe()
+        public async Task<ISubscriber> Subscribe(bool startConsuming = true)
         {
-            var subscriber = new TapetiSubscriber(() => worker.Value);
+            var subscriber = new TapetiSubscriber(() => worker.Value, config.Queues.ToList());
-            await subscriber.BindQueues(config.Queues);
+            await subscriber.BindQueues();
+
+            if (startConsuming)
+                await subscriber.Resume();
 
             return subscriber;
         }