Apply settings to threaded dequeuer

Author: ethanhann

src/InEngine.Core/Queuing/Dequeue.cs

@@ -6,11 +6,11 @@
 
 namespace InEngine.Core.Queuing
 {
-    public class Dequeue
+    public class Dequeue : IDisposable
     {
         IList<QueueAdapter> queueAdapters;
         public CancellationTokenSource CancellationTokenSource { get; set; }
-        public int TaskCount { get; set; }
+        public QueueSettings QueueSettings { get; set; }
 
         public Dequeue()
         {
@@ -20,25 +20,28 @@ public Dequeue()
 
         public async Task StartAsync()
         {
+            if (QueueSettings == null)
+                QueueSettings = InEngineSettings.Make().Queue;
+            
             // Create dequeue tasks for primary and secondary queues.
             var allTasks = new List<Task>();
             Console.WriteLine("Start dequeue tasks for primary queue...");
-            allTasks.AddRange(MakeTasks());
+            allTasks.AddRange(MakeTasks(true, QueueSettings.PrimaryQueueConsumers));
             Console.WriteLine("Start dequeue tasks for secondary queue...");
-            allTasks.AddRange(MakeTasks());
+            allTasks.AddRange(MakeTasks(false, QueueSettings.SecondaryQueueConsumers));
             await Task.WhenAll(allTasks);
 
             // Recover from restart, if necessary.
             QueueAdapter.Make().Recover();
             QueueAdapter.Make(true).Recover();
         }
 
-        public IList<Task> MakeTasks()
+        IList<Task> MakeTasks(bool useSecondaryQueue = false, int numberOfTasks = 0)
         {
-            return Enumerable.Range(0, TaskCount).Select((i) => {
+            return Enumerable.Range(0, numberOfTasks).Select((i) => {
                 Console.WriteLine($"Registering Dequeuer #{i}");
                 return Task.Factory.StartNew(() => {
-                    var queue = QueueAdapter.Make();
+                    var queue = QueueAdapter.Make(useSecondaryQueue, QueueSettings);
                     queue.Id = i;
                     queueAdapters.Add(queue);
                     queue.Consume(CancellationTokenSource.Token);

src/InEngine.Core/Queuing/QueueAdapter.cs

@@ -16,9 +16,10 @@ public class QueueAdapter : IQueueClient
         public string QueueName { get => QueueClient.QueueName; set => QueueClient.QueueName = value; }
         public bool UseCompression { get => QueueClient.UseCompression; set => QueueClient.UseCompression = value; }
 
-        public static QueueAdapter Make(bool useSecondaryQueue = false)
+        public static QueueAdapter Make(bool useSecondaryQueue = false, QueueSettings queueSettings = null)
         {
-            var queueSettings = InEngineSettings.Make().Queue;
+            if (queueSettings == null)
+                queueSettings = InEngineSettings.Make().Queue;
             var queueDriverName = queueSettings.QueueDriver.ToLower();
             var queue = new QueueAdapter();
 

src/InEngine/ServerHost.cs

@@ -9,16 +9,14 @@ namespace InEngine
     public class ServerHost : IDisposable
     {
         public SuperScheduler SuperScheduler { get; set; }
-        public Dequeue ConsumeServer { get; set; }
+        public Dequeue Dequeue { get; set; }
 
         public ServerHost()
         {
             SuperScheduler = new SuperScheduler();
             Common.Logging.LogManager.Adapter = new Common.Logging.Simple.ConsoleOutLoggerFactoryAdapter { Level = Common.Logging.LogLevel.Info };
             SuperScheduler.Initialize();
-            ConsumeServer = new Dequeue() {
-                TaskCount = 10
-            };
+            Dequeue = new Dequeue();
         }
 
         public void Dispose()
@@ -34,7 +32,7 @@ public void Start()
 
         public async void StartQueueServerAsync()
         {
-            await ConsumeServer.StartAsync();
+            await Dequeue.StartAsync();
         }
     }
 }