ptrdom commented on code in PR #664:
URL: https://github.com/apache/pekko-projection/pull/664#discussion_r4377550266
##########
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:
The rest of this page uses `@@snip` with Scala and Java tabs. Consider
adding short snippets for both, paired with the Java usage suggested on the
javadsl file, plus `@apidoc[R2dbcSession$]`. Its own subsection may read better
than the intro paragraph.
##########
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:
`connectionFactoryFor(path)` and `R2dbcProjectionSettings(system)` run
before any `Future` exists. A typo in the config path throws `ConfigException`
straight out of `withSession`, so a Java caller chaining `.exceptionally(...)`
never sees it. Consider wrapping it in `try { ... } catch { case NonFatal(e) =>
Future.failed(e) }`. Upstream does the same thing, so this is optional.
##########
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:
Nit: `R2dbcProjectionSettings(system)` re-parses the whole
`pekko.projection.r2dbc` block on every call, and this helper then reads
`use-connection-factory` a second time, but the settings object already has it
as `useConnectionFactory`. You could use that and drop the helper. It's
`@InternalApi`, so removing it is binary-safe.
##########
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:
Nit: the javadsl class keeps `@ApiMayChange`, but the new companion object
and the scaladsl side don't have it. Upstream dropped the annotation in #1017.
Either drop it here too, which needs maintainers to agree to stabilise the
javadsl, or annotate the new `object R2dbcSession` as well so both sides
promise the same stability.
##########
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:
The javadsl tests are in Scala, so they call `withSession` through the Scala
object. They never exercise the static forwarders that Java users rely on, or
how Java infers lambda types against the overloads. If a later change
suppresses the forwarders, every test still passes, but
`R2dbcSession.withSession(system, session -> ...)` stops compiling for Java
users.
Could you add a small Java usage? For example, a snippet in
`R2dbcProjectionDocExample.java` or a Java compile test along the lines of
grpc's `ConsumerCompileTest`. That also keeps the Scala and Java DSLs in step,
as AGENTS.md asks.
--
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]
