Wireup scheduler to consume from secondary queue

Author: ethanhann

src/InEngine.Core/AbstractCommand.cs

@@ -45,5 +45,13 @@ public void Execute(IJobExecutionContext context)
             JobExecutionContext = context;
             Run();
         }
+
+        public T GetJobContextData<T>(string key)
+        {
+            if (JobExecutionContext == null || JobExecutionContext.MergedJobDataMap == null)
+                return default(T);
+            var objectVal = JobExecutionContext.MergedJobDataMap.Get(key);
+            return  objectVal == null ? default(T) : (T)objectVal;
+        }
     }
 }

src/InEngine.Core/Queue/Commands/Consume.cs

@@ -6,17 +6,18 @@ namespace InEngine.Core.Queue.Commands
     public class Consume : AbstractCommand
     {
         [Option("all", DefaultValue = false)]
-        public bool All { get; set; }
+        public bool ShouldConsumeAll { get; set; }
 
         [Option("secondary", DefaultValue = false, HelpText = "Consume from a secondary queue.")]
         public bool UseSecondaryQueue { get; set; }
 
         public override CommandResult Run()
         {
+            UseSecondaryQueue = UseSecondaryQueue || GetJobContextData<bool>("useSecondaryQueue");
             var broker = Broker.Make();
             var shouldConsume = true;
             while (shouldConsume)
-                shouldConsume = broker.Consume(UseSecondaryQueue) && All;
+                shouldConsume = broker.Consume(UseSecondaryQueue) && ShouldConsumeAll;
             return new CommandResult(true, "Consumed");
         }
     }

src/InEngine.Core/Queue/Jobs.cs

@@ -11,35 +11,35 @@ public class Jobs : IJobs
         public void Schedule(IScheduler scheduler)
         {
             var consume = new Consume();
-            foreach (var index in Enumerable.Range(0, 16).ToList()) 
+            foreach (var index in Enumerable.Range(0, 8).ToList()) 
             {
-                var defaultQueueConsumer = JobBuilder
+                var primaryQueueConsumer = JobBuilder
                     .Create<Consume>()
-                    .WithIdentity(consume.Name + index, "defaultQueueConsumer")
+                    .WithIdentity(consume.Name + index, "primaryQueueConsumer")
                     .Build();
 
-                var expressQueueConsumer = JobBuilder
+                var secondaryQueueConsumer = JobBuilder
                     .Create<Consume>()
-                    .WithIdentity(consume.Name + index, "expressQueueConsumer")
+                    .WithIdentity(consume.Name + index, "secondaryQueueConsumer")
                     .Build();
+                secondaryQueueConsumer.JobDataMap.Add("useSecondaryQueue", true);
 
-
-                var defaultTrigger = TriggerBuilder
+                var primaryTrigger = TriggerBuilder
                     .Create()
-                    .WithIdentity($"{consume.Name}-default-{index}", "queue")
+                    .WithIdentity($"{consume.Name}-primary-{index}", "queue")
                     .StartNow()
                     .WithSimpleSchedule(x => x.WithIntervalInSeconds(1).RepeatForever())
                     .Build();
-                
-                var expressTrigger = TriggerBuilder
+
+                var secondaryTrigger = TriggerBuilder
                     .Create()
-                    .WithIdentity($"{consume.Name}-express-{index}", "queue")
+                    .WithIdentity($"{consume.Name}-secondary-{index}", "queue")
                     .StartNow()
                     .WithSimpleSchedule(x => x.WithIntervalInSeconds(1).RepeatForever())
                     .Build();
 
-                scheduler.ScheduleJob(defaultQueueConsumer, defaultTrigger);
-                scheduler.ScheduleJob(expressQueueConsumer, expressTrigger);
+                scheduler.ScheduleJob(primaryQueueConsumer, primaryTrigger);
+                scheduler.ScheduleJob(secondaryQueueConsumer, secondaryTrigger);
             }
         }
     }