chore: minor cleanup

core/src/main/scala/akka/persistence/r2dbc/query/scaladsl/R2dbcReadJournal.scala

@@ -113,25 +113,7 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
     row.payload.map(payload => serialization.deserialize(payload, row.serId, row.serManifest).get.asInstanceOf[Event])
 
   private val _bySlice: BySliceQuery[SerializedJournalRow, EventEnvelope[Any]] = {
-    val createEnvelope: (TimestampOffset, SerializedJournalRow) => EventEnvelope[Any] = (offset, row) => {
-      val event = deserializePayload(row)
-      val metadata = row.metadata.map(meta => serialization.deserialize(meta.payload, meta.serId, meta.serManifest).get)
-      val source = if (event.isDefined) EnvelopeOrigin.SourceQuery else EnvelopeOrigin.SourceBacktracking
-      val filtered = row.serId == filteredPayloadSerId
-
-      new EventEnvelope(
-        offset,
-        row.persistenceId,
-        row.seqNr,
-        if (filtered) None else event,
-        row.dbTimestamp.toEpochMilli,
-        metadata,
-        row.entityType,
-        row.slice,
-        filtered,
-        source,
-        tags = row.tags)
-    }
+    val createEnvelope: (TimestampOffset, SerializedJournalRow) => EventEnvelope[Any] = createEventEnvelope
 
     val extractOffset: EventEnvelope[Any] => TimestampOffset = env => env.offset.asInstanceOf[TimestampOffset]
 
@@ -141,6 +123,43 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
   private def bySlice[Event]: BySliceQuery[SerializedJournalRow, EventEnvelope[Event]] =
     _bySlice.asInstanceOf[BySliceQuery[SerializedJournalRow, EventEnvelope[Event]]]
 
+  private def deserializeBySliceRow[Event](row: SerializedJournalRow): EventEnvelope[Event] = {
+    val offset = TimestampOffset(row.dbTimestamp, row.readDbTimestamp, Map(row.persistenceId -> row.seqNr))
+    createEventEnvelope(offset, row)
+  }
+
+  private def createEventEnvelope[Event](offset: TimestampOffset, row: SerializedJournalRow): EventEnvelope[Event] = {
+    val event = deserializePayload(row)
+    val metadata = row.metadata.map(meta => serialization.deserialize(meta.payload, meta.serId, meta.serManifest).get)
+    val source = if (event.isDefined) EnvelopeOrigin.SourceQuery else EnvelopeOrigin.SourceBacktracking
+    val filtered = row.serId == filteredPayloadSerId
+
+    new EventEnvelope(
+      offset,
+      row.persistenceId,
+      row.seqNr,
+      if (filtered) None else event,
+      row.dbTimestamp.toEpochMilli,
+      metadata,
+      row.entityType,
+      row.slice,
+      filtered,
+      source,
+      tags = row.tags)
+  }
+
+  private def deserializeRow(row: SerializedJournalRow): ClassicEventEnvelope = {
+    val event = deserializePayload(row)
+    // note that it's not possible to filter out FilteredPayload here
+    val offset = TimestampOffset(row.dbTimestamp, row.readDbTimestamp, Map(row.persistenceId -> row.seqNr))
+    val envelope = ClassicEventEnvelope(offset, row.persistenceId, row.seqNr, event.get, row.dbTimestamp.toEpochMilli)
+    row.metadata match {
+      case None => envelope
+      case Some(meta) =>
+        envelope.withMetadata(serialization.deserialize(meta.payload, meta.serId, meta.serManifest).get)
+    }
+  }
+
   private def snapshotsBySlice[Snapshot, Event](
       transformSnapshot: Snapshot => Event): BySliceQuery[SerializedSnapshotRow, EventEnvelope[Event]] = {
     val createEnvelope: (TimestampOffset, SerializedSnapshotRow) => EventEnvelope[Event] =
@@ -753,39 +772,6 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
       .mapMaterializedValue(_ => NotUsed)
   }
 
-  private def deserializeBySliceRow[Event](row: SerializedJournalRow): EventEnvelope[Event] = {
-    val event = deserializePayload[Event](row)
-    val offset = TimestampOffset(row.dbTimestamp, row.readDbTimestamp, Map(row.persistenceId -> row.seqNr))
-    val metadata = row.metadata.map(meta => serialization.deserialize(meta.payload, meta.serId, meta.serManifest).get)
-    val source = if (event.isDefined) EnvelopeOrigin.SourceQuery else EnvelopeOrigin.SourceBacktracking
-    val filtered = row.serId == filteredPayloadSerId
-
-    new EventEnvelope(
-      offset,
-      row.persistenceId,
-      row.seqNr,
-      if (filtered) None else event,
-      row.dbTimestamp.toEpochMilli,
-      metadata,
-      row.entityType,
-      row.slice,
-      filtered,
-      source,
-      tags = row.tags)
-  }
-
-  private def deserializeRow(row: SerializedJournalRow): ClassicEventEnvelope = {
-    val event = deserializePayload(row)
-    // note that it's not possible to filter out FilteredPayload here
-    val offset = TimestampOffset(row.dbTimestamp, row.readDbTimestamp, Map(row.persistenceId -> row.seqNr))
-    val envelope = ClassicEventEnvelope(offset, row.persistenceId, row.seqNr, event.get, row.dbTimestamp.toEpochMilli)
-    row.metadata match {
-      case None => envelope
-      case Some(meta) =>
-        envelope.withMetadata(serialization.deserialize(meta.payload, meta.serId, meta.serManifest).get)
-    }
-  }
-
   override def currentPersistenceIds(afterId: Option[String], limit: Long): Source[String, NotUsed] =
     queryDao.persistenceIds(afterId, limit)
 

core/src/test/scala/akka/persistence/r2dbc/PayloadSpec.scala

@@ -6,15 +6,17 @@ package akka.persistence.r2dbc
 
 import java.nio.charset.StandardCharsets.UTF_8
 
+import com.typesafe.config.ConfigFactory
+import org.scalatest.wordspec.AnyWordSpecLike
+
 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.r2dbc.TestActors.DurableStatePersister
 import akka.persistence.r2dbc.TestActors.Persister
+import akka.persistence.r2dbc.internal.codec.PayloadCodec
 import akka.persistence.r2dbc.internal.codec.PayloadCodec.RichRow
-import com.typesafe.config.ConfigFactory
-import org.scalatest.wordspec.AnyWordSpecLike
 
 /**
  * The purpose of this test is to verify JSONB payloads, but it can also be run with ordinary BYTEA payloads. To test
@@ -59,7 +61,6 @@ class PayloadSpec
   import PayloadSpec._
 
   override def typedSystem: ActorSystem[_] = system
-  private val settings = R2dbcSettings(system.settings.config.getConfig("akka.persistence.r2dbc"))
 
   private def testJournalPersister(persistenceId: String, msg: Any): Unit = {
     val probe = createTestProbe[Any]()
@@ -88,7 +89,7 @@ class PayloadSpec
   }
 
   private def selectJournalRow(persistenceId: String): TestRow = {
-    import settings.codecSettings.JournalImplicits.journalPayloadCodec
+    implicit val codec: PayloadCodec = settings.codecSettings.JournalImplicits.journalPayloadCodec
     val slice = persistenceExt.sliceForPersistenceId(persistenceId)
 
     r2dbcExecutor(slice)
@@ -108,7 +109,7 @@ class PayloadSpec
   }
 
   private def selectSnapshotRow(persistenceId: String): TestRow = {
-    import settings.codecSettings.SnapshotImplicits.snapshotPayloadCodec
+    implicit val codec: PayloadCodec = settings.codecSettings.SnapshotImplicits.snapshotPayloadCodec
     val slice = persistenceExt.sliceForPersistenceId(persistenceId)
 
     r2dbcExecutor(slice)
@@ -128,7 +129,7 @@ class PayloadSpec
   }
 
   private def selectDurableStateRow(persistenceId: String): TestRow = {
-    import settings.codecSettings.DurableStateImplicits.durableStatePayloadCodec
+    implicit val codec: PayloadCodec = settings.codecSettings.DurableStateImplicits.durableStatePayloadCodec
     val slice = persistenceExt.sliceForPersistenceId(persistenceId)
 
     r2dbcExecutor(slice)

core/src/test/scala/akka/persistence/r2dbc/TestDbLifecycle.scala

@@ -28,14 +28,14 @@ trait TestDbLifecycle extends BeforeAndAfterAll { this: Suite =>
 
   def testConfigPath: String = "akka.persistence.r2dbc"
 
-  lazy val r2dbcSettings: R2dbcSettings =
+  lazy val settings: R2dbcSettings =
     R2dbcSettings(typedSystem.settings.config.getConfig(testConfigPath))
 
   lazy val r2dbcExecutorProvider: R2dbcExecutorProvider =
     new R2dbcExecutorProvider(
       typedSystem,
-      r2dbcSettings.connectionFactorySettings.dialect.daoExecutionContext(r2dbcSettings, typedSystem),
-      r2dbcSettings,
+      settings.connectionFactorySettings.dialect.daoExecutionContext(settings, typedSystem),
+      settings,
       testConfigPath + ".connection-factory",
       LoggerFactory.getLogger(getClass))
 
@@ -48,22 +48,22 @@ trait TestDbLifecycle extends BeforeAndAfterAll { this: Suite =>
   lazy val persistenceExt: Persistence = Persistence(typedSystem)
 
   def pendingIfMoreThanOneDataPartition(): Unit =
-    if (r2dbcSettings.numberOfDataPartitions > 1)
+    if (settings.numberOfDataPartitions > 1)
       pending
 
   override protected def beforeAll(): Unit = {
     try {
-      r2dbcSettings.allJournalTablesWithSchema.foreach { case (table, minSlice) =>
+      settings.allJournalTablesWithSchema.foreach { case (table, minSlice) =>
         Await.result(
           r2dbcExecutor(minSlice).updateOne("beforeAll delete")(_.createStatement(s"delete from $table")),
           10.seconds)
       }
-      r2dbcSettings.allSnapshotTablesWithSchema.foreach { case (table, minSlice) =>
+      settings.allSnapshotTablesWithSchema.foreach { case (table, minSlice) =>
         Await.result(
           r2dbcExecutor(minSlice).updateOne("beforeAll delete")(_.createStatement(s"delete from $table")),
           10.seconds)
       }
-      r2dbcSettings.allDurableStateTablesWithSchema.foreach { case (table, minSlice) =>
+      settings.allDurableStateTablesWithSchema.foreach { case (table, minSlice) =>
         Await.result(
           r2dbcExecutor(minSlice).updateOne("beforeAll delete")(_.createStatement(s"delete from $table")),
           10.seconds)

core/src/test/scala/akka/persistence/r2dbc/internal/H2AdditionalInitForSchemaSpec.scala

@@ -55,8 +55,7 @@ class H2AdditionalInitForSchemaSpec
   private def exists(slice: Int, whereCondition: String): Boolean =
     r2dbcExecutor(slice)
       .selectOne("count")(
-        _.createStatement(
-          s"select count(*) from ${r2dbcSettings.durableStateTableWithSchema(slice)} where $whereCondition"),
+        _.createStatement(s"select count(*) from ${settings.durableStateTableWithSchema(slice)} where $whereCondition"),
         row => row.get(0, classOf[java.lang.Long]).longValue())
       .futureValue
       .contains(1)

core/src/test/scala/akka/persistence/r2dbc/internal/R2dbcExecutorSpec.scala

@@ -48,12 +48,12 @@ class R2dbcExecutorSpec
 
   // need pg_sleep or similar
   private def canBeTestedWithDialect: Boolean =
-    r2dbcSettings.connectionFactorySettings.dialect == PostgresDialect ||
-    r2dbcSettings.connectionFactorySettings.dialect == YugabyteDialect
+    settings.connectionFactorySettings.dialect == PostgresDialect ||
+    settings.connectionFactorySettings.dialect == YugabyteDialect
 
   private def pendingIfCannotBeTestedWithDialect(): Unit = {
     if (!canBeTestedWithDialect) {
-      info(s"Can't be tested with dialect [${r2dbcSettings.dialectName}]")
+      info(s"Can't be tested with dialect [${settings.dialectName}]")
       pending
     }
   }
@@ -89,7 +89,7 @@ class R2dbcExecutorSpec
           connection.createStatement(s"insert into $table (col) values ('b')"))
       }
 
-      Thread.sleep(r2dbcSettings.connectionFactorySettings.poolSettings.closeCallsExceeding.get.toMillis)
+      Thread.sleep(settings.connectionFactorySettings.poolSettings.closeCallsExceeding.get.toMillis)
 
       // The request will fail with PostgresConnectionClosedException
       result.failed.futureValue shouldBe a[R2dbcNonTransientResourceException]
@@ -118,7 +118,7 @@ class R2dbcExecutorSpec
 
       val result = r2dbcExecutor.updateOne("test")(_.createStatement("select pg_sleep(4)"))
 
-      Thread.sleep(r2dbcSettings.connectionFactorySettings.poolSettings.closeCallsExceeding.get.toMillis)
+      Thread.sleep(settings.connectionFactorySettings.poolSettings.closeCallsExceeding.get.toMillis)
 
       // The request will fail with PostgresConnectionClosedException
       result.failed.futureValue shouldBe a[R2dbcNonTransientResourceException]

core/src/test/scala/akka/persistence/r2dbc/journal/PersistSerializedEventSpec.scala

@@ -9,7 +9,6 @@ import akka.actor.testkit.typed.scaladsl.LogCapturing
 import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
 import akka.actor.typed.ActorSystem
 import akka.persistence.SerializedEvent
-import akka.persistence.r2dbc.R2dbcSettings
 import akka.persistence.r2dbc.TestActors.Persister
 import akka.persistence.r2dbc.TestConfig
 import akka.persistence.r2dbc.TestData
@@ -27,7 +26,6 @@ class PersistSerializedEventSpec
     with LogCapturing {
 
   override def typedSystem: ActorSystem[_] = system
-  private val settings = R2dbcSettings(system.settings.config.getConfig("akka.persistence.r2dbc"))
 
   case class Row(pid: String, seqNr: Long, serializerId: Int, manifest: String)
 

core/src/test/scala/akka/persistence/r2dbc/journal/PersistTagsSpec.scala

@@ -6,18 +6,19 @@ package akka.persistence.r2dbc.journal
 
 import scala.concurrent.duration._
 
+import org.scalatest.wordspec.AnyWordSpecLike
+
 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.r2dbc.R2dbcSettings
 import akka.persistence.r2dbc.TestActors.Persister
 import akka.persistence.r2dbc.TestConfig
 import akka.persistence.r2dbc.TestData
 import akka.persistence.r2dbc.TestDbLifecycle
+import akka.persistence.r2dbc.internal.codec.TagsCodec
 import akka.persistence.r2dbc.internal.codec.TagsCodec.TagsCodecRichRow
 import akka.persistence.typed.PersistenceId
-import org.scalatest.wordspec.AnyWordSpecLike
 
 class PersistTagsSpec
     extends ScalaTestWithActorTestKit(TestConfig.config)
@@ -27,8 +28,7 @@ class PersistTagsSpec
     with LogCapturing {
 
   override def typedSystem: ActorSystem[_] = system
-  private val settings = R2dbcSettings(system.settings.config.getConfig("akka.persistence.r2dbc"))
-  import settings.codecSettings.JournalImplicits.tagsCodec
+  implicit val codec: TagsCodec = settings.codecSettings.JournalImplicits.tagsCodec
   case class Row(pid: String, seqNr: Long, tags: Set[String])
 
   private def selectRows(table: String, minSlice: Int): IndexedSeq[Row] = {
@@ -44,7 +44,7 @@ class PersistTagsSpec
   }
 
   private def selectAllRows(): IndexedSeq[Row] =
-    r2dbcSettings.allJournalTablesWithSchema.toVector.sortBy(_._1).flatMap { case (table, minSlice) =>
+    settings.allJournalTablesWithSchema.toVector.sortBy(_._1).flatMap { case (table, minSlice) =>
       selectRows(table, minSlice)
     }
 

core/src/test/scala/akka/persistence/r2dbc/journal/PersistTimestampSpec.scala

@@ -5,23 +5,27 @@
 package akka.persistence.r2dbc.journal
 
 import java.time.Instant
+
 import scala.concurrent.duration._
+
+import org.scalatest.wordspec.AnyWordSpecLike
+
 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.r2dbc.internal.codec.PayloadCodec.RichRow
-import akka.persistence.r2dbc.R2dbcSettings
 import akka.persistence.r2dbc.TestActors.Persister
 import akka.persistence.r2dbc.TestConfig
 import akka.persistence.r2dbc.TestData
 import akka.persistence.r2dbc.TestDbLifecycle
+import akka.persistence.r2dbc.internal.codec.PayloadCodec
+import akka.persistence.r2dbc.internal.codec.PayloadCodec.RichRow
 import akka.persistence.r2dbc.internal.codec.TimestampCodec
-import akka.persistence.r2dbc.internal.codec.TimestampCodec.{ PostgresTimestampCodec, SqlServerTimestampCodec }
 import akka.persistence.r2dbc.internal.codec.TimestampCodec.TimestampCodecRichRow
+import akka.persistence.r2dbc.internal.codec.TimestampCodec.PostgresTimestampCodec
+import akka.persistence.r2dbc.internal.codec.TimestampCodec.SqlServerTimestampCodec
 import akka.persistence.typed.PersistenceId
 import akka.serialization.SerializationExtension
-import org.scalatest.wordspec.AnyWordSpecLike
 
 class PersistTimestampSpec
     extends ScalaTestWithActorTestKit(TestConfig.config)
@@ -30,9 +34,8 @@ class PersistTimestampSpec
     with TestData
     with LogCapturing {
 
-  import settings.codecSettings.JournalImplicits.journalPayloadCodec
+  implicit val payloadCodec: PayloadCodec = settings.codecSettings.JournalImplicits.journalPayloadCodec
   override def typedSystem: ActorSystem[_] = system
-  private val settings = R2dbcSettings(system.settings.config.getConfig("akka.persistence.r2dbc"))
   private val serialization = SerializationExtension(system)
   case class Row(pid: String, seqNr: Long, dbTimestamp: Instant, event: String)
 
@@ -64,7 +67,7 @@ class PersistTimestampSpec
   }
 
   private def selectAllRows(): IndexedSeq[Row] =
-    r2dbcSettings.allJournalTablesWithSchema.toVector.sortBy(_._1).flatMap { case (table, minSlice) =>
+    settings.allJournalTablesWithSchema.toVector.sortBy(_._1).flatMap { case (table, minSlice) =>
       selectRows(table, minSlice)
     }
 

core/src/test/scala/akka/persistence/r2dbc/journal/R2dbcJournalPerfManyActorsSpec.scala

@@ -34,7 +34,7 @@ class R2dbcJournalPerfManyActorsSpec extends JournalPerfSpec(R2dbcJournalPerfSpe
 
   "A PersistentActor's performance" must {
 
-    if (r2dbcSettings.dialectName == "sqlserver") {
+    if (settings.dialectName == "sqlserver") {
       pending
     }
 

core/src/test/scala/akka/persistence/r2dbc/journal/R2dbcJournalPerfSpec.scala

@@ -31,7 +31,7 @@ class R2dbcJournalPerfSpec extends JournalPerfSpec(R2dbcJournalPerfSpec.config)
   override def typedSystem: ActorSystem[_] = system.toTyped
 
   override def benchActor(replyAfter: Int): ActorRef = {
-    if (r2dbcSettings.dialectName == "sqlserver")
+    if (settings.dialectName == "sqlserver")
       throw new TestPendingException
     else
       super.benchActor(replyAfter)