Move messages to failed queue when plugin isn't registered

Author: ethanhann

src/InEngine.Core/Exceptions/CommandNotExtractableFromEnvelopeException.cs

@@ -0,0 +1,14 @@
+using System;
+
+namespace InEngine.Core.Exceptions
+{
+    public class CommandNotExtractableFromEnvelopeException : Exception
+    {
+        public CommandNotExtractableFromEnvelopeException(string command, Exception exception)
+            : base(
+                $"The plugin is (probably) not registered - check the message in the failure queue for details. Command name: {command}", 
+                exception
+            )
+        {}
+    }
+}

src/InEngine.Core/Queuing/Clients/FileClient.cs

@@ -54,28 +54,40 @@ public string FailedQueuePath
 
         public void Publish(AbstractCommand command)
         {
-            if (!Directory.Exists(PendingQueuePath))
-                Directory.CreateDirectory(PendingQueuePath);
-            var serializedMessage = new CommandEnvelope() {
+            PublishToQueue(new CommandEnvelope() {
                 IsCompressed = UseCompression,
                 CommandClassName = command.GetType().FullName,
                 PluginName = command.GetType().Assembly.GetName().Name,
                 SerializedCommand = command.SerializeToJson(UseCompression)
-            }.SerializeToJson();
-            using (var streamWriter = File.CreateText(Path.Combine(PendingQueuePath, Guid.NewGuid().ToString()))) 
-            {
-                streamWriter.Write(serializedMessage);
-            }   
+            }, PendingQueuePath);
+        }
+
+        void PublishToQueue(CommandEnvelope commandEnvelope, string queuePath)
+        {
+            if (!Directory.Exists(queuePath))
+                Directory.CreateDirectory(queuePath);
+
+            File.WriteAllText(
+                Path.Combine(queuePath, Guid.NewGuid().ToString()),
+                commandEnvelope.SerializeToJson()
+            );
         }
 
         public void Consume(CancellationToken cancellationToken)
         {
             try
             {
-                while(true)
+                while(true) 
                 {
-                    if (Consume() == null)
-                        Thread.Sleep(5000);   
+                    try
+                    {
+                        if (Consume() == null)
+                            Thread.Sleep(5000);   
+                    }
+                    catch (Exception exception)
+                    {
+                        Log.Error(exception);
+                    }
                     cancellationToken.ThrowIfCancellationRequested();
                 }
             }
@@ -109,11 +121,12 @@ public ICommandEnvelope Consume()
                 return null;
             }
             consumeLock.ReleaseMutex();
