pjfanning commented on code in PR #194:
URL:
https://github.com/apache/pekko-persistence-r2dbc/pull/194#discussion_r1993826263
##########
projection/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala:
##########
@@ -71,6 +71,9 @@ object TestConfig {
}
}
pekko.actor.testkit.typed.default-timeout = 10s
- """)))
+ """))
}
+
+ // FIXME ideally every dependant that combines this config with other
configs should load/resolve at their callsites
Review Comment:
I don't understand this comment. Dependent is the correct spelling. If you
don't intend to fix this, could it be made into a GitHub issue and you can even
put its URL here?
##########
core/src/test/scala/org/apache/pekko/persistence/r2dbc/journal/RuntimePluginConfigSpec.scala:
##########
@@ -0,0 +1,322 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pekko.persistence.r2dbc.journal
+
+import scala.collection.immutable.ListSet
+import scala.concurrent.Await
+import scala.concurrent.duration._
+import com.typesafe.config.Config
+import com.typesafe.config.ConfigFactory
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ExtendedActorSystem
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.actor.typed.ActorRef
+import pekko.actor.typed.ActorSystem
+import pekko.actor.typed.Behavior
+import pekko.actor.typed.scaladsl.adapter._
+import pekko.persistence.JournalProtocol.RecoverySuccess
+import pekko.persistence.JournalProtocol.ReplayMessages
+import pekko.persistence.JournalProtocol.ReplayedMessage
+import pekko.persistence.Persistence
+import pekko.persistence.SelectedSnapshot
+import pekko.persistence.SnapshotProtocol.LoadSnapshot
+import pekko.persistence.SnapshotProtocol.LoadSnapshotResult
+import pekko.persistence.SnapshotSelectionCriteria
+import pekko.persistence.query.PersistenceQuery
+import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.JournalSettings
+import pekko.persistence.r2dbc.SnapshotSettings
+import pekko.persistence.r2dbc.StateSettings
+import pekko.persistence.r2dbc.TestConfig
+import pekko.persistence.r2dbc.internal.R2dbcExecutor
+import pekko.persistence.r2dbc.query.scaladsl.R2dbcReadJournal
+import pekko.persistence.r2dbc.state.scaladsl.R2dbcDurableStateStore
+import pekko.persistence.state.scaladsl.GetObjectResult
+import pekko.persistence.typed.PersistenceId
+import pekko.persistence.typed.scaladsl.Effect
+import pekko.persistence.typed.scaladsl.EventSourcedBehavior
+import pekko.persistence.typed.scaladsl.RetentionCriteria
+import pekko.stream.scaladsl.Sink
+import org.scalatest.BeforeAndAfterEach
+import org.scalatest.Inside
+import org.scalatest.freespec.AnyFreeSpecLike
+import org.slf4j.LoggerFactory
+
+object RuntimePluginConfigSpec {
+
+ trait EventSourced {
+ import EventSourced._
+
+ def configKey: String
+ def database: String
+
+ lazy val config: Config =
+ ConfigFactory
+ .load(
+ ConfigFactory
+ .parseString(
+ s"""
+ $configKey = $${pekko.persistence.r2dbc}
+ $configKey = {
+ connection-factory {
+ database = "$database"
+ }
+
+ journal.$configKey.connection-factory =
$${$configKey.connection-factory}
+ journal.use-connection-factory =
"$configKey.connection-factory"
+ query.$configKey.connection-factory =
$${$configKey.connection-factory}
+ query.use-connection-factory = "$configKey.connection-factory"
+ snapshot.$configKey.connection-factory =
$${$configKey.connection-factory}
+ snapshot.use-connection-factory =
"$configKey.connection-factory"
+ }
+ """
+ )
+ .withFallback(TestConfig.unresolvedConfig)
+ )
+
+ def apply(persistenceId: String): Behavior[Command] =
+ EventSourcedBehavior[Command, String, String](
+ PersistenceId.ofUniqueId(persistenceId),
+ "",
+ (state, cmd) =>
+ cmd match {
+ case Save(text, replyTo) =>
+ Effect.persist(text).thenRun(_ => replyTo ! Done)
+ case ShowMeWhatYouGot(replyTo) =>
+ replyTo ! state
+ Effect.none
+ case Stop =>
+ Effect.stop()
+ },
+ (state, evt) => Seq(state, evt).filter(_.nonEmpty).mkString("|"))
+ .withRetention(RetentionCriteria.snapshotEvery(1, Int.MaxValue))
+ .withJournalPluginId(s"$configKey.journal")
+ .withJournalPluginConfig(Some(config))
+ .withSnapshotPluginId(s"$configKey.snapshot")
+ .withSnapshotPluginConfig(Some(config))
+ }
+ object EventSourced {
+ sealed trait Command
+ case class Save(text: String, replyTo: ActorRef[Done]) extends Command
+ case class ShowMeWhatYouGot(replyTo: ActorRef[String]) extends Command
+ case object Stop extends Command
+ }
+
+ trait DurableState {
+ def typedSystem: ActorSystem[_]
+ def configKey: String
+ def database: String
+
+ lazy val config: Config =
+ ConfigFactory
+ .load(
+ ConfigFactory
+ .parseString(
+ s"""
+ $configKey = $${pekko.persistence.r2dbc}
+ $configKey = {
+ connection-factory {
+ database = "$database"
+ }
+
+ state.$configKey.connection-factory =
$${$configKey.connection-factory}
+ state.use-connection-factory = "$configKey.connection-factory"
+ }
+ """
+ )
+ .withFallback(TestConfig.unresolvedConfig)
+ )
+
+ // TODO refactor to use DurableState instead of plugin implementation
directly once DurableState implements runtime config
Review Comment:
I don't like tracking issues in code, can this not be an issue logged in
GitHub and you can provide the URL here?
##########
projection/src/test/scala/org/apache/pekko/projection/r2dbc/EventSourcedPubSubSpec.scala:
##########
@@ -43,23 +43,23 @@ import org.slf4j.LoggerFactory
object EventSourcedPubSubSpec {
- val config: Config = ConfigFactory
- .parseString("""
- pekko.persistence.r2dbc {
- journal.publish-events = on
- query {
+ val config: Config = ConfigFactory.load(
Review Comment:
why do you need ConfigFactory.load to wrap ConfigFactory.parseString?
##########
core/src/test/scala/org/apache/pekko/persistence/r2dbc/journal/RuntimePluginConfigSpec.scala:
##########
@@ -0,0 +1,322 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pekko.persistence.r2dbc.journal
+
+import scala.collection.immutable.ListSet
+import scala.concurrent.Await
+import scala.concurrent.duration._
+import com.typesafe.config.Config
+import com.typesafe.config.ConfigFactory
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ExtendedActorSystem
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.actor.typed.ActorRef
+import pekko.actor.typed.ActorSystem
+import pekko.actor.typed.Behavior
+import pekko.actor.typed.scaladsl.adapter._
+import pekko.persistence.JournalProtocol.RecoverySuccess
+import pekko.persistence.JournalProtocol.ReplayMessages
+import pekko.persistence.JournalProtocol.ReplayedMessage
+import pekko.persistence.Persistence
+import pekko.persistence.SelectedSnapshot
+import pekko.persistence.SnapshotProtocol.LoadSnapshot
+import pekko.persistence.SnapshotProtocol.LoadSnapshotResult
+import pekko.persistence.SnapshotSelectionCriteria
+import pekko.persistence.query.PersistenceQuery
+import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.JournalSettings
+import pekko.persistence.r2dbc.SnapshotSettings
+import pekko.persistence.r2dbc.StateSettings
+import pekko.persistence.r2dbc.TestConfig
+import pekko.persistence.r2dbc.internal.R2dbcExecutor
+import pekko.persistence.r2dbc.query.scaladsl.R2dbcReadJournal
+import pekko.persistence.r2dbc.state.scaladsl.R2dbcDurableStateStore
+import pekko.persistence.state.scaladsl.GetObjectResult
+import pekko.persistence.typed.PersistenceId
+import pekko.persistence.typed.scaladsl.Effect
+import pekko.persistence.typed.scaladsl.EventSourcedBehavior
+import pekko.persistence.typed.scaladsl.RetentionCriteria
+import pekko.stream.scaladsl.Sink
+import org.scalatest.BeforeAndAfterEach
+import org.scalatest.Inside
+import org.scalatest.freespec.AnyFreeSpecLike
+import org.slf4j.LoggerFactory
+
+object RuntimePluginConfigSpec {
+
+ trait EventSourced {
+ import EventSourced._
+
+ def configKey: String
+ def database: String
+
+ lazy val config: Config =
+ ConfigFactory
+ .load(
+ ConfigFactory
+ .parseString(
+ s"""
+ $configKey = $${pekko.persistence.r2dbc}
+ $configKey = {
+ connection-factory {
+ database = "$database"
+ }
+
+ journal.$configKey.connection-factory =
$${$configKey.connection-factory}
+ journal.use-connection-factory =
"$configKey.connection-factory"
+ query.$configKey.connection-factory =
$${$configKey.connection-factory}
+ query.use-connection-factory = "$configKey.connection-factory"
+ snapshot.$configKey.connection-factory =
$${$configKey.connection-factory}
+ snapshot.use-connection-factory =
"$configKey.connection-factory"
+ }
+ """
+ )
+ .withFallback(TestConfig.unresolvedConfig)
+ )
+
+ def apply(persistenceId: String): Behavior[Command] =
+ EventSourcedBehavior[Command, String, String](
+ PersistenceId.ofUniqueId(persistenceId),
+ "",
+ (state, cmd) =>
+ cmd match {
+ case Save(text, replyTo) =>
+ Effect.persist(text).thenRun(_ => replyTo ! Done)
+ case ShowMeWhatYouGot(replyTo) =>
+ replyTo ! state
+ Effect.none
+ case Stop =>
+ Effect.stop()
+ },
+ (state, evt) => Seq(state, evt).filter(_.nonEmpty).mkString("|"))
+ .withRetention(RetentionCriteria.snapshotEvery(1, Int.MaxValue))
+ .withJournalPluginId(s"$configKey.journal")
+ .withJournalPluginConfig(Some(config))
+ .withSnapshotPluginId(s"$configKey.snapshot")
+ .withSnapshotPluginConfig(Some(config))
+ }
+ object EventSourced {
+ sealed trait Command
+ case class Save(text: String, replyTo: ActorRef[Done]) extends Command
+ case class ShowMeWhatYouGot(replyTo: ActorRef[String]) extends Command
+ case object Stop extends Command
+ }
+
+ trait DurableState {
+ def typedSystem: ActorSystem[_]
+ def configKey: String
+ def database: String
+
+ lazy val config: Config =
+ ConfigFactory
Review Comment:
again I'm confused by need for load and parseString
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@pekko.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscr...@pekko.apache.org
For additional commands, e-mail: notifications-h...@pekko.apache.org
