Messages now carry information about their creation time

Nybus-project · Oct 12, 2015 · 21e59c1 · 21e59c1
1 parent e88dd4b
commit 21e59c1
Show file tree

Hide file tree

Showing 7 changed files with 42 additions and 4 deletions.
diff --git a/src/Core/Configuration/ICommandContextFactory.cs b/src/Core/Configuration/ICommandContextFactory.cs
@@ -20,7 +20,7 @@ public DefaultCommandContextFactory(IClock clock)
 
         public CommandContext<TCommand> CreateContext<TCommand>(CommandMessage<TCommand> message, INybusOptions options) where TCommand : class, ICommand
         {
-            return new CommandContext<TCommand>(message.Command, _clock.Now, message.CorrelationId);
+            return new CommandContext<TCommand>(message.Command, _clock.Now, message.CorrelationId) {InvokedOn = message.SentOn};
         }
     }
 }
diff --git a/src/Core/Configuration/IEventContextFactory.cs b/src/Core/Configuration/IEventContextFactory.cs
@@ -20,7 +20,7 @@ public DefaultEventContextFactory(IClock clock)
 
         public EventContext<TEvent> CreateContext<TEvent>(EventMessage<TEvent> message, INybusOptions options) where TEvent : class, IEvent
         {
-            return new EventContext<TEvent>(message.Event, _clock.Now, message.CorrelationId);
+            return new EventContext<TEvent>(message.Event, _clock.Now, message.CorrelationId) {RaisedOn = message.SentOn};
         }
     }
 }
diff --git a/src/Interfaces/IBusEngine.cs b/src/Interfaces/IBusEngine.cs
@@ -1,5 +1,6 @@
 using System;
 using System.Threading.Tasks;
+using Nybus.Utils;
 
 namespace Nybus
 {
@@ -29,9 +30,12 @@ public abstract class Message
         protected Message()
         {
             CorrelationId = Guid.NewGuid();
+            SentOn = Clock.Default.Now;
         }
 
         public Guid CorrelationId { get; set; }
+
+        public DateTimeOffset SentOn { get; set; }
     }
 
     public class CommandMessage<TCommand> : Message

diff --git a/src/Interfaces/ICommand.cs b/src/Interfaces/ICommand.cs
@@ -25,7 +25,11 @@ public CommandContext(TCommand message, DateTimeOffset receivedOn, Guid correlat
         }
 
         public TCommand Message { get; private set; }
+
         public DateTimeOffset ReceivedOn { get; private set; }
+
         public Guid CorrelationId { get; set; }
+
+        public DateTimeOffset InvokedOn { get; set; }
     }
 }
diff --git a/src/Interfaces/IEvent.cs b/src/Interfaces/IEvent.cs
@@ -28,6 +28,8 @@ public EventContext(TEvent eventMessage, DateTimeOffset receivedOn, Guid correla
 
         public DateTimeOffset ReceivedOn { get; private set; }
 
+        public DateTimeOffset RaisedOn { get; set; }
+
         public TEvent Message { get; private set; }
     }
 }
diff --git a/src/MassTransit/Configuration/IContextManager.cs b/src/MassTransit/Configuration/IContextManager.cs
@@ -1,5 +1,6 @@
 using System;
 using MassTransit;