-            
-            var commandEnvelope = File.ReadAllText(inProgressFilePath).DeserializeFromJson<CommandEnvelope>() as ICommandEnvelope;
-            var command = commandEnvelope.GetCommandInstance();
-            command.CommandLifeCycle.IncrementRetry();
-            commandEnvelope.SerializedCommand = command.SerializeToJson(UseCompression);
+
+            var commandEnvelope = File.ReadAllText(inProgressFilePath).DeserializeFromJson<CommandEnvelope>();
+            var command = commandEnvelope.GetCommandInstanceAndIncrementRetry(() => {
+                File.Move(inProgressFilePath, Path.Combine(FailedQueuePath, fileInfo.Name));
+            });
+
             try
             {
                 command.WriteSummaryToConsole();
@@ -122,8 +135,10 @@ public ICommandEnvelope Consume()
             catch (Exception exception)
             {
                 Log.Error(exception);
-                if (command.CommandLifeCycle.ShouldRetry())
-                    File.Move(inProgressFilePath, Path.Combine(PendingQueuePath, fileInfo.Name));
+                if (command.CommandLifeCycle.ShouldRetry()) {
+                    PublishToQueue(commandEnvelope, PendingQueuePath);
+                    File.Delete(Path.Combine(PendingQueuePath, fileInfo.Name));
+                }
                 else
                     File.Move(inProgressFilePath, Path.Combine(FailedQueuePath, fileInfo.Name));
                 throw new CommandFailedException("Failed to consume command.", exception);

src/InEngine.Core/Queuing/Clients/RabbitMQClient.cs

@@ -100,9 +100,10 @@ public void Consume(CancellationToken cancellationToken)
                 if (commandEnvelope == null)
                     throw new CommandFailedException("Could not deserialize the command.");
 
-                var command = commandEnvelope.GetCommandInstance();
-                command.CommandLifeCycle.IncrementRetry();
-                commandEnvelope.SerializedCommand = command.SerializeToJson(UseCompression);
+                var command = commandEnvelope.GetCommandInstanceAndIncrementRetry(() => {
+                    eventingConsumer.Model.BasicNack(result.DeliveryTag, false, false);
+                });
+
                 try
                 {
                     command.WriteSummaryToConsole();
@@ -137,9 +138,10 @@ public ICommandEnvelope Consume()
             if (commandEnvelope == null)
                 throw new CommandFailedException("Could not deserialize the command.");
 
-            var command = commandEnvelope.GetCommandInstance();
-            command.CommandLifeCycle.IncrementRetry();
-            commandEnvelope.SerializedCommand = command.SerializeToJson(UseCompression);
+            var command = commandEnvelope.GetCommandInstanceAndIncrementRetry(() => {
+                Channel.BasicNack(result.DeliveryTag, false, false);
+            });
+
             try
             {
                 command.WriteSummaryToConsole();

src/InEngine.Core/Queuing/Clients/RedisClient.cs

@@ -102,9 +102,10 @@ public ICommandEnvelope Consume()
             if (commandEnvelope == null)
                 throw new CommandFailedException("Could not deserialize the command.");
 
-            var command = commandEnvelope.GetCommandInstance();
-            command.CommandLifeCycle.IncrementRetry();
-            commandEnvelope.SerializedCommand = command.SerializeToJson(UseCompression);
+            var command = commandEnvelope.GetCommandInstanceAndIncrementRetry(() => {
+                Redis.ListLeftPush(FailedQueueName, commandEnvelope.SerializeToJson());
+            });
+    
             try
             {
                 command.WriteSummaryToConsole();

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

@@ -91,7 +91,7 @@ public void PrintMessages(List<ICommandEnvelope> messages, string queueName)
                 if (JsonFormat)
                     Line(commandEnvelope.SerializeToJson());
                 else
-                    konsoleForm.Write(commandEnvelope.GetCommandInstance());
+                    konsoleForm.Write(commandEnvelope.GetCommandInstanceAndIncrementRetry());
             });
         }
     }

src/InEngine.Core/Queuing/Commands/Status.cs

@@ -0,0 +1,25 @@
+using System.Linq;
+
+namespace InEngine.Core.Queuing.Commands
+{
+    public class Length : AbstractCommand, IHasQueueSettings
+    {
+        public QueueSettings QueueSettings { get; set; }
+
+        public override void Run()
+        {
+            PrintQueueLengths(QueueAdapter.Make(false, QueueSettings, MailSettings));
+            PrintQueueLengths(QueueAdapter.Make(true, QueueSettings, MailSettings));
+        }
+
+        public void PrintQueueLengths(QueueAdapter queue)
+        {
+            Warning($"{queue.QueueName} Queue:");
+            queue.GetQueueLengths().ToList().ForEach(x => { 
+                InfoText(x.Key.PadLeft(15));
+                Line(x.Value.ToString().PadLeft(10));
+            });
+            Newline();
+        }
+    }
+}

src/InEngine.Core/Queuing/Message/CommandEnvelope.cs

@@ -12,12 +12,21 @@ public class CommandEnvelope : ICommandEnvelope
         public DateTime QueuedAt { get; set; } = DateTime.UtcNow;
         public bool IsCompressed { get; set; }
 
-        public AbstractCommand GetCommandInstance()
+        public AbstractCommand GetCommandInstanceAndIncrementRetry(Action actionOnFail = null)
         {
-            var commandType = PluginAssembly.LoadFrom(PluginName).GetCommandType(CommandClassName);
-            if (commandType == null)
-                throw new CommandFailedException($"Could not locate command {CommandClassName}. Is the {PluginName} plugin registered in the settings file?");
-            return SerializedCommand.DeserializeFromJson<AbstractCommand>(IsCompressed);
+            try
+            {
+                PluginAssembly.LoadFrom(PluginName).GetCommandType(CommandClassName);
+                var command = SerializedCommand.DeserializeFromJson<AbstractCommand>(IsCompressed);
+                command.CommandLifeCycle.IncrementRetry();
+                SerializedCommand = command.SerializeToJson(IsCompressed);
+                return command;
+            }
+            catch (Exception exception)
+            {
+                actionOnFail.Invoke();
+                throw new CommandNotExtractableFromEnvelopeException(CommandClassName, exception);
+            }
         }
     }
 }

src/InEngine.Core/Queuing/Message/ICommandEnvelope.cs

@@ -10,6 +10,6 @@ public interface ICommandEnvelope
         string SerializedCommand { get; set; }
         DateTime QueuedAt { get; set; }
         bool IsCompressed { get; set; }
-        AbstractCommand GetCommandInstance();
+        AbstractCommand GetCommandInstanceAndIncrementRetry(Action actionOnFail = null);
     }
 }