Merge pull request #176 from akkadotnet/dev

v1.4.14 Release
akkadotnet · Jan 13, 2021 · 226c2aa · 226c2aa
2 parents aeaff6c + 481fa67
commit 226c2aa
Show file tree

Hide file tree

Showing 11 changed files with 57 additions and 27 deletions.
diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md
@@ -1,6 +1,3 @@
-#### 1.4.12 November 26 2020 ####
+#### 1.4.14 January 13 2021 ####
 
-* Bump Akka version to 1.4.12
-* Corrected `CurrentPersistentIds` query and `AllPersistentIds` queries to be more memory efficient and query entity ID data directly from Mongo
-* Introduced `AllEvents` and `CurrentEvents` query to read the entire MongoDb journal
-* Deprecated previous `GetMaxSeqNo` behavior - we no longer query the max sequence number directly from the journal AND the metadata collection. We only get that data directly from the metadata collection itself, which should make this query an O(1) operation rather than O(n)
+* Bump [Akka.NET version to 1.4.14](https://github.com/akkadotnet/akka.net/releases/tag/1.4.14), which adds the `Timestamp` to the `EventEnvelope` data structure - so it can be used for sorting / ordering after the fact in an Akka.Persistence.Query
diff --git a/src/Akka.Persistence.MongoDb.Tests/Akka.Persistence.MongoDb.Tests.csproj b/src/Akka.Persistence.MongoDb.Tests/Akka.Persistence.MongoDb.Tests.csproj
@@ -13,8 +13,8 @@
     </PackageReference>
     <PackageReference Include="xunit" Version="$(XunitVersion)" />
     <PackageReference Include="Akka.Persistence.TCK" Version="$(AkkaVersion)" />
-    <PackageReference Include="FluentAssertions" Version="5.10.3" />
-    <PackageReference Include="Mongo2Go" Version="2.2.14" />
+    <PackageReference Include="FluentAssertions" Version="$(FluentAssertionsVersion)" />
+    <PackageReference Include="Mongo2Go" Version="2.2.16" />
     <PackageReference Include="System.Net.NetworkInformation" Version="4.3.0" />
   </ItemGroup>
 

diff --git a/src/Akka.Persistence.MongoDb.Tests/Bug61FixSpec.cs b/src/Akka.Persistence.MongoDb.Tests/Bug61FixSpec.cs
@@ -79,7 +79,8 @@ public async Task Bug61_Events_Recovered_By_Id_Should_Match_Tag()
         public void Bug80_CurrentEventsByTag_should_Recover_until_end()
         {
             var actor = Sys.ActorOf(TagActor.Props("y"));
-            var msgCount = 1200;
+            //increased this to test for non-collision with the generated timestamps
+            var msgCount = 5000;
             actor.Tell(msgCount);
             ExpectMsg($"{msgCount}-done", TimeSpan.FromSeconds(20));
 
@@ -96,7 +97,8 @@ public void Bug80_CurrentEventsByTag_should_Recover_until_end()
         public void Bug80_AllEventsByTag_should_Recover_all_messages()
         {
             var actor = Sys.ActorOf(TagActor.Props("y"));
-            var msgCount = 1200;
+            //increased this to test for non-collision with the generated timestamps
+            var msgCount = 5000;
             actor.Tell(msgCount);
             ExpectMsg($"{msgCount}-done", TimeSpan.FromSeconds(20));
 
@@ -176,7 +178,7 @@ public object ToJournal(object evt)
         private static Config CreateSpecConfig(DatabaseFixture databaseFixture, int id)
         {
             var specString = @"
-                akka.test.single-expect-default = 3s
+                akka.test.single-expect-default = 10s
                 akka.persistence {
                     publish-plugin-commands = on
                     journal {

diff --git a/src/Akka.Persistence.MongoDb.Tests/MongoDbCurrentEventsByPersistenceIdsSpec.cs b/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;

diff --git a/src/Akka.Persistence.MongoDb/Akka.Persistence.MongoDb.csproj b/src/Akka.Persistence.MongoDb/Akka.Persistence.MongoDb.csproj
@@ -11,6 +11,6 @@
   <ItemGroup>
     <PackageReference Include="Akka.Persistence.Query" Version="$(AkkaVersion)" />
     <PackageReference Include="akka.streams" Version="$(AkkaVersion)" />
-    <PackageReference Include="MongoDB.Driver" Version="2.11.4" />
+    <PackageReference Include="MongoDB.Driver" Version="2.11.5" />
   </ItemGroup>
 </Project>
diff --git a/src/Akka.Persistence.MongoDb/Journal/JournalEntry.cs b/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; }
 

diff --git a/src/Akka.Persistence.MongoDb/Journal/MongoDbJournal.cs b/src/Akka.Persistence.MongoDb/Journal/MongoDbJournal.cs
@@ -317,9 +317,7 @@ private JournalEntry ToJournalEntry(IPersistentRepresentation message)
                 return new JournalEntry
                 {
                     Id = message.PersistenceId + "_" + message.SequenceNr,
-                    //Ordering = _sequenceRepository.GetSequenceValue("journalentry"), 
                     Ordering = new BsonTimestamp(0), // Auto-populates with timestamp
-                    //Timestamp = new BsonTimestamp(0), 
                     IsDeleted = message.IsDeleted,
                     Payload = payload,
                     PersistenceId = message.PersistenceId,
@@ -338,9 +336,7 @@ private JournalEntry ToJournalEntry(IPersistentRepresentation message)
             return new JournalEntry
             {
                 Id = message.PersistenceId + "_" + message.SequenceNr,
-                //Ordering = _sequenceRepository.GetSequenceValue("journalentry"), 
                 Ordering = new BsonTimestamp(0), // Auto-populates with timestamp
-                                                 //Timestamp = new BsonTimestamp(0), 
                 IsDeleted = message.IsDeleted,
                 Payload = binary,
                 PersistenceId = message.PersistenceId,
@@ -351,8 +347,25 @@ private JournalEntry ToJournalEntry(IPersistentRepresentation message)
             };
         }
 
+        private static long ToTicks(BsonTimestamp bson)
+        {
+
+
+            // BSON Timestamps are stored natively as Unix epoch seconds + an ordinal value
+
+            // need to use BsonTimestamp.Timestamp because the ordinal value doesn't actually have any
+            // bearing on the time - it's used to try to somewhat order the events that all occurred concurrently
+            // according to the MongoDb clock. No need to include that data in the EventEnvelope.Timestamp field
+            // which is used entirely for end-user purposes.
+            //
+            // See https://docs.mongodb.com/manual/reference/bson-types/#timestamps
+
+            return DateTimeOffset.FromUnixTimeSeconds(bson.Timestamp).Ticks;
+        }
+
         private Persistent ToPersistenceRepresentation(JournalEntry entry, IActorRef sender)
         {
+
             if (_settings.LegacySerialization)
             {
                 var manifest = string.IsNullOrEmpty(entry.Manifest) ? entry.Payload.GetType().TypeQualifiedName() : entry.Manifest;
@@ -363,14 +376,24 @@ private Persistent ToPersistenceRepresentation(JournalEntry entry, IActorRef sen
                     entry.PersistenceId,
                     manifest,
                     entry.IsDeleted,
-                    sender);
+                    sender,
+                    timestamp: ToTicks(entry.Ordering)); // MongoDb timestamps are stored as Unix Epoch
             }
 
             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 output = ser.FromBinary<Persistent>((byte[])entry.Payload);
+
+                // backwards compatibility for https://github.com/akkadotnet/akka.net/pull/4680
+                // it the timestamp is not defined in the binary payload
+                if (output.Timestamp == 0L)
+                {
+                    output = (Persistent)output.WithTimestamp(ToTicks(entry.Ordering));
+                }
+
+                return output;
             }
 
             int? serializerId = null;
@@ -396,14 +419,14 @@ private Persistent ToPersistenceRepresentation(JournalEntry entry, IActorRef sen
                 }
 
                 if (deserialized is Persistent p)
-                    return p;
+                    return (Persistent)p.WithTimestamp(ToTicks(entry.Ordering));
 
-                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: ToTicks(entry.Ordering));
             }
             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: ToTicks(entry.Ordering));
             }
 
         }

diff --git a/src/Akka.Persistence.MongoDb/Query/AllEventsPublisher.cs b/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;

diff --git a/src/Akka.Persistence.MongoDb/Query/EventByPersistenceIdPublisher.cs b/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);

diff --git a/src/Akka.Persistence.MongoDb/Query/EventsByTagPublisher.cs b/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;

diff --git a/src/common.props b/src/common.props
@@ -1,20 +1,26 @@
 <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
+      Corrected `CurrentPersistentIds` query and `AllPersistentIds` queries to be more memory efficient and query entity ID data directly from Mongo
+      Introduced `AllEvents` and `CurrentEvents` query to read the entire MongoDb journal
+      Deprecated previous `GetMaxSeqNo` behavior - we no longer query the max sequence number directly from the journal AND the metadata collection. We only get that data directly from the metadata collection itself, which should make this query an O(1) operation rather than O(n)
+    </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.0</TestSdkVersion>
-    <AkkaVersion>1.4.12</AkkaVersion>
+    <TestSdkVersion>16.8.3</TestSdkVersion>
+    <AkkaVersion>1.4.14</AkkaVersion>
+    <FluentAssertionsVersion>4.14.0</FluentAssertionsVersion>
   </PropertyGroup>
   <!-- SourceLink support for all Akka.NET projects -->
   <ItemGroup>