Event envelope updates

src/Akka.Persistence.MongoDb.Tests/Akka.Persistence.MongoDb.Tests.csproj

@@ -13,7 +13,7 @@
     </PackageReference>
     <PackageReference Include="xunit" Version="$(XunitVersion)" />
     <PackageReference Include="Akka.Persistence.TCK" Version="$(AkkaVersion)" />
-    <PackageReference Include="FluentAssertions" Version="5.10.3" />
+    <PackageReference Include="FluentAssertions" Version="$(FluentAssertionsVersion)" />
     <PackageReference Include="Mongo2Go" Version="2.2.16" />
     <PackageReference Include="System.Net.NetworkInformation" Version="4.3.0" />
   </ItemGroup>

src/Akka.Persistence.MongoDb.Tests/MongoDbCurrentEventsByPersistenceIdsSpec.cs

@@ -16,7 +16,7 @@
 namespace Akka.Persistence.MongoDb.Tests
 {
     [Collection("MongoDbSpec")]
-    public class MongoDbCurrentEventsByPersistenceIdsSpec : Akka.Persistence.TCK.Query.CurrentEventsByPersistenceIdSpec, IClassFixture<DatabaseFixture>
+    public class MongoDbCurrentEventsByPersistenceIdsSpec : TCK.Query.CurrentEventsByPersistenceIdSpec, IClassFixture<DatabaseFixture>
     {
         public static readonly AtomicCounter Counter = new AtomicCounter(0);
         private readonly ITestOutputHelper _output;

src/Akka.Persistence.MongoDb/Journal/JournalEntry.cs

@@ -35,7 +35,6 @@ public class JournalEntry
         [BsonElement("Manifest")]
         public string Manifest { get; set; }
 
-
         [BsonElement("Ordering")]
         public BsonTimestamp Ordering { get; set; }
 

src/Akka.Persistence.MongoDb/Journal/MongoDbJournal.cs

@@ -302,7 +302,9 @@ protected override Task DeleteMessagesToAsync(string persistenceId, long toSeque
 
         private JournalEntry ToJournalEntry(IPersistentRepresentation message)
         {
+            //var timeStamp = DateTimeOffset.Now.ToUnixTimeMilliseconds();
             object payload = message.Payload;
+            //message = message.WithTimestamp(timeStamp);
             if (message.Payload is Tagged tagged)
             {
                 payload = tagged.Payload;
@@ -319,7 +321,7 @@ private JournalEntry ToJournalEntry(IPersistentRepresentation message)
                     Id = message.PersistenceId + "_" + message.SequenceNr,
                     //Ordering = _sequenceRepository.GetSequenceValue("journalentry"), 
                     Ordering = new BsonTimestamp(0), // Auto-populates with timestamp
-                    //Timestamp = new BsonTimestamp(0), 
+                    //Timestamp = new BsonTimestamp(timeStamp), 
                     IsDeleted = message.IsDeleted,
                     Payload = payload,
                     PersistenceId = message.PersistenceId,
@@ -340,7 +342,7 @@ private JournalEntry ToJournalEntry(IPersistentRepresentation message)
                 Id = message.PersistenceId + "_" + message.SequenceNr,
                 //Ordering = _sequenceRepository.GetSequenceValue("journalentry"), 
                 Ordering = new BsonTimestamp(0), // Auto-populates with timestamp
-                                                 //Timestamp = new BsonTimestamp(0), 
+                //Timestamp = new BsonTimestamp(timeStamp),
                 IsDeleted = message.IsDeleted,
                 Payload = binary,
                 PersistenceId = message.PersistenceId,
@@ -363,14 +365,17 @@ private Persistent ToPersistenceRepresentation(JournalEntry entry, IActorRef sen
                     entry.PersistenceId,
                     manifest,
                     entry.IsDeleted,
-                    sender);
+                    sender,
+                    timestamp: entry.Ordering.Timestamp);
             }
 
             var legacy = entry.SerializerId.HasValue || !string.IsNullOrEmpty(entry.Manifest);
             if (!legacy)
             {
                 var ser = _serialization.FindSerializerForType(typeof(Persistent));
-                return ser.FromBinary<Persistent>((byte[]) entry.Payload);
+                var serPersistent = ser.FromBinary<Persistent>((byte[]) entry.Payload);
+                serPersistent.WithTimestamp(entry.Ordering.Timestamp);
+                return serPersistent;
             }
 
             int? serializerId = null;
@@ -398,12 +403,12 @@ private Persistent ToPersistenceRepresentation(JournalEntry entry, IActorRef sen
                 if (deserialized is Persistent p)
                     return p;
 
-                return new Persistent(deserialized, entry.SequenceNr, entry.PersistenceId, entry.Manifest, entry.IsDeleted, sender);
+                return new Persistent(deserialized, entry.SequenceNr, entry.PersistenceId, entry.Manifest, entry.IsDeleted, sender, timestamp: entry.Ordering.Timestamp);
             }
             else // backwards compat for object serialization - Payload was already deserialized by BSON
             {
                 return new Persistent(entry.Payload, entry.SequenceNr, entry.PersistenceId, entry.Manifest,
-                    entry.IsDeleted, sender);
+                    entry.IsDeleted, sender, timestamp: entry.Ordering.Timestamp);
             }
 
         }

src/Akka.Persistence.MongoDb/Query/AllEventsPublisher.cs

@@ -113,6 +113,7 @@ protected bool Replaying(object message)
                         offset: new Sequence(replayed.Offset),
                         persistenceId: replayed.Persistent.PersistenceId,
                         sequenceNr: replayed.Persistent.SequenceNr,
+                        timestamp: replayed.Persistent.Timestamp,
                         @event: replayed.Persistent.Payload));
 
                     CurrentOffset = replayed.Offset;

src/Akka.Persistence.MongoDb/Query/EventByPersistenceIdPublisher.cs

@@ -131,6 +131,7 @@ protected Receive Replaying(int limit)
                             offset: new Sequence(seqNr),
                             persistenceId: PersistenceId,
                             sequenceNr: seqNr,
+                            timestamp: replayed.Persistent.Timestamp,
                             @event: replayed.Persistent.Payload));
                         CurrentSequenceNr = seqNr + 1;
                         Buffer.DeliverBuffer(TotalDemand);

src/Akka.Persistence.MongoDb/Query/EventsByTagPublisher.cs

@@ -125,6 +125,7 @@ protected Receive Replaying(int limit)
                             offset: new Sequence(replayed.Offset),
                             persistenceId: replayed.Persistent.PersistenceId,
                             sequenceNr: replayed.Persistent.SequenceNr,
+                            timestamp: replayed.Persistent.Timestamp,
                             @event: replayed.Persistent.Payload));
 
                         CurrentOffset = replayed.Offset;

src/common.props

@@ -1,20 +1,21 @@
 <Project>
   <PropertyGroup>
-    <Copyright>Copyright © 2013-2020 Akka.NET Project</Copyright>
+    <Copyright>Copyright © 2013-2021 Akka.NET Project</Copyright>
     <Authors>Akka.NET Contrib</Authors>
-    <VersionPrefix>1.4.1</VersionPrefix>
+    <VersionPrefix>1.4.14</VersionPrefix>
     <PackageIconUrl>http://getakka.net/images/akkalogo.png</PackageIconUrl>
     <PackageProjectUrl>https://github.com/akkadotnet/Akka.Persistence.MongoDB</PackageProjectUrl>
     <PackageLicenseUrl>https://github.com/akkadotnet/Akka.Persistence.MongoDB/blob/master/LICENSE.md</PackageLicenseUrl>
-    <PackageReleaseNotes>Bump Akka version to 1.4.1</PackageReleaseNotes>
+    <PackageReleaseNotes>Bump Akka version to 1.4.14</PackageReleaseNotes>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
     <Description>Akka Persistence journal and snapshot store backed by MongoDB database.</Description>
     <NoWarn>$(NoWarn);CS1591</NoWarn>
   </PropertyGroup>
   <PropertyGroup>
     <XunitVersion>2.4.1</XunitVersion>
     <TestSdkVersion>16.8.3</TestSdkVersion>
-    <AkkaVersion>1.4.12</AkkaVersion>
+    <AkkaVersion>1.4.14</AkkaVersion>
+    <FluentAssertionsVersion>4.14.0</FluentAssertionsVersion>
   </PropertyGroup>
   <!-- SourceLink support for all Akka.NET projects -->
   <ItemGroup>