Allow number of queue consumers to be configurable

Author: ethanhann

src/InEngine.Core.Tests/Queue/Commands/PublishTest.cs

@@ -1,4 +1,5 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using System.Linq;
 using BeekmanLabs.UnitTesting;
 using InEngine.Core.Commands;
@@ -62,11 +63,11 @@ public void ShouldFailWhenCommandDoesNotExist()
         {
             Subject.CommandAssembly = "foo";
             Subject.CommandClass = "bar";
-            var expectedMessage = "Could not load plugin: " + Subject.CommandAssembly;
+            var expectedMessage = "Plugin not found at ";
 
             var result = Assert.Throws<PluginNotFoundException>(() => { Subject.Run(); });
 
-            Assert.AreEqual(expectedMessage, result.Message);
+            Assert.IsTrue(result.Message.StartsWith(expectedMessage, StringComparison.InvariantCulture));
         }
     }
 }

src/InEngine.Core/Queue/Jobs.cs

@@ -10,36 +10,17 @@ public class Jobs : IJobs
     {
         public void Schedule(IScheduler scheduler)
         {
-            ScheduleQueueConsumerJobs(scheduler);
-            ScheduleQueueConsumerJobs(scheduler, true);
-            //foreach (var index in Enumerable.Range(0, 8).ToList()) 
-            //{
-            //    var consume = new Consume();
-            //    consume.Name = consume.Name + index; 
-            //    var primaryQueueConsumer = consume.MakeJobBuilder().Build();
-            //    var secondaryQueueConsumer = consume.MakeJobBuilder().Build();
-            //    secondaryQueueConsumer.JobDataMap.Add("useSecondaryQueue", true);
-
-            //    var primaryTrigger = consume
-            //        .MakeTriggerBuilder()
-            //        .StartNow()
-            //        .WithSimpleSchedule(x => x.WithIntervalInSeconds(1).RepeatForever())
-            //        .Build();
-
-            //    var secondaryTrigger = consume
-            //        .MakeTriggerBuilder()
-            //        .StartNow()
-            //        .WithSimpleSchedule(x => x.WithIntervalInSeconds(1).RepeatForever())
-            //        .Build();
-
-            //    scheduler.ScheduleJob(primaryQueueConsumer, primaryTrigger);
-            //    scheduler.ScheduleJob(secondaryQueueConsumer, secondaryTrigger);
-            //}
+            var queueSettings = InEngineSettings.Make().Queue;
+            ScheduleQueueConsumerJobs(scheduler, queueSettings.PrimaryQueueConsumers);
+            ScheduleQueueConsumerJobs(scheduler, queueSettings.SecondaryQueueConsumers, true);
         }
 
-        private void ScheduleQueueConsumerJobs(IScheduler scheduler, bool useSecondaryQueue = false)
+        private void ScheduleQueueConsumerJobs(IScheduler scheduler, int consumers, bool useSecondaryQueue = false)
         {
-            foreach (var index in Enumerable.Range(0, 8).ToList())
+            if (consumers < 0) {
+                throw new ArgumentOutOfRangeException(nameof(consumers), consumers, "The number of queue consumers must be 0 or greater.");
+            }
+            foreach (var index in Enumerable.Range(0, consumers).ToList())
             {
                 var consume = new Consume() {
                     ScheduleId = $"{(useSecondaryQueue ? "secondary" : "primary")}:{index.ToString()}"

src/InEngine.Core/Queue/QueueSettings.cs

@@ -3,6 +3,8 @@ namespace InEngine.Core.Queue
 {
     public class QueueSettings
     {
+        public int PrimaryQueueConsumers { get; set; } = 8;
+        public int SecondaryQueueConsumers { get; set; } = 8;
         public string QueueName { get; set; }
         public string RedisHost { get; set; }
         public int RedisPort { get; set; }

src/InEngine/appsettings.json

@@ -1,6 +1,8 @@
 {
   "InEngine": {
     "Queue": {
+      "PrimaryQueueConsumers":  16,
+      "SecondaryQueueConsumers": -1,
       "QueueName": "InEngine:Queue",
       "RedisHost": "localhost",
       "RedisPort": 6379,