+using Nybus.Utils;
 
 namespace Nybus.Configuration
 {
@@ -21,33 +22,38 @@ void SetCommandMessageHeaders<TCommand>(CommandMessage<TCommand> message, IPubli
     public class RabbitMqContextManager : IContextManager
     {
         public const string CorrelationIdKey = "nybus:CorrelationId";
+        public const string MessageSentKey = "nybus:MessageSent";
 
         public EventMessage<TEvent> CreateEventMessage<TEvent>(IConsumeContext<TEvent> context) where TEvent : class, IEvent
         {
             return new EventMessage<TEvent>
             {
                 CorrelationId = ExtractCorrelationId(context.Headers),
-                Event = context.Message
+                Event = context.Message,
+                SentOn = ExtractMessageSentTime(context.Headers)
             };
         }
 
         public void SetEventMessageHeaders<TEvent>(EventMessage<TEvent> message, IPublishContext<TEvent> context) where TEvent : class, IEvent
         {
             PersistCorrelationId(context, message.CorrelationId);
+            PersistMessageSentTime(context, message.SentOn);
         }
 
         public CommandMessage<TCommand> CreateCommandMessage<TCommand>(IConsumeContext<TCommand> context) where TCommand : class, ICommand
         {
             return new CommandMessage<TCommand>
             {
                 Command = context.Message,
-                CorrelationId = ExtractCorrelationId(context.Headers)
+                CorrelationId = ExtractCorrelationId(context.Headers),
+                SentOn = ExtractMessageSentTime(context.Headers)
             };
         }
 
         public void SetCommandMessageHeaders<TCommand>(CommandMessage<TCommand> message, IPublishContext<TCommand> context) where TCommand : class, ICommand
         {
             PersistCorrelationId(context, message.CorrelationId);
+            PersistMessageSentTime(context, message.SentOn);
         }
 
         private Guid ExtractCorrelationId(IMessageHeaders messageHeaders)
@@ -59,5 +65,15 @@ private void PersistCorrelationId(ISendContext sendContext, Guid correlationId)
         {
             sendContext.SetHeader(CorrelationIdKey, correlationId.ToString("D"));
         }
+
+        private DateTimeOffset ExtractMessageSentTime(IMessageHeaders messageHeaders)
+        {
+            return DateTimeOffset.Parse(messageHeaders[MessageSentKey]);
+        }
+
+        private void PersistMessageSentTime(ISendContext sendContext, DateTimeOffset sentTime)
+        {
+            sendContext.SetHeader(MessageSentKey, sentTime.ToString("O"));
+        }
     }
 }
diff --git a/tests/Tests.MassTransit/Configuration/RabbitMqContextManagerTests.cs b/tests/Tests.MassTransit/Configuration/RabbitMqContextManagerTests.cs
@@ -20,6 +20,8 @@ public class RabbitMqContextManagerTests
 
         private Mock<IPublishContext<TestCommand>> mockCommandPublishContext;
 
+        private DateTimeOffset now;
+
         [SetUp]
         public void Initialize()
         {
@@ -32,6 +34,8 @@ public void Initialize()
             mockCommandConsumeContext = new Mock<IConsumeContext<TestCommand>>();
 
             mockCommandPublishContext = new Mock<IPublishContext<TestCommand>>();
+
+            now = fixture.Create<DateTimeOffset>();
         }
 
         private RabbitMqContextManager CreateSystemUnderTest()
@@ -50,6 +54,7 @@ public void CreateEventMessage_retrieves_CorrelationId_from_context()
 
             mockEventConsumeContext.SetupGet(p => p.Message).Returns(body);
             mockEventConsumeContext.Setup(p => p.Headers[RabbitMqContextManager.CorrelationIdKey]).Returns(correlationId.ToString());
+            mockEventConsumeContext.Setup(p => p.Headers[RabbitMqContextManager.MessageSentKey]).Returns(now.ToString("O"));
 
             var message = sut.CreateEventMessage(mockEventConsumeContext.Object);
 
@@ -67,6 +72,7 @@ public void CreateEventMessage_retrieves_Event_from_context()
 
             mockEventConsumeContext.SetupGet(p => p.Message).Returns(body);
             mockEventConsumeContext.Setup(p => p.Headers[RabbitMqContextManager.CorrelationIdKey]).Returns(correlationId.ToString());
+            mockEventConsumeContext.Setup(p => p.Headers[RabbitMqContextManager.MessageSentKey]).Returns(now.ToString("O"));
 
             var message = sut.CreateEventMessage(mockEventConsumeContext.Object);
 
@@ -83,6 +89,7 @@ public void SetEventMessageHeaders_sets_correlationId_to_context()
             sut.SetEventMessageHeaders(message, mockEventPublishContext.Object);
 
             mockEventPublishContext.Verify(p => p.SetHeader(RabbitMqContextManager.CorrelationIdKey, message.CorrelationId.ToString("D")));
+            mockEventPublishContext.Verify(p => p.SetHeader(RabbitMqContextManager.MessageSentKey, message.SentOn.ToString("O")));
         }
 
         [Test]
@@ -96,10 +103,12 @@ public void CreateCommandMessage_retrieves_CorrelationId_from_context()
 
             mockCommandConsumeContext.SetupGet(p => p.Message).Returns(body);
             mockCommandConsumeContext.Setup(p => p.Headers[RabbitMqContextManager.CorrelationIdKey]).Returns(correlationId.ToString());
+            mockCommandConsumeContext.Setup(p => p.Headers[RabbitMqContextManager.MessageSentKey]).Returns(now.ToString("O"));
 
             var message = sut.CreateCommandMessage(mockCommandConsumeContext.Object);
 
             Assert.That(message.CorrelationId, Is.EqualTo(correlationId));
+            Assert.That(message.SentOn, Is.EqualTo(now));
         }
 
         [Test]
@@ -113,10 +122,12 @@ public void CreateCommandMessage_retrieves_Command_from_context()
 
             mockCommandConsumeContext.SetupGet(p => p.Message).Returns(body);
             mockCommandConsumeContext.Setup(p => p.Headers[RabbitMqContextManager.CorrelationIdKey]).Returns(correlationId.ToString());
+            mockCommandConsumeContext.Setup(p => p.Headers[RabbitMqContextManager.MessageSentKey]).Returns(now.ToString("O"));
 
             var message = sut.CreateCommandMessage(mockCommandConsumeContext.Object);
 
             Assert.That(message.Command, Is.SameAs(body));
+            Assert.That(message.SentOn, Is.EqualTo(now));
         }
 
         [Test]
@@ -129,6 +140,7 @@ public void SetCommandMessageHeaders_sets_correlationId_to_context()
             sut.SetCommandMessageHeaders(message, mockCommandPublishContext.Object);
 
             mockCommandPublishContext.Verify(p => p.SetHeader(RabbitMqContextManager.CorrelationIdKey, message.CorrelationId.ToString("D")));
+            mockCommandPublishContext.Verify(p => p.SetHeader(RabbitMqContextManager.MessageSentKey, message.SentOn.ToString("O")));
         }