pjfanning commented on code in PR #664:
URL: https://github.com/apache/pekko-projection/pull/664#discussion_r4382784690
##########
r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/javadsl/R2dbcSession.scala:
##########
@@ -25,10 +26,45 @@ import org.apache.pekko
import pekko.actor.typed.ActorSystem
import pekko.annotation.ApiMayChange
import pekko.persistence.r2dbc.internal.R2dbcExecutor
+import pekko.projection.r2dbc.scaladsl
import io.r2dbc.spi.Connection
import io.r2dbc.spi.Row
import io.r2dbc.spi.Statement
+object R2dbcSession {
+
+ /**
+ * Runs the passed function using a R2dbcSession with a new transaction. The
connection is closed and the
+ * transaction is committed at the end or rolled back in case of failures.
+ *
+ * The connection factory is the one configured by
`pekko.projection.r2dbc.use-connection-factory`.
+ *
+ * @since 2.0.0
+ */
+ def withSession[A](system: ActorSystem[?], fun: JFunction[R2dbcSession,
CompletionStage[A]]): CompletionStage[A] =
+ withSession(system,
scaladsl.R2dbcSession.connectionFactoryConfigPath(system), fun)
+
+ /**
+ * Runs the passed function using a R2dbcSession with a new transaction. The
connection is closed and the
+ * transaction is committed at the end or rolled back in case of failures.
+ *
+ * @param connectionFactoryConfigPath config path of the connection factory
to use, for example
+ * `pekko.persistence.r2dbc.connection-factory`
+ * @since 2.0.0
+ */
+ def withSession[A](
+ system: ActorSystem[?],
+ connectionFactoryConfigPath: String,
+ fun: JFunction[R2dbcSession, CompletionStage[A]]): CompletionStage[A] =
+ scaladsl.R2dbcSession
+ .withSession(system, connectionFactoryConfigPath) { scaladslSession =>
+ val javadslSession = new
R2dbcSession(scaladslSession.connection)(system.executionContext, system)
+ fun(javadslSession).asScala
+ }
+ .asJava
Review Comment:
Added a Java usage in e56bbc84: a `#withSession` snippet in
`R2dbcProjectionDocExample.java` that calls `R2dbcSession.withSession(system,
session -> ...)` through the static forwarders, so `r2dbc/Test/compile` breaks
if they go away. A matching Scala snippet is in
`R2dbcProjectionDocExample.scala`.
##########
docs/src/main/paradox/r2dbc.md:
##########
@@ -14,6 +14,10 @@ The target database operations can be run in the same
transaction as the storage
that @ref:[exactly-once](#exactly-once) processing semantics is supported. It
also offers
@ref:[at-least-once](#at-least-once) semantics.
+An `R2dbcSession` can also be used outside of a projection handler, for
example to query the projected data.
+`R2dbcSession.withSession` runs the given function with a new transaction that
is committed when the returned
+@scala[`Future`]@java[`CompletionStage`] completes successfully, and rolled
back if it fails.
Review Comment:
Done in e56bbc84. The intro paragraph is now its own section, "Using
R2dbcSession outside a handler", under Handler. It has `@apidoc[R2dbcSession$]`
and Scala/Java `@@snip` tabs.
##########
r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcSession.scala:
##########
@@ -19,10 +19,55 @@ import scala.concurrent.Future
import org.apache.pekko
import pekko.actor.typed.ActorSystem
+import pekko.annotation.InternalApi
+import pekko.persistence.r2dbc.ConnectionFactoryProvider
import pekko.persistence.r2dbc.internal.R2dbcExecutor
+import pekko.projection.r2dbc.R2dbcProjectionSettings
import io.r2dbc.spi.Connection
import io.r2dbc.spi.Row
import io.r2dbc.spi.Statement
+import org.slf4j.LoggerFactory
+
+object R2dbcSession {
+ private val log = LoggerFactory.getLogger(classOf[R2dbcSession])
+
+ /**
+ * Runs the passed function using a R2dbcSession with a new transaction. The
connection is closed and the
+ * transaction is committed at the end or rolled back in case of failures.
+ *
+ * The connection factory is the one configured by
`pekko.projection.r2dbc.use-connection-factory`.
+ *
+ * @since 2.0.0
+ */
+ def withSession[A](system: ActorSystem[?])(fun: R2dbcSession => Future[A]):
Future[A] =
+ withSession(system, connectionFactoryConfigPath(system))(fun)
+
+ /**
+ * Runs the passed function using a R2dbcSession with a new transaction. The
connection is closed and the
+ * transaction is committed at the end or rolled back in case of failures.
+ *
+ * @param connectionFactoryConfigPath config path of the connection factory
to use, for example
+ * `pekko.persistence.r2dbc.connection-factory`
+ * @since 2.0.0
+ */
+ def withSession[A](system: ActorSystem[?], connectionFactoryConfigPath:
String)(
+ fun: R2dbcSession => Future[A]): Future[A] = {
+ val connectionFactory =
ConnectionFactoryProvider(system).connectionFactoryFor(connectionFactoryConfigPath)
+ val r2dbcExecutor =
+ new R2dbcExecutor(connectionFactory, log,
R2dbcProjectionSettings(system).logDbCallsExceeding)(
Review Comment:
Done in e56bbc84. The settings and connection factory lookup are wrapped in
`try`/`NonFatal`, so the error comes back as a failed
`Future`/`CompletionStage`. New `R2dbcSessionSpec` cases for an unknown config
path in both DSLs fail if the error is rethrown.
##########
r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcSession.scala:
##########
@@ -19,10 +19,55 @@ import scala.concurrent.Future
import org.apache.pekko
import pekko.actor.typed.ActorSystem
+import pekko.annotation.InternalApi
+import pekko.persistence.r2dbc.ConnectionFactoryProvider
import pekko.persistence.r2dbc.internal.R2dbcExecutor
+import pekko.projection.r2dbc.R2dbcProjectionSettings
import io.r2dbc.spi.Connection
import io.r2dbc.spi.Row
import io.r2dbc.spi.Statement
+import org.slf4j.LoggerFactory
+
+object R2dbcSession {
+ private val log = LoggerFactory.getLogger(classOf[R2dbcSession])
+
+ /**
+ * Runs the passed function using a R2dbcSession with a new transaction. The
connection is closed and the
+ * transaction is committed at the end or rolled back in case of failures.
+ *
+ * The connection factory is the one configured by
`pekko.projection.r2dbc.use-connection-factory`.
+ *
+ * @since 2.0.0
+ */
+ def withSession[A](system: ActorSystem[?])(fun: R2dbcSession => Future[A]):
Future[A] =
+ withSession(system, connectionFactoryConfigPath(system))(fun)
+
+ /**
+ * Runs the passed function using a R2dbcSession with a new transaction. The
connection is closed and the
+ * transaction is committed at the end or rolled back in case of failures.
+ *
+ * @param connectionFactoryConfigPath config path of the connection factory
to use, for example
+ * `pekko.persistence.r2dbc.connection-factory`
+ * @since 2.0.0
+ */
+ def withSession[A](system: ActorSystem[?], connectionFactoryConfigPath:
String)(
+ fun: R2dbcSession => Future[A]): Future[A] = {
+ val connectionFactory =
ConnectionFactoryProvider(system).connectionFactoryFor(connectionFactoryConfigPath)
+ val r2dbcExecutor =
+ new R2dbcExecutor(connectionFactory, log,
R2dbcProjectionSettings(system).logDbCallsExceeding)(
+ system.executionContext,
+ system)
+ r2dbcExecutor.withConnection("R2dbcSession") { connection =>
+ fun(new R2dbcSession(connection)(system.executionContext, system))
+ }
+ }
+
+ /**
+ * INTERNAL API
+ */
+ @InternalApi private[pekko] def connectionFactoryConfigPath(system:
ActorSystem[?]): String =
+
system.settings.config.getConfig(R2dbcProjectionSettings.DefaultConfigPath).getString("use-connection-factory")
Review Comment:
Done in e56bbc84. Settings are parsed once per call,
`settings.useConnectionFactory` is used, and the `connectionFactoryConfigPath`
helper is removed. It was only added in this PR, so nothing released changes.
##########
r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/javadsl/R2dbcSession.scala:
##########
@@ -25,10 +26,45 @@ import org.apache.pekko
import pekko.actor.typed.ActorSystem
import pekko.annotation.ApiMayChange
import pekko.persistence.r2dbc.internal.R2dbcExecutor
+import pekko.projection.r2dbc.scaladsl
import io.r2dbc.spi.Connection
import io.r2dbc.spi.Row
import io.r2dbc.spi.Statement
+object R2dbcSession {
+
+ /**
+ * Runs the passed function using a R2dbcSession with a new transaction. The
connection is closed and the
+ * transaction is committed at the end or rolled back in case of failures.
+ *
+ * The connection factory is the one configured by
`pekko.projection.r2dbc.use-connection-factory`.
+ *
+ * @since 2.0.0
+ */
+ def withSession[A](system: ActorSystem[?], fun: JFunction[R2dbcSession,
CompletionStage[A]]): CompletionStage[A] =
+ withSession(system,
scaladsl.R2dbcSession.connectionFactoryConfigPath(system), fun)
+
+ /**
+ * Runs the passed function using a R2dbcSession with a new transaction. The
connection is closed and the
+ * transaction is committed at the end or rolled back in case of failures.
+ *
+ * @param connectionFactoryConfigPath config path of the connection factory
to use, for example
+ * `pekko.persistence.r2dbc.connection-factory`
+ * @since 2.0.0
+ */
+ def withSession[A](
+ system: ActorSystem[?],
+ connectionFactoryConfigPath: String,
+ fun: JFunction[R2dbcSession, CompletionStage[A]]): CompletionStage[A] =
+ scaladsl.R2dbcSession
+ .withSession(system, connectionFactoryConfigPath) { scaladslSession =>
+ val javadslSession = new
R2dbcSession(scaladslSession.connection)(system.executionContext, system)
+ fun(javadslSession).asScala
+ }
+ .asJava
+
+}
+
@ApiMayChange
Review Comment:
Annotated both the scaladsl and javadsl `object R2dbcSession` with
`@ApiMayChange` in e56bbc84. That matches `R2dbcProjection` and `R2dbcHandler`,
which have it in both DSLs. Stabilising the javadsl can be a separate decision.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
