Expose Tags property in EventEnvelope (#6862)

* Expose `Tags` property in `EventEnvelope` * update API approval list * Change `Tags` default value from `null` to empty array * Update API approval list
akkadotnet · Aug 1, 2023 · 147d039 · 147d039
1 parent 4df9a60
commit 147d039
Show file tree

Hide file tree

Showing 15 changed files with 137 additions and 49 deletions.
diff --git a/...ntrib/persistence/Akka.Persistence.Query.InMemory.Tests/InMemoryCurrentEventsByTagSpec.cs b/...ntrib/persistence/Akka.Persistence.Query.InMemory.Tests/InMemoryCurrentEventsByTagSpec.cs
@@ -32,5 +32,7 @@ public InMemoryCurrentEventsByTagSpec(ITestOutputHelper output) :
         {
             ReadJournal = Sys.ReadJournalFor<InMemoryReadJournal>(InMemoryReadJournal.Identifier);
         }
+
+        protected override bool SupportsTagsInEventEnvelope => true;
     }
 }
diff --git a/src/contrib/persistence/Akka.Persistence.Query.InMemory.Tests/InMemoryEventsByTagSpec.cs b/src/contrib/persistence/Akka.Persistence.Query.InMemory.Tests/InMemoryEventsByTagSpec.cs
@@ -32,5 +32,7 @@ public InMemoryEventsByTagSpec(ITestOutputHelper output) :
         {
             ReadJournal = Sys.ReadJournalFor<InMemoryReadJournal>(InMemoryReadJournal.Identifier);
         }
+
+        protected override bool SupportsTagsInEventEnvelope => true;
     }
 }
diff --git a/src/contrib/persistence/Akka.Persistence.Query.InMemory/AllEventsPublisher.cs b/src/contrib/persistence/Akka.Persistence.Query.InMemory/AllEventsPublisher.cs
@@ -108,12 +108,14 @@ protected bool Replaying( object message)
                     if (replayed.Offset > ToOffset)
                         return true;
 
+                    // NOTES: tags is empty because tags are not retrieved from the database query (as of this writing)
                     Buffer.Add(new EventEnvelope(
                         offset: new Sequence(replayed.Offset),
                         persistenceId: replayed.Persistent.PersistenceId,
                         sequenceNr: replayed.Persistent.SequenceNr,
                         @event: replayed.Persistent.Payload,
-                        timestamp: replayed.Persistent.Timestamp));
+                        timestamp: replayed.Persistent.Timestamp,
+                        tags: Array.Empty<string>()));
 
                     CurrentOffset = replayed.Offset + 1;
                     Buffer.DeliverBuffer(TotalDemand);

diff --git a/src/contrib/persistence/Akka.Persistence.Query.InMemory/EventsByPersistenceIdPublisher.cs b/src/contrib/persistence/Akka.Persistence.Query.InMemory/EventsByPersistenceIdPublisher.cs
@@ -121,12 +121,14 @@ protected Receive Replaying(int limit)
                 {
                     case ReplayedMessage replayed:
                         var seqNr = replayed.Persistent.SequenceNr;
+                        // NOTES: tags is empty because tags are not retrieved from the database query (as of this writing)
                         Buffer.Add(new EventEnvelope(
                             offset: new Sequence(seqNr),
                             persistenceId: PersistenceId,
                             sequenceNr: seqNr,
                             @event: replayed.Persistent.Payload,
-                            timestamp: replayed.Persistent.Timestamp));
+                            timestamp: replayed.Persistent.Timestamp,
+                            tags: Array.Empty<string>()));
                         CurrentSequenceNr = seqNr + 1;
                         Buffer.DeliverBuffer(TotalDemand);
                         return true;

diff --git a/src/contrib/persistence/Akka.Persistence.Query.InMemory/EventsByTagPublisher.cs b/src/contrib/persistence/Akka.Persistence.Query.InMemory/EventsByTagPublisher.cs
@@ -120,7 +120,8 @@ protected Receive Replaying(int limit)
                             persistenceId: replayed.Persistent.PersistenceId,
                             sequenceNr: replayed.Persistent.SequenceNr,
                             @event: replayed.Persistent.Payload,
-                            timestamp: replayed.Persistent.Timestamp));
+                            timestamp: replayed.Persistent.Timestamp,
+                            tags: new [] { replayed.Tag }));
 
                         CurrentOffset = replayed.Offset + 1;
                         Buffer.DeliverBuffer(TotalDemand);

diff --git a/src/contrib/persistence/Akka.Persistence.Query.Sql/AllEventsPublisher.cs b/src/contrib/persistence/Akka.Persistence.Query.Sql/AllEventsPublisher.cs
@@ -139,12 +139,14 @@ protected bool Replaying( object message )
                     if (replayed.Offset > ToOffset)
                         return true;
 
+                    // NOTES: tags is empty because tags are not retrieved from the database query (as of this writing)
                     Buffer.Add(new EventEnvelope(
                         offset: new Sequence(replayed.Offset),
                         persistenceId: replayed.Persistent.PersistenceId,
                         sequenceNr: replayed.Persistent.SequenceNr,
                         @event: replayed.Persistent.Payload,
-                        timestamp: replayed.Persistent.Timestamp));
+                        timestamp: replayed.Persistent.Timestamp,
+                        tags: Array.Empty<string>()));
 
                     CurrentOffset = replayed.Offset;
                     Buffer.DeliverBuffer(TotalDemand);

diff --git a/src/contrib/persistence/Akka.Persistence.Query.Sql/EventsByPersistenceIdPublisher.cs b/src/contrib/persistence/Akka.Persistence.Query.Sql/EventsByPersistenceIdPublisher.cs
@@ -153,12 +153,14 @@ protected bool Replaying(object message)
             {
                 case ReplayedMessage replayed:
                     var seqNr = replayed.Persistent.SequenceNr;
+                    // NOTES: tags is empty because tags are not retrieved from the database query (as of this writing)
                     Buffer.Add(new EventEnvelope(
                         offset: new Sequence(seqNr),
                         persistenceId: PersistenceId,
                         sequenceNr: seqNr,
                         @event: replayed.Persistent.Payload,
-                        timestamp: replayed.Persistent.Timestamp));
+                        timestamp: replayed.Persistent.Timestamp,
+                        tags: Array.Empty<string>()));
                     CurrentSequenceNr = seqNr + 1;
                     Buffer.DeliverBuffer(TotalDemand);
                     return true;

diff --git a/src/contrib/persistence/Akka.Persistence.Query.Sql/EventsByTagPublisher.cs b/src/contrib/persistence/Akka.Persistence.Query.Sql/EventsByTagPublisher.cs
@@ -153,7 +153,8 @@ protected bool Replaying(object message)
                         persistenceId: replayed.Persistent.PersistenceId,
                         sequenceNr: replayed.Persistent.SequenceNr,
                         @event: replayed.Persistent.Payload,
-                        timestamp: replayed.Persistent.Timestamp));
+                        timestamp: replayed.Persistent.Timestamp,
+                        tags: new [] { replayed.Tag }));
 
                     CurrentOffset = replayed.Offset;
                     Buffer.DeliverBuffer(TotalDemand);

diff --git a/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Query/SqliteCurrentEventsByTagSpec.cs b/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Query/SqliteCurrentEventsByTagSpec.cs
@@ -44,5 +44,7 @@ public SqliteCurrentEventsByTagSpec(ITestOutputHelper output) : base(Config(Coun
         {
             ReadJournal = Sys.ReadJournalFor<SqlReadJournal>(SqlReadJournal.Identifier);
         }
+
+        protected override bool SupportsTagsInEventEnvelope => true;
     }
 }
diff --git a/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Query/SqliteEventsByTagSpec.cs b/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Query/SqliteEventsByTagSpec.cs
@@ -44,5 +44,7 @@ public SqliteEventsByTagSpec(ITestOutputHelper output) : base(Config(Counter.Get
         {
             ReadJournal = Sys.ReadJournalFor<SqlReadJournal>(SqlReadJournal.Identifier);
         }
+
+        protected override bool SupportsTagsInEventEnvelope => true;
     }
 }
diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApprovePersistenceQuery.DotNet.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApprovePersistenceQuery.DotNet.verified.txt
@@ -4,15 +4,19 @@
 [assembly: System.Runtime.Versioning.TargetFrameworkAttribute(".NETCoreApp,Version=v6.0", FrameworkDisplayName=".NET 6.0")]
 namespace Akka.Persistence.Query
 {
+    [System.Runtime.CompilerServices.NullableAttribute(0)]
     public sealed class EventEnvelope : System.IEquatable<Akka.Persistence.Query.EventEnvelope>
     {
-        [System.ObsoleteAttribute("For binary compatibility with previous releases")]
+        [System.ObsoleteAttribute("For binary compatibility with previous releases. Since 1.4.14")]
         public EventEnvelope(Akka.Persistence.Query.Offset offset, string persistenceId, long sequenceNr, object @event) { }
+        [System.ObsoleteAttribute("For binary compatibility with previous releases. Since 1.5.11")]
         public EventEnvelope(Akka.Persistence.Query.Offset offset, string persistenceId, long sequenceNr, object @event, long timestamp) { }
+        public EventEnvelope(Akka.Persistence.Query.Offset offset, string persistenceId, long sequenceNr, object @event, long timestamp, string[] tags) { }
         public object Event { get; }
         public Akka.Persistence.Query.Offset Offset { get; }
         public string PersistenceId { get; }
         public long SequenceNr { get; }
+        public string[] Tags { get; }
         public long Timestamp { get; }
         public bool Equals(Akka.Persistence.Query.EventEnvelope other) { }
         public override bool Equals(object obj) { }

diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApprovePersistenceQuery.Net.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApprovePersistenceQuery.Net.verified.txt
@@ -4,15 +4,19 @@
 [assembly: System.Runtime.Versioning.TargetFrameworkAttribute(".NETStandard,Version=v2.0", FrameworkDisplayName=".NET Standard 2.0")]
 namespace Akka.Persistence.Query
 {
+    [System.Runtime.CompilerServices.NullableAttribute(0)]
     public sealed class EventEnvelope : System.IEquatable<Akka.Persistence.Query.EventEnvelope>
     {
-        [System.ObsoleteAttribute("For binary compatibility with previous releases")]
+        [System.ObsoleteAttribute("For binary compatibility with previous releases. Since 1.4.14")]
         public EventEnvelope(Akka.Persistence.Query.Offset offset, string persistenceId, long sequenceNr, object @event) { }
+        [System.ObsoleteAttribute("For binary compatibility with previous releases. Since 1.5.11")]
         public EventEnvelope(Akka.Persistence.Query.Offset offset, string persistenceId, long sequenceNr, object @event, long timestamp) { }
+        public EventEnvelope(Akka.Persistence.Query.Offset offset, string persistenceId, long sequenceNr, object @event, long timestamp, string[] tags) { }
         public object Event { get; }
         public Akka.Persistence.Query.Offset Offset { get; }
         public string PersistenceId { get; }
         public long SequenceNr { get; }
+        public string[] Tags { get; }
         public long Timestamp { get; }
         public bool Equals(Akka.Persistence.Query.EventEnvelope other) { }
         public override bool Equals(object obj) { }

diff --git a/src/core/Akka.Persistence.Query/EventEnvelope.cs b/src/core/Akka.Persistence.Query/EventEnvelope.cs
@@ -7,6 +7,7 @@
 
 using System;
 
+#nullable enable
 namespace Akka.Persistence.Query
 {
     /// <summary>
@@ -23,21 +24,27 @@ public sealed class EventEnvelope : IEquatable<EventEnvelope>
         /// <summary>
         /// Initializes a new instance of the <see cref="EventEnvelope"/> class.
         /// </summary>
-        [Obsolete("For binary compatibility with previous releases")]
+        [Obsolete("For binary compatibility with previous releases. Since 1.4.14")]
         public EventEnvelope(Offset offset, string persistenceId, long sequenceNr, object @event)
-            : this(offset, persistenceId, sequenceNr, @event, 0L)
+            : this(offset, persistenceId, sequenceNr, @event, 0L, Array.Empty<string>())
         { }
 
         /// <summary>
         /// Initializes a new instance of the <see cref="EventEnvelope"/> class.
         /// </summary>
+        [Obsolete("For binary compatibility with previous releases. Since 1.5.11")]
         public EventEnvelope(Offset offset, string persistenceId, long sequenceNr, object @event, long timestamp)
+            : this(offset, persistenceId, sequenceNr, @event, timestamp, Array.Empty<string>())
+        { }        
+
+        public EventEnvelope(Offset offset, string persistenceId, long sequenceNr, object @event, long timestamp, string[] tags)
         {
             Offset = offset;
             PersistenceId = persistenceId;
             SequenceNr = sequenceNr;
             Event = @event;
             Timestamp = timestamp;
+            Tags = tags ?? Array.Empty<string>();
         }        
 
         public Offset Offset { get; }
@@ -50,28 +57,30 @@ public EventEnvelope(Offset offset, string persistenceId, long sequenceNr, objec
 
         public long Timestamp { get; }
 
-        public bool Equals(EventEnvelope other)
+        public string[] Tags { get; }
+
+        public bool Equals(EventEnvelope? other)
         {
             if (ReferenceEquals(this, other)) return true;
             if (ReferenceEquals(other, null)) return false;
 
-            // timestamp not included in Equals for backwards compatibility
+            // Timestamp and Tags not included in Equals for backwards compatibility
             return Offset == other.Offset
                    && PersistenceId == other.PersistenceId
                    && SequenceNr == other.SequenceNr
                    && Equals(Event, other.Event);
         }
 
-        public override bool Equals(object obj) => obj is EventEnvelope evt && Equals(evt);
+        public override bool Equals(object? obj) => obj is EventEnvelope evt && Equals(evt);
 
         public override int GetHashCode()
         {
             unchecked
             {
                 var hashCode = Offset.GetHashCode();
-                hashCode = (hashCode*397) ^ (PersistenceId != null ? PersistenceId.GetHashCode() : 0);
+                hashCode = (hashCode*397) ^ (PersistenceId?.GetHashCode() ?? 0);
                 hashCode = (hashCode*397) ^ SequenceNr.GetHashCode();
-                hashCode = (hashCode*397) ^ (Event != null ? Event.GetHashCode() : 0);
+                hashCode = (hashCode*397) ^ (Event?.GetHashCode() ?? 0);
                 return hashCode;
             }
         }