Refactor broker

Author: ethanhann

src/InEngine.Core/Queue/Broker.cs

@@ -10,12 +10,10 @@ namespace InEngine.Core.Queue
     public class Broker
     {
         public string QueueBaseName { get; set; } = "InEngine:Queue";
-        public string PrimaryWaitingQueueName { get { return QueueBaseName + ":Primary:Waiting"; } }
-        public string PrimaryProcessingQueueName { get { return QueueBaseName + ":Primary:Processing"; } }
-        public string PrimaryFailedQueueName { get { return QueueBaseName + ":Primary:Failed"; } }
-        public string SecondaryWaitingQueueName { get { return QueueBaseName + ":Secondary:Waiting"; } }
-        public string SecondaryProcessingQueueName { get { return QueueBaseName + ":Secondary:Processing"; } }
-        public string SecondaryFailedQueueName { get { return QueueBaseName + ":Secondary:Failed"; } }
+        public string QueueName { get; internal set; } = "Primary";
+        public string PendingQueueName { get { return QueueBaseName + $":{QueueName}:Pending"; } }
+        public string InProgressQueueName { get { return QueueBaseName + $":{QueueName}:InProgress"; } }
+        public string FailedQueueName { get { return QueueBaseName + $":{QueueName}:Failed"; } }
         public static Lazy<ConnectionMultiplexer> lazyConnection = new Lazy<ConnectionMultiplexer>(() => { 
             var queueSettings = InEngineSettings.Make().Queue;
             var redisConfig = ConfigurationOptions.Parse($"{queueSettings.RedisHost}:{queueSettings.RedisPort}");
@@ -39,17 +37,26 @@ public IDatabase Redis
         public static int RedisPort { get; set; }
         public static string RedisPassword { get; set; }
 
-        public static Broker Make()
+        public Broker()
+        {}
+
+        public Broker(bool useSecondaryQueue) : this()
+        {
+            if (useSecondaryQueue)
+                QueueName = "Secondary";
+        }
+
+        public static Broker Make(bool useSecondaryQueue = false)
         {
-            return new Broker() {
+            return new Broker(useSecondaryQueue) {
                 QueueBaseName = InEngineSettings.Make().Queue.QueueName
             };
         }
 
-        public void Publish(ICommand command, bool useSecondaryQueue = false)
+        public void Publish(ICommand command)
         {
             Redis.ListLeftPush(
-                useSecondaryQueue ? SecondaryWaitingQueueName : PrimaryWaitingQueueName,
+                PendingQueueName,
                 new Message() {
                     CommandClassName = command.GetType().FullName,
                     CommandAssemblyName = command.GetType().Assembly.GetName().Name + ".dll",
@@ -58,13 +65,9 @@ public void Publish(ICommand command, bool useSecondaryQueue = false)
             );
         }
 
-        public bool Consume(bool useSecondaryQueue = false)
+        public bool Consume()
         {
-            var waitingQueueName = useSecondaryQueue ? SecondaryWaitingQueueName : PrimaryWaitingQueueName;
-            var processingQueueName = useSecondaryQueue ? SecondaryProcessingQueueName : PrimaryProcessingQueueName;
-            var failedQueueName = useSecondaryQueue ? SecondaryFailedQueueName : PrimaryFailedQueueName;
-
-            var stageMessageTask = Redis.ListRightPopLeftPush(waitingQueueName, processingQueueName);
+            var stageMessageTask = Redis.ListRightPopLeftPush(PendingQueueName, InProgressQueueName);
             var serializedMessage = stageMessageTask.ToString();
             if (serializedMessage == null)
                 return false;
@@ -83,94 +86,58 @@ public bool Consume(bool useSecondaryQueue = false)
             }
             catch (Exception exception)
             {
-                Redis.ListRemove(processingQueueName, serializedMessage, 1);
-                Redis.ListLeftPush(failedQueueName, stageMessageTask);
+                Redis.ListRemove(InProgressQueueName, serializedMessage, 1);
+                Redis.ListLeftPush(FailedQueueName, stageMessageTask);
                 throw new CommandFailedException("Consumed command failed.", exception);
             }
 
             try
             {
-                Redis.ListRemove(processingQueueName, serializedMessage, 1);
+                Redis.ListRemove(InProgressQueueName, serializedMessage, 1);
             }
             catch (Exception exception)
             {
-                throw new CommandFailedException($"Failed to remove completed message from queue: {processingQueueName}", exception);
+                throw new CommandFailedException($"Failed to remove completed message from queue: {InProgressQueueName}", exception);
             }
 
             return true;
         }
 
-        #region Primary Queue Management Methods
-        public long GetPrimaryWaitingQueueLength()
-        {
-            return Redis.ListLength(PrimaryWaitingQueueName);
-        }
-
-        public long GetPrimaryProcessingQueueLength()
-        {
-            return Redis.ListLength(PrimaryProcessingQueueName);
-        }
-
-        public long GetPrimaryFailedQueueLength()
-        {
-            return Redis.ListLength(PrimaryFailedQueueName);
-        }
-
-        public bool ClearPrimaryWaitingQueue()
+        #region Queue Management Methods
+        public long GetPendingQueueLength()
         {
-            return Redis.KeyDelete(PrimaryWaitingQueueName);
+            return Redis.ListLength(PendingQueueName);
         }
 
-        public bool ClearPrimaryProcessingQueue()
+        public long GetInProgressQueueLength()
         {
-            return Redis.KeyDelete(PrimaryProcessingQueueName);
+            return Redis.ListLength(InProgressQueueName);
         }
 
-        public bool ClearPrimaryFailedQueue()
+        public long GetFailedQueueLength()
         {
-            return Redis.KeyDelete(PrimaryFailedQueueName);
+            return Redis.ListLength(FailedQueueName);
         }
-        #endregion
 
-        #region Secondary Queue Management Methods
-        public long GetSecondaryWaitingQueueLength()
+        public bool ClearPendingQueue()
         {
-            return Redis.ListLength(SecondaryWaitingQueueName);
+            return Redis.KeyDelete(PendingQueueName);
         }
 
-        public long GetSecondaryProcessingQueueLength()
+        public bool ClearInProgressQueue()
         {
-            return Redis.ListLength(SecondaryProcessingQueueName);
+            return Redis.KeyDelete(InProgressQueueName);
         }
 
-        public long GetSecondaryFailedQueueLength()
+        public bool ClearFailedQueue()
         {
-            return Redis.ListLength(SecondaryFailedQueueName);
+            return Redis.KeyDelete(FailedQueueName);
         }
 
-        public bool ClearSecondaryWaitingQueue()
+        public void RepublishFailedMessages()
         {
-            return Redis.KeyDelete(SecondaryWaitingQueueName);
-        }
-
-        public bool ClearSecondaryProcessingQueue()
-        {
-            return Redis.KeyDelete(SecondaryProcessingQueueName);
-        }
-
-
-        public bool ClearSecondaryFailedQueue()
-        {
-            return Redis.KeyDelete(SecondaryFailedQueueName);
+            Redis.ListRightPopLeftPush(FailedQueueName, PendingQueueName);
         }
         #endregion
-
-        public void RepublishFailedMessages(bool useSecondaryQueue)
-        {
-            Redis.ListRightPopLeftPush(
-                useSecondaryQueue ? SecondaryFailedQueueName : PrimaryFailedQueueName,
-                useSecondaryQueue ? SecondaryWaitingQueueName : PrimaryWaitingQueueName
-            );
-        }
     }
 }

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

@@ -5,33 +5,24 @@ namespace InEngine.Core.Queue.Commands
 {
     public class ClearAll : AbstractCommand
     {
-        [Option("failed-queue", HelpText = "Clear the failed queue.")]
+        [Option("failed", HelpText = "Clear the failed queue.")]
         public bool ClearFailedQueue { get; set; }
 
-        [Option("processing-queue", HelpText = "Clear the processing queue.")]
+        [Option("in-progress", HelpText = "Clear the processing queue.")]
         public bool ClearProcessingQueue { get; set; }
 
         [Option("secondary", HelpText = "Clear the secondary queue.")]
         public bool UseSecondaryQueue { get; set; }
 
         public override void Run()
         {
-            var broker = Broker.Make();
-            if (UseSecondaryQueue) {
-                if (ClearProcessingQueue)
-                    Info(broker.ClearSecondaryProcessingQueue().ToString());
-                else if (ClearFailedQueue)
-                    Info(broker.ClearSecondaryFailedQueue().ToString());
-                else
-                    Info(broker.ClearSecondaryWaitingQueue().ToString());
-            } else {
-                if (ClearProcessingQueue)
-                    Info(broker.ClearPrimaryProcessingQueue().ToString());
-                else if (ClearFailedQueue)
-                    Info(broker.ClearPrimaryFailedQueue().ToString());
-                else
-                    Info(broker.ClearPrimaryWaitingQueue().ToString());
-            }
+            var broker = Broker.Make(UseSecondaryQueue);
+            if (ClearProcessingQueue)
+                Info(broker.ClearInProgressQueue().ToString());
+            else if (ClearFailedQueue)
+                Info(broker.ClearFailedQueue().ToString());
+            else
+                Info(broker.ClearPendingQueue().ToString());
         }
     }
 }

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

@@ -15,10 +15,10 @@ public class Consume : AbstractCommand
         public override void Run()
         {
             UseSecondaryQueue = UseSecondaryQueue || GetJobContextData<bool>("useSecondaryQueue");
-            var broker = Broker.Make();
+            var broker = Broker.Make(UseSecondaryQueue);
             var shouldConsume = true;
             while (shouldConsume)
-                shouldConsume = broker.Consume(UseSecondaryQueue) && ShouldConsumeAll;
+                shouldConsume = broker.Consume() && ShouldConsumeAll;
         }
 
         public override void Failed(Exception exception)

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

@@ -7,24 +7,21 @@ public class Length : AbstractCommand
     {
         public override void Run()
         {
-            var broker = Broker.Make();
-            var leftPadding = 15;
-            Warning("Primary Queue:");
-            InfoText("Pending".PadLeft(leftPadding));
-            Line(broker.GetPrimaryWaitingQueueLength().ToString().PadLeft(10));
-            InfoText("In-progress".PadLeft(leftPadding));
-            Line(broker.GetPrimaryProcessingQueueLength().ToString().PadLeft(10));
-            ErrorText("Failed".PadLeft(leftPadding));
-            Line(broker.GetPrimaryFailedQueueLength().ToString().PadLeft(10));
-            Newline();
+            PrintUsage(Broker.Make());
+            PrintUsage(Broker.Make(true));
+        }
 
-            Warning("Secondary Queue:");
+        public void PrintUsage(Broker broker)
+        {
+            var leftPadding = 15;
+            Warning($"{broker.QueueName} Queue:");
+            broker = Broker.Make(true);
             InfoText("Pending".PadLeft(leftPadding));
-            Line(broker.GetSecondaryWaitingQueueLength().ToString().PadLeft(10));
+            Line(broker.GetPendingQueueLength().ToString().PadLeft(10));
             InfoText("In-progress".PadLeft(leftPadding));
-            Line(broker.GetSecondaryProcessingQueueLength().ToString().PadLeft(10));
+            Line(broker.GetInProgressQueueLength().ToString().PadLeft(10));
             ErrorText("Failed".PadLeft(leftPadding));
-            Line(broker.GetSecondaryFailedQueueLength().ToString().PadLeft(10));
+            Line(broker.GetFailedQueueLength().ToString().PadLeft(10));
             Newline();
         }
     }

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

@@ -30,7 +30,7 @@ public override void Run()
             if (Arguments != null)
                 Parser.Default.ParseArguments(Arguments.ToList().Select(x => $"--{x}").ToArray(), command);
 
-            Broker.Make().Publish(command, UseSecondaryQueue);
+            Broker.Make(UseSecondaryQueue).Publish(command);
         }
 
         public override void Failed(Exception exception)

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

@@ -14,10 +14,10 @@ public class RepublishFailed : AbstractCommand
 
         public override void Run()
         {
-            var broker = Broker.Make();
+            var broker = Broker.Make(UseSecondaryQueue);
             Enumerable.Range(0, Count)
                       .ToList()
-                      .ForEach(x => broker.RepublishFailedMessages(UseSecondaryQueue));
+                      .ForEach(x => broker.RepublishFailedMessages());
         }
     }
 }