Copilot commented on code in PR #377:
URL:
https://github.com/apache/pekko-persistence-r2dbc/pull/377#discussion_r3263399507
##########
docs/src/main/paradox/projection.md:
##########
@@ -8,7 +8,7 @@ The source of the envelopes is from a `SourceProvider`, which
can be:
* state changes for Durable State entities via the @extref:[SourceProvider for
changesBySlices](pekko-projection:durable-state.html#sourceprovider-for-changesbyslices)
with the @ref:[changesBySlices query](query.md#changesbyslices)
* any other `SourceProvider` with supported @ref:[offset types](#offset-types)
-A @apidoc[R2dbcHandler] receives a @apidoc[R2dbcSession] instance and an
envelope. The
+A @apidoc[R2dbcHandler] receives a
@apidoc[org.apache.pekko.projection.r2dbc.*.R2dbcSession] instance and an
envelope. The
Review Comment:
`@apidoc` expects a fully-qualified type name (package + class). Using
`org.apache.pekko.projection.r2dbc.*.R2dbcSession` (with `*`) is not a valid
class reference and is likely to break Paradox/API doc generation. Consider
referencing the concrete session type explicitly (e.g. the scaladsl or javadsl
`R2dbcSession`) to avoid the new name clash with
`pekko.persistence.r2dbc.session.*.R2dbcSession`.
##########
core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/DurableStateDao.scala:
##########
@@ -313,58 +469,82 @@ private[r2dbc] class DurableStateDao(settings:
StateSettings, connectionFactory:
}
}
- private def hardDeleteState(persistenceId: String): Future[Long] = {
- val result =
- r2dbcExecutor.updateOne(s"hard delete [$persistenceId]") { connection =>
- connection
- .createStatement(hardDeleteStateSql)
- .bind(0, persistenceId)
- }
+ private def processChange(
+ handler: ChangeHandler[Any],
+ connection: Connection,
+ change: DurableStateChange[Any]): Future[Done] = {
+ val session = new R2dbcSession(connection)
- if (log.isDebugEnabled())
- result.foreach(_ => log.debug("Hard deleted durable state for
persistenceId [{}]", persistenceId))
+ def excMessage(cause: Throwable): String = {
+ val (changeType, revision) = change match {
+ case upd: UpdatedDurableState[_] => "update" -> upd.revision
+ case del: DeletedDurableState[_] => "delete" -> del.revision
+ }
+ s"Change handler $changeType failed for [${change.persistenceId}]
revision [$revision], due to ${cause.getMessage}"
+ }
- result
+ try handler.process(session, change).recoverWith { case cause =>
Review Comment:
`recoverWith { case cause => ... }` matches all Throwables, including fatal
JVM errors. This can unintentionally wrap/handle fatal errors (e.g.
`VirtualMachineError`, `ThreadDeath`) as `ChangeHandlerException`. Restrict the
recovery to `NonFatal` (as you already do in the surrounding `catch`) so fatal
errors are not intercepted.
##########
core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/AdditionalColumn.scala:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.r2dbc.state.scaladsl
+
+import scala.reflect.ClassTag
+
+import org.apache.pekko
+import pekko.annotation.ApiMayChange
+import pekko.annotation.InternalApi
+
+@ApiMayChange
+object AdditionalColumn {
+ final case class Upsert[A](persistenceId: String, entityType: String, slice:
Int, revision: Long, value: A)
+
+ sealed trait Binding[+B]
+
+ final case class BindValue[B](value: B) extends Binding[B]
+
+ case object BindNull extends Binding[Nothing]
+
+ case object Skip extends Binding[Nothing]
+
+ private val scalaPrimitivesMapping: Map[Class[_], Class[_]] =
+ Map(
+ classOf[Int] -> classOf[java.lang.Integer],
+ classOf[Long] -> classOf[java.lang.Long],
+ classOf[Float] -> classOf[java.lang.Float],
+ classOf[Double] -> classOf[java.lang.Double],
+ classOf[Byte] -> classOf[java.lang.Byte],
+ classOf[Short] -> classOf[java.lang.Short],
+ classOf[Char] -> classOf[java.lang.Character])
Review Comment:
`scalaPrimitivesMapping` is missing a mapping for `Boolean` (primitive
`boolean`). If an `AdditionalColumn` has type parameter `B = Boolean`,
`fieldClass` will be the primitive class, and `Statement.bindNull(index,
fieldClass)` may fail because R2DBC drivers typically expect boxed types. Add a
`classOf[Boolean] -> classOf[java.lang.Boolean]` mapping for consistency with
the other primitives.
##########
core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/AdditionalColumnFactory.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.r2dbc.internal
+
+import scala.util.Try
+
+import org.apache.pekko
+import pekko.actor.{ ActorSystem => ClassicActorSystem }
+import pekko.actor.ExtendedActorSystem
+import pekko.actor.typed.ActorSystem
+import pekko.annotation.InternalApi
+import pekko.persistence.r2dbc.state.scaladsl.AdditionalColumn
+import pekko.persistence.r2dbc.state.{ javadsl => javadslState }
+
+/**
+ * INTERNAL API
+ */
+@InternalApi private[pekko] object AdditionalColumnFactory {
+
+ /**
+ * Adapter from javadsl.AdditionalColumn to scaladsl.AdditionalColumn
+ */
+ final class AdditionColumnAdapter(delegate:
javadslState.AdditionalColumn[Any, Any])
+ extends AdditionalColumn[Any, Any] {
+
Review Comment:
The adapter class name `AdditionColumnAdapter` looks like a typo (missing
"al" from "Additional"). Renaming it to `AdditionalColumnAdapter` would make
intent clearer and avoid confusion when debugging stack traces.
--
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]
