Refactor message class to command envelope

Author: ethanhann

src/InEngine.Core/AbstractCommand.cs

@@ -51,12 +51,14 @@ public void UpdateProgress(int tick)
         #region Scheduling
         public virtual void Execute(IJobExecutionContext context)
         {
-            var properties = GetType().GetProperties();
-            context.MergedJobDataMap.ToList().ForEach(x => {
-                var property = properties.FirstOrDefault(y => y.Name == x.Key);
-                if (property != null)
-                    property.SetValue(this, x.Value);                
-            });
+            if (context != null) {
+                var properties = GetType().GetProperties();
+                context.MergedJobDataMap.ToList().ForEach(x => {
+                    var property = properties.FirstOrDefault(y => y.Name == x.Key);
+                    if (property != null)
+                        property.SetValue(this, x.Value);                
+                });
+            }
 
             try
             {

…Queuing/Clients/Database/MessageModel.cs …Clients/Database/CommandEnvelopeModel.cssrc/InEngine.Core/Queuing/Clients/Database/MessageModel.cs renamed to src/InEngine.Core/Queuing/Clients/Database/CommandEnvelopeModel.cs src/InEngine.Core/Queuing/Clients/Database/MessageModel.cs renamed to src/InEngine.Core/Queuing/Clients/Database/CommandEnvelopeModel.cs

@@ -1,6 +1,6 @@
 namespace InEngine.Core.Queuing.Clients.Database
 {
-    public class MessageModel : Message
+    public class CommandEnvelopeModel : CommandEnvelope
     {
         public string Status { get; set; }
         public string QueueName { get; set; }

…ueuing/Clients/Database/MessageStatus.cs …lients/Database/CommandEnvelopeStatus.cssrc/InEngine.Core/Queuing/Clients/Database/MessageStatus.cs renamed to src/InEngine.Core/Queuing/Clients/Database/CommandEnvelopeStatus.cs src/InEngine.Core/Queuing/Clients/Database/MessageStatus.cs renamed to src/InEngine.Core/Queuing/Clients/Database/CommandEnvelopeStatus.cs

@@ -1,6 +1,6 @@
 namespace InEngine.Core.Queuing.Clients.Database
 {
-    public static class MessageStatus
+    public static class CommandEnvelopeStatus
     {
         public static string Pending { get => "Pending"; }
         public static string InProgress { get => "InProgress"; }

src/InEngine.Core/Queuing/Clients/Database/QueueDbContext.cs

@@ -5,7 +5,7 @@ namespace InEngine.Core.Queuing.Clients.Database
 {
     public class QueueDbContext : DbContext
     {
-        public DbSet<MessageModel> Messages { get; set; }
+        public DbSet<CommandEnvelopeModel> Messages { get; set; }
         public string MessageTableName { get; set; }
 
         public QueueDbContext(string messageTableName) 
@@ -21,7 +21,7 @@ public QueueDbContext(DbConnection existingConnection, bool contextOwnsConnectio
 
         protected override void OnModelCreating(DbModelBuilder modelBuilder)
         {
-            modelBuilder.Entity<MessageModel>().ToTable(MessageTableName);
+            modelBuilder.Entity<CommandEnvelopeModel>().ToTable(MessageTableName);
         }
     }
 }

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

@@ -21,8 +21,8 @@ public void Publish(ICommand command)
 
             using (var context = new QueueDbContext(QueueBaseName))
             {
-                context.Messages.Add(new MessageModel() {
-                    Status = MessageStatus.Pending,
+                context.Messages.Add(new CommandEnvelopeModel() {
+                    Status = CommandEnvelopeStatus.Pending,
                     IsCompressed = UseCompression,
                     CommandClassName = command.GetType().FullName,
                     CommandAssemblyName = command.GetType().Assembly.GetName().Name + ".dll",
@@ -35,7 +35,7 @@ public void Publish(ICommand command)
         public bool Consume()
         {
 
-            IMessage message;
+            ICommandEnvelope commandEnvelope;
             using (var conn = new SqlConnection())
             {
                 conn.Open();
@@ -48,10 +48,10 @@ public bool Consume()
                             context.Database.UseTransaction(transaction);
                             var messageModel = context.Messages
                                                       .OrderBy(x => x.QueuedAt)
-                                                      .FirstOrDefault(x => x.Status == MessageStatus.Pending&& x.QueueName == QueueName);
-                            messageModel.Status = MessageStatus.InProgress;
+                                                      .FirstOrDefault(x => x.Status == CommandEnvelopeStatus.Pending&& x.QueueName == QueueName);
+                            messageModel.Status = CommandEnvelopeStatus.InProgress;
                             context.SaveChanges();
-                            message = messageModel;
+                            commandEnvelope = messageModel;
                         }
 
                         transaction.Commit();
@@ -66,7 +66,7 @@ public bool Consume()
 
             try
             {
-                Queue.ExtractCommandInstanceFromMessage(message).Run();
+                Queue.ExtractCommandInstanceFromMessageAndRun(commandEnvelope);
             }
             catch (Exception exception)
             {
@@ -80,18 +80,18 @@ public bool Consume()
                             using (var context = new QueueDbContext(conn, false, QueueBaseName))
                             {
                                 context.Database.UseTransaction(transaction);
-                                var messageModel = context.Messages.FirstOrDefault(x => x.Id == message.Id);
-                                messageModel.Status = MessageStatus.Failed;
+                                var messageModel = context.Messages.FirstOrDefault(x => x.Id == commandEnvelope.Id);
+                                messageModel.Status = CommandEnvelopeStatus.Failed;
                                 context.SaveChanges();
-                                message = messageModel;
+                                commandEnvelope = messageModel;
                             }
 
                             transaction.Commit();
                         }
                         catch (Exception dbException)
                         {
                             transaction.Rollback();
-                            throw new CommandFailedException("Failed to set queue message from in-progress to failed.", dbException);
+                            throw new CommandFailedException("Failed to set queue commandEnvelope from in-progress to failed.", dbException);
                         }
                     }
                 }
@@ -110,42 +110,42 @@ public bool Consume()
                             using (var context = new QueueDbContext(conn, false, QueueBaseName))
                             {
                                 context.Database.UseTransaction(transaction);
-                                var messageModel = context.Messages.FirstOrDefault(x => x.Id == message.Id);
-                                messageModel.Status = MessageStatus.Completed;
+                                var messageModel = context.Messages.FirstOrDefault(x => x.Id == commandEnvelope.Id);
+                                messageModel.Status = CommandEnvelopeStatus.Completed;
                                 context.SaveChanges();
-                                message = messageModel;
+                                commandEnvelope = messageModel;
                             }
                             transaction.Commit();
                         }
                         catch (Exception dbException)
                         {
                             transaction.Rollback();
-                            throw new CommandFailedException("Failed to set queue message from in-progress to completed.", dbException);
+                            throw new CommandFailedException("Failed to set queue commandEnvelope from in-progress to completed.", dbException);
                         }
                     }
                 }
             }
             catch (Exception exception)
             {
-                throw new CommandFailedException($"Failed to remove completed message from queue", exception);
+                throw new CommandFailedException($"Failed to remove completed commandEnvelope from queue", exception);
             }
 
             return true;
         }
 
         public long GetPendingQueueLength()
         {
-            return GetQueueLength(MessageStatus.Pending);
+            return GetQueueLength(CommandEnvelopeStatus.Pending);
         }
 
         public long GetInProgressQueueLength()
         {
-            return GetQueueLength(MessageStatus.InProgress);
+            return GetQueueLength(CommandEnvelopeStatus.InProgress);
         }
 
         public long GetFailedQueueLength()
         {
-            return GetQueueLength(MessageStatus.Failed);
+            return GetQueueLength(CommandEnvelopeStatus.Failed);
         }
 
         public long GetQueueLength(string status)
@@ -158,19 +158,19 @@ public long GetQueueLength(string status)
 
         public bool ClearPendingQueue()
         {
-            ClearQueue(MessageStatus.Pending);
+            ClearQueue(CommandEnvelopeStatus.Pending);
             return true;
         }
 
         public bool ClearInProgressQueue()
         {
-            ClearQueue(MessageStatus.InProgress);
+            ClearQueue(CommandEnvelopeStatus.InProgress);
             return true;
         }
 
         public bool ClearFailedQueue()
         {
-            ClearQueue(MessageStatus.Failed);
+            ClearQueue(CommandEnvelopeStatus.Failed);
             return true;
         }
 
@@ -198,10 +198,10 @@ public void RepublishFailedMessages()
                         {
                             context.Database.UseTransaction(transaction);
                             var messageModels = context.Messages
-                                                       .Where(x => x.QueueName == QueueName && x.Status == MessageStatus.Pending)
+                                                       .Where(x => x.QueueName == QueueName && x.Status == CommandEnvelopeStatus.Pending)
                                                         .OrderBy(x => x.QueuedAt);
                             foreach(var messageModel in messageModels)
-                                messageModel.Status = MessageStatus.InProgress;
+                                messageModel.Status = CommandEnvelopeStatus.InProgress;
                             context.SaveChanges();
                         }
 
@@ -216,30 +216,30 @@ public void RepublishFailedMessages()
             }
         }
 
-        public List<IMessage> PeekPendingMessages(long from, long to)
+        public List<ICommandEnvelope> PeekPendingMessages(long from, long to)
         {
-            return GetMessages(MessageStatus.Pending, from, to);
+            return GetMessages(CommandEnvelopeStatus.Pending, from, to);
         }
 
-        public List<IMessage> PeekInProgressMessages(long from, long to)
+        public List<ICommandEnvelope> PeekInProgressMessages(long from, long to)
         {
-            return GetMessages(MessageStatus.InProgress, from, to);
+            return GetMessages(CommandEnvelopeStatus.InProgress, from, to);
         }
 
-        public List<IMessage> PeekFailedMessages(long from, long to)
+        public List<ICommandEnvelope> PeekFailedMessages(long from, long to)
         {
-            return GetMessages(MessageStatus.Failed, from, to);
+            return GetMessages(CommandEnvelopeStatus.Failed, from, to);
         }
 
-        public List<IMessage> GetMessages(string status, long from, long to)
+        public List<ICommandEnvelope> GetMessages(string status, long from, long to)
         {
             using (var context = new QueueDbContext(QueueBaseName))
             {
                 return context.Messages
                               .Where(x => x.QueueName == QueueName && x.Status == status)
                               .Skip(Convert.ToInt32(from))
                               .Take(Convert.ToInt32(to - from))
-                              .Select(x => x as IMessage)
+                              .Select(x => x as ICommandEnvelope)
                               .ToList();
             }
         }

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

@@ -46,7 +46,7 @@ public void Publish(ICommand command)
         {
             if (!Directory.Exists(PendingQueuePath))
                 Directory.CreateDirectory(PendingQueuePath);
-            var serializedMessage = new Message()
+            var serializedMessage = new CommandEnvelope()
             {
                 IsCompressed = UseCompression,
                 CommandClassName = command.GetType().FullName,
@@ -79,15 +79,15 @@ public bool Consume()
                 return false;
             }
 
-            var message = File.ReadAllText(inProgressFilePath).DeserializeFromJson<Message>();
+            var commandEnvelope = File.ReadAllText(inProgressFilePath).DeserializeFromJson<CommandEnvelope>();
             try
             {
-                Queue.ExtractCommandInstanceFromMessageAndRun(message as IMessage);
+                Queue.ExtractCommandInstanceFromMessageAndRun(commandEnvelope as ICommandEnvelope);
             }
             catch (Exception exception)
             {
                 File.Move(inProgressFilePath, Path.Combine(FailedQueuePath, fileInfo.Name));
-                throw new CommandFailedException("Failed to consume message.", exception);
+                throw new CommandFailedException("Failed to consume commandEnvelope.", exception);
             }
 
             try
@@ -96,7 +96,7 @@ public bool Consume()
             }
             catch (Exception exception)
             {
-                throw new CommandFailedException("Failed to move message from in-progress queue.", exception);
+                throw new CommandFailedException("Failed to move commandEnvelope from in-progress queue.", exception);
             }
             return true;
         }
@@ -148,22 +148,22 @@ public bool ClearQueue(string queuePath)
             return true;
         }
 
-        public List<IMessage> PeekFailedMessages(long from, long to)
+        public List<ICommandEnvelope> PeekFailedMessages(long from, long to)
         {
             return PeekMessages(FailedQueuePath, from, to);
         }
 
-        public List<IMessage> PeekInProgressMessages(long from, long to)
+        public List<ICommandEnvelope> PeekInProgressMessages(long from, long to)
         {
             return PeekMessages(InProgressQueuePath, from, to);
         }
 
-        public List<IMessage> PeekPendingMessages(long from, long to)
+        public List<ICommandEnvelope> PeekPendingMessages(long from, long to)
         {
             return PeekMessages(PendingQueuePath, from, to);
         }
 
-        public List<IMessage> PeekMessages(string queuePath, long from, long to)
+        public List<ICommandEnvelope> PeekMessages(string queuePath, long from, long to)
         {
             var maxResults = Convert.ToInt32(to + from);
             var files = new DirectoryInfo(queuePath)
@@ -172,11 +172,11 @@ public List<IMessage> PeekMessages(string queuePath, long from, long to)
                 .ToList();
 
             if (files.Count() <= maxResults)
-                return files.Select(x => File.ReadAllText(x.FullName).DeserializeFromJson<Message>() as IMessage).ToList();
+                return files.Select(x => File.ReadAllText(x.FullName).DeserializeFromJson<CommandEnvelope>() as ICommandEnvelope).ToList();
 
             return files
                 .GetRange(Convert.ToInt32(from), Convert.ToInt32(to))
-                .Select(x => File.ReadAllText(x.FullName).DeserializeFromJson<Message>() as IMessage)
+                .Select(x => File.ReadAllText(x.FullName).DeserializeFromJson<CommandEnvelope>() as ICommandEnvelope)
                 .ToList();
         }
     }

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

@@ -32,7 +32,7 @@ public void Publish(ICommand command)
         {
             Redis.ListLeftPush(
                 PendingQueueName,
-                new Message() {
+                new CommandEnvelope() {
                     IsCompressed = UseCompression,
                     CommandClassName = command.GetType().FullName,
                     CommandAssemblyName = command.GetType().Assembly.GetName().Name + ".dll",
@@ -47,13 +47,13 @@ public bool Consume()
             var serializedMessage = rawRedisMessageValue.ToString();
             if (serializedMessage == null)
                 return false;
-            var message = serializedMessage.DeserializeFromJson<Message>();
-            if (message == null)
+            var commandEnvelope = serializedMessage.DeserializeFromJson<CommandEnvelope>();
+            if (commandEnvelope == null)
                 return false;
 
             try
             {
-                Queue.ExtractCommandInstanceFromMessageAndRun(message as IMessage);
+                Queue.ExtractCommandInstanceFromMessageAndRun(commandEnvelope as ICommandEnvelope);
             }
             catch (Exception exception)
             {
@@ -68,7 +68,7 @@ public bool Consume()
             }
             catch (Exception exception)
             {
-                throw new CommandFailedException($"Failed to remove completed message from queue: {InProgressQueueName}", exception);
+                throw new CommandFailedException($"Failed to remove completed commandEnvelope from queue: {InProgressQueueName}", exception);
             }
 
             return true;
@@ -109,26 +109,26 @@ public void RepublishFailedMessages()
             Redis.ListRightPopLeftPush(FailedQueueName, PendingQueueName);
         }
 
-        public List<IMessage> PeekPendingMessages(long from, long to)
+        public List<ICommandEnvelope> PeekPendingMessages(long from, long to)
         {
             return GetMessages(PendingQueueName, from, to);
         }
 
-        public List<IMessage> PeekInProgressMessages(long from, long to)
+        public List<ICommandEnvelope> PeekInProgressMessages(long from, long to)
         {
             return GetMessages(InProgressQueueName, from, to);
         }
 
-        public List<IMessage> PeekFailedMessages(long from, long to)
+        public List<ICommandEnvelope> PeekFailedMessages(long from, long to)
         {
             return GetMessages(FailedQueueName, from, to);
         }
 
-        public List<IMessage> GetMessages(string queueName, long from, long to)
+        public List<ICommandEnvelope> GetMessages(string queueName, long from, long to)
         {
             return Redis.ListRange(queueName, from, to)
                         .ToStringArray()
-                        .Select(x => x.DeserializeFromJson<Message>() as IMessage).ToList();
+                        .Select(x => x.DeserializeFromJson<CommandEnvelope>() as ICommandEnvelope).ToList();
         }
     }
 }