Proper EnvelopeOrigin

akka · Oct 19, 2023 · aa314c9 · aa314c9
1 parent fd11c0c
commit aa314c9
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 4 deletions.
diff --git a/core/src/main/scala/akka/persistence/r2dbc/internal/EnvelopeOrigin.scala b/core/src/main/scala/akka/persistence/r2dbc/internal/EnvelopeOrigin.scala
@@ -15,6 +15,7 @@ import akka.persistence.query.typed.EventEnvelope
   val SourceQuery = ""
   val SourceBacktracking = "BT"
   val SourcePubSub = "PS"
+  val SourceSnapshot = "SN"
 
   def fromQuery(env: EventEnvelope[_]): Boolean =
     env.source == SourceQuery
@@ -28,6 +29,9 @@ import akka.persistence.query.typed.EventEnvelope
   def fromPubSub(env: EventEnvelope[_]): Boolean =
     env.source == SourcePubSub
 
+  def fromSnapshot(env: EventEnvelope[_]): Boolean =
+    env.source == SourceSnapshot
+
   def isFilteredEvent(env: Any): Boolean =
     env match {
       case e: EventEnvelope[_] => e.filtered

diff --git a/core/src/main/scala/akka/persistence/r2dbc/query/scaladsl/R2dbcReadJournal.scala b/core/src/main/scala/akka/persistence/r2dbc/query/scaladsl/R2dbcReadJournal.scala
@@ -162,8 +162,7 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
       row.entityType,
       row.slice,
       filtered = false,
-      // FIXME move constant to Akka/EnvelopeOrigin
-      source = "SN",
+      source = EnvelopeOrigin.SourceSnapshot,
       tags = row.tags)
   }
 

diff --git a/...st/scala/akka/persistence/r2dbc/query/EventsByPersistenceIdStartingFromSnapshotSpec.scala b/...st/scala/akka/persistence/r2dbc/query/EventsByPersistenceIdStartingFromSnapshotSpec.scala
@@ -5,7 +5,6 @@
 package akka.persistence.r2dbc.query
 
 import java.time.Instant
-
 import akka.Done
 import akka.NotUsed
 import akka.actor.testkit.typed.scaladsl.LogCapturing
@@ -20,6 +19,7 @@ import akka.persistence.r2dbc.TestActors.Persister.PersistWithAck
 import akka.persistence.r2dbc.TestConfig
 import akka.persistence.r2dbc.TestData
 import akka.persistence.r2dbc.TestDbLifecycle
+import akka.persistence.r2dbc.internal.EnvelopeOrigin
 import akka.persistence.r2dbc.query.scaladsl.R2dbcReadJournal
 import akka.persistence.typed.PersistenceId
 import akka.stream.scaladsl.Source
@@ -126,7 +126,7 @@ class EventsByPersistenceIdStartingFromSnapshotSpec
 
         val evt17 = result.expectNext()
         evt17.event shouldBe expectedSnapshotEvent(17)
-        evt17.source should ===("SN")
+        EnvelopeOrigin.fromSnapshot(evt17) shouldBe true
         for (i <- 18 to 20) {
           result.expectNext().event shouldBe s"e-$i"
         }