feat: Add CurrentEventsByPersistenceIdTypedQuery

core/src/main/scala/akka/persistence/dynamodb/query/javadsl/DynamoDBReadJournal.scala

@@ -20,6 +20,7 @@ import akka.persistence.dynamodb.query.scaladsl
 import akka.persistence.query.Offset
 import akka.persistence.query.javadsl._
 import akka.persistence.query.typed.EventEnvelope
+import akka.persistence.query.typed.javadsl.CurrentEventsByPersistenceIdTypedQuery
 import akka.persistence.query.typed.javadsl.CurrentEventsBySliceQuery
 import akka.persistence.query.typed.javadsl.CurrentEventsBySliceStartingFromSnapshotsQuery
 import akka.persistence.query.typed.javadsl.EventTimestampQuery
@@ -38,6 +39,7 @@ final class DynamoDBReadJournal(delegate: scaladsl.DynamoDBReadJournal)
     with EventsBySliceQuery
     with CurrentEventsBySliceStartingFromSnapshotsQuery
     with EventsBySliceStartingFromSnapshotsQuery
+    with CurrentEventsByPersistenceIdTypedQuery
     with EventTimestampQuery
     with LoadEventQuery {
 
@@ -135,9 +137,16 @@ final class DynamoDBReadJournal(delegate: scaladsl.DynamoDBReadJournal)
       transformSnapshot: java.util.function.Function[Snapshot, Event]): Source[EventEnvelope[Event], NotUsed] =
     delegate.eventsBySlicesStartingFromSnapshots(entityType, minSlice, maxSlice, offset, transformSnapshot(_)).asJava
 
+  override def currentEventsByPersistenceIdTyped[Event](
+      persistenceId: String,
+      fromSequenceNr: Long,
+      toSequenceNr: Long): Source[EventEnvelope[Event], NotUsed] =
+    delegate.currentEventsByPersistenceIdTyped(persistenceId, fromSequenceNr, toSequenceNr).asJava
+
   override def timestampOf(persistenceId: String, sequenceNr: Long): CompletionStage[Optional[Instant]] =
     delegate.timestampOf(persistenceId, sequenceNr).map(_.toJava)(ExecutionContext.parasitic).asJava
 
   override def loadEnvelope[Event](persistenceId: String, sequenceNr: Long): CompletionStage[EventEnvelope[Event]] =
     delegate.loadEnvelope[Event](persistenceId, sequenceNr).asJava
+
 }

core/src/main/scala/akka/persistence/dynamodb/query/scaladsl/DynamoDBReadJournal.scala

@@ -8,8 +8,10 @@ import java.time.Clock
 import java.time.Instant
 import java.time.{ Duration => JDuration }
 import java.util.concurrent.ConcurrentHashMap
+
 import scala.collection.mutable
 import scala.concurrent.Future
+
 import akka.NotUsed
 import akka.actor.ExtendedActorSystem
 import akka.actor.typed.pubsub.Topic
@@ -47,11 +49,13 @@ import akka.stream.scaladsl.Flow
 import akka.stream.scaladsl.Source
 import com.typesafe.config.Config
 import org.slf4j.LoggerFactory
-
 import scala.annotation.nowarn
 import java.util.UUID
+
 import scala.annotation.tailrec
 
+import akka.persistence.query.typed.scaladsl.CurrentEventsByPersistenceIdTypedQuery
+
 object DynamoDBReadJournal {
   val Identifier = "akka.persistence.dynamodb.query"
 }
@@ -62,6 +66,7 @@ final class DynamoDBReadJournal(system: ExtendedActorSystem, config: Config, cfg
     with EventsBySliceQuery
     with CurrentEventsBySliceStartingFromSnapshotsQuery
     with EventsBySliceStartingFromSnapshotsQuery
+    with CurrentEventsByPersistenceIdTypedQuery
     with EventTimestampQuery
     with LoadEventQuery {
 
@@ -217,6 +222,14 @@ final class DynamoDBReadJournal(system: ExtendedActorSystem, config: Config, cfg
     queryDao.eventsByPersistenceId(persistenceId, fromSequenceNr, toSequenceNr, includeDeleted)
   }
 
+  override def currentEventsByPersistenceIdTyped[Event](
+      persistenceId: String,
+      fromSequenceNr: Long,
+      toSequenceNr: Long): Source[EventEnvelope[Event], NotUsed] = {
+    internalCurrentEventsByPersistenceId(persistenceId, fromSequenceNr, toSequenceNr, includeDeleted = false)
+      .map(deserializeBySliceItem[Event])
+  }
+
   override def sliceForPersistenceId(persistenceId: String): Int = {
     persistenceExt.sliceForPersistenceId(persistenceId)
   }
@@ -613,4 +626,5 @@ final class DynamoDBReadJournal(system: ExtendedActorSystem, config: Config, cfg
             s"Event with persistenceId [$persistenceId] and sequenceNr [$sequenceNr] not found.")
       }
   }
+
 }

core/src/test/scala/akka/persistence/dynamodb/query/EventsByPersistenceIdSpec.scala

@@ -0,0 +1,106 @@
+/*
+ * Copyright (C) 2024 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.persistence.dynamodb.query
+
+import akka.Done
+import akka.actor.testkit.typed.scaladsl.LogCapturing
+import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import akka.actor.typed.ActorSystem
+import akka.persistence.dynamodb.TestActors
+import akka.persistence.dynamodb.TestActors.Persister
+import akka.persistence.dynamodb.TestActors.Persister.PersistWithAck
+import akka.persistence.dynamodb.TestConfig
+import akka.persistence.dynamodb.TestData
+import akka.persistence.dynamodb.TestDbLifecycle
+import akka.persistence.dynamodb.query.scaladsl.DynamoDBReadJournal
+import akka.persistence.query.PersistenceQuery
+import akka.persistence.typed.PersistenceId
+import akka.persistence.typed.internal.ReplicatedEventMetadata
+import akka.stream.scaladsl.Sink
+import com.typesafe.config.Config
+import org.scalatest.wordspec.AnyWordSpecLike
+
+object EventsByPersistenceIdSpec {
+
+  def config: Config = TestConfig.config
+}
+
+class EventsByPersistenceIdSpec
+    extends ScalaTestWithActorTestKit(EventsByPersistenceIdSpec.config)
+    with AnyWordSpecLike
+    with TestDbLifecycle
+    with TestData
+    with LogCapturing {
+
+  override def typedSystem: ActorSystem[_] = system
+
+  private val query =
+    PersistenceQuery(testKit.system).readJournalFor[DynamoDBReadJournal](DynamoDBReadJournal.Identifier)
+
+  "CurrentEventsByPersistenceIdTypedQuery" should {
+    "retrieve requested events" in {
+      val entityType = nextEntityType()
+      val pid = nextPersistenceId(entityType)
+      val persister = testKit.spawn(Persister(pid))
+      val probe = testKit.createTestProbe[Done]()
+      val events = (1 to 20).map { i =>
+        val payload = s"e-$i"
+        persister ! PersistWithAck(payload, probe.ref)
+        probe.expectMessage(Done)
+        payload
+      }
+
+      val result1 =
+        query.currentEventsByPersistenceIdTyped[String](pid.id, 0, Long.MaxValue).runWith(Sink.seq).futureValue
+      result1.map(_.event) shouldBe events
+
+      val result2 =
+        query.currentEventsByPersistenceIdTyped[String](pid.id, 1, 5).runWith(Sink.seq).futureValue
+      result2.map(_.event) shouldBe events.take(5)
+
+      val result3 =
+        query.currentEventsByPersistenceIdTyped[String](pid.id, 3, 7).runWith(Sink.seq).futureValue
+      result3.head.event shouldBe "e-3"
+      result3.last.event shouldBe "e-7"
+      result3.map(_.event) shouldBe events.slice(2, 7)
+
+      query
+        .currentEventsByPersistenceIdTyped[String](nextPersistenceId(entityType).id, 1, 10)
+        .runWith(Sink.seq)
+        .futureValue
+        .size shouldBe 0
+    }
+
+    "include metadata" in {
+      val entityType = nextEntityType()
+      val entityId = "entity-1"
+      val persister = testKit.spawn(TestActors.replicatedEventSourcedPersister(entityType, entityId))
+      val probe = testKit.createTestProbe[Done]()
+      persister ! Persister.PersistWithAck("e-1", probe.ref)
+      probe.expectMessage(Done)
+      persister ! Persister.PersistWithAck("e-2", probe.ref)
+      probe.expectMessage(Done)
+
+      val result1 =
+        query
+          .currentEventsByPersistenceIdTyped[String](PersistenceId(entityType, entityId).id, 0, Long.MaxValue)
+          .runWith(Sink.seq)
+          .futureValue
+
+      val env1 = result1.head
+      env1.event shouldBe "e-1"
+      val meta1 = env1.eventMetadata.get.asInstanceOf[ReplicatedEventMetadata]
+      meta1.originReplica.id shouldBe "dc-1"
+      meta1.originSequenceNr shouldBe 1L
+
+      val env2 = result1(1)
+      env2.event shouldBe "e-2"
+      val meta2 = env2.eventMetadata.get.asInstanceOf[ReplicatedEventMetadata]
+      meta2.originReplica.id shouldBe "dc-1"
+      meta2.originSequenceNr shouldBe 2L
+    }
+  }
+
+}