Add queue peek command

Author: ethanhann

src/InEngine.Core/InEngine.Core.csproj

@@ -8,11 +8,11 @@
     <PackageReference Include="CommandLineParser" Version="1.9.71" />
     <PackageReference Include="NLog" Version="4.4.12" />
     <PackageReference Include="Quartz" Version="2.6.1" />
-    <PackageReference Include="Goblinfactory.Konsole.ProgressBar.Core" Version="1.0.2" />
     <PackageReference Include="Microsoft.Extensions.Configuration" Version="2.0.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="2.0.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="2.0.0" />
     <PackageReference Include="StackExchange.Redis" Version="1.2.4" />
+    <PackageReference Include="Goblinfactory.Konsole" Version="3.1.0" />
   </ItemGroup>
   <ItemGroup>
     <Compile Include="..\SharedAssemblyInfo.cs">

src/InEngine.Core/Queue/Broker.cs

@@ -1,4 +1,6 @@
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Reflection;
 using System.Threading.Tasks;
 using InEngine.Core.Exceptions;
@@ -74,11 +76,7 @@ public bool Consume()
             var message = serializedMessage.DeserializeFromJson<Message>();
             if (message == null)
                 return false;
-
-            var commandType = Type.GetType($"{message.CommandClassName}, {message.CommandAssemblyName}");
-            if (commandType == null)
-                throw new CommandFailedException("Consumed command failed: could not locate command type.");
-            var commandInstance = JsonConvert.DeserializeObject(message.SerializedCommand, commandType) as ICommand;
+            var commandInstance = ExtractCommandInstanceFromMessage(message);
 
             try
             {
@@ -103,6 +101,14 @@ public bool Consume()
             return true;
         }
 
+        public ICommand ExtractCommandInstanceFromMessage(Message message)
+        {
+            var commandType = Type.GetType($"{message.CommandClassName}, {message.CommandAssemblyName}");
+            if (commandType == null)
+                throw new CommandFailedException("Could not locate command type.");
+            return JsonConvert.DeserializeObject(message.SerializedCommand, commandType) as ICommand;
+        }
+
         #region Queue Management Methods
         public long GetPendingQueueLength()
         {
@@ -138,6 +144,26 @@ public void RepublishFailedMessages()
         {
             Redis.ListRightPopLeftPush(FailedQueueName, PendingQueueName);
         }
+
+        public List<Message> PeekPendingMessages(long from, long to)
+        {
+            return GetMessages(PendingQueueName, from, to);
+        }
+
+        public List<Message> PeekInProgressMessages(long from, long to)
+        {
+            return GetMessages(InProgressQueueName, from, to);
+        }
+
+        public List<Message> PeekFailedMessages(long from, long to)
+        {
+            return GetMessages(FailedQueueName, from, to);
+        }
+
+        public List<Message> GetMessages(string queueName, long from, long to)
+        {
+            return Redis.ListRange(queueName, from, to).ToStringArray().Select(x => x.DeserializeFromJson<Message>()).ToList();
+        }
         #endregion
     }
 }

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

@@ -0,0 +1,57 @@
+using System;
+using CommandLine;
+using InEngine.Core.Exceptions;
+using Konsole;
+using Konsole.Forms;
+using Newtonsoft.Json;
+
+namespace InEngine.Core.Queue.Commands
+{
+    public class Peek : AbstractCommand
+    {
+        [Option("offset", DefaultValue = 0, HelpText = "The maximum number of messages to peek.")]
+        public long Offset { get; set; }
+
+        [Option("limit", DefaultValue = 10, HelpText = "The maximum number of messages to peek.")]
+        public long Limit { get; set; }
+
+        [Option("json", HelpText = "View the messages as JSON.")]
+        public bool JsonFormat { get; set; }
+
+        [Option("pending", HelpText = "Peek at messages in the pending queue.")]
+        public bool PendingQueue { get; set; }
+
+        [Option("failed", HelpText = "Peek at messages in the failed queue.")]
+        public bool FailedQueue { get; set; }
+
+        [Option("in-progress", HelpText = "Peek at messages in the in-progress queue.")]
+        public bool InProgressQueue { get; set; }
+
+        [Option("secondary", HelpText = "Peek at messages in secondary queues. Primary queues are used by default.")]
+        public bool UseSecondaryQueue { get; set; }
+
+        public override void Run()
+        {
+            if (PendingQueue == false && FailedQueue == false && InProgressQueue == false)
+                throw new CommandFailedException("Must specify at least one queue to peek in. Use -h to see available options.");
+            var broker = Broker.Make(UseSecondaryQueue);
+            var from = Offset;
+            var to = Offset + Limit - 1;
+            if (PendingQueue) {
+                Warning("Pending:");
+                var konsoleForm = new Form(120, new ThinBoxStyle());
+                broker.PeekPendingMessages(from, to).ForEach(x => {
+                    var message = x as Message;
+                    if (JsonFormat)
+                        Line(message.SerializeToJson());
+                    else
+                        konsoleForm.Write(broker.ExtractCommandInstanceFromMessage(message));
+                });
+            }
+            if (InProgressQueue)
+                Info($"In-progress: {broker.PeekInProgressMessages(from, to).ToString()}");
+            if (FailedQueue)
+                Info($"Failed: {broker.PeekFailedMessages(from, to).ToString()}");
+        }
+    }
+}

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

@@ -6,16 +6,16 @@ namespace InEngine.Core.Queue.Commands
 {
     public class RepublishFailed : AbstractCommand
     {
-        [Option("count", DefaultValue = 3, HelpText = "The maximum number of messages to republish.")]
-        public int Count { get; set; }
+        [Option("limit", DefaultValue = 100, HelpText = "The maximum number of messages to republish.")]
+        public int Limit { get; set; }
 
         [Option("secondary", DefaultValue = false, HelpText = "Republish failed secondary queue messages.")]
         public bool UseSecondaryQueue { get; set; }
 
         public override void Run()
         {
             var broker = Broker.Make(UseSecondaryQueue);
-            Enumerable.Range(0, Count)
+            Enumerable.Range(0, Limit)
                       .ToList()
                       .ForEach(x => broker.RepublishFailedMessages());
         }

src/InEngine.Core/Queue/Options.cs

@@ -15,12 +15,15 @@ public class Options : IOptions
         [VerbOption("queue:length", HelpText = "Get the number of messages in the primary and secondary queues.")]
         public Length Length { get; set; }
 
-        [VerbOption("queue:flush", HelpText = "Clear the primary or secondary queues, and optionally.")]
+        [VerbOption("queue:flush", HelpText = "Clear the primary or secondary queues.")]
         public Flush Flush { get; set; }
 
         [VerbOption("queue:republish", HelpText = "Republish failed messages to the queue.")]
         public RepublishFailed RepublishFailed { get; set; }
 
+        [VerbOption("queue:peek", HelpText = "Peek at messages in the primary or secondary queues.")]
+        public Peek Peek { get; set; }
+
         [HelpVerbOption]
         public string GetUsage(string verb)
         {