feat: custom connection factory when using withSession

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/javadsl/R2dbcSession.scala

@@ -17,6 +17,7 @@ import akka.actor.typed.ActorSystem
 import akka.persistence.r2dbc.internal.R2dbcExecutor
 import akka.projection.r2dbc.scaladsl
 import io.r2dbc.spi.Connection
+import io.r2dbc.spi.ConnectionFactory
 import io.r2dbc.spi.Row
 import io.r2dbc.spi.Statement
 
@@ -40,6 +41,20 @@ object R2dbcSession {
     }
   }.asJava
 
+  /**
+   * Provide a custom connectionFactory and an optional config path to load closeCallsExceeding from.
+   */
+  def withSession[A](
+      system: ActorSystem[_],
+      connectionFactory: ConnectionFactory,
+      configPath: Optional[String],
+      fun: JFunction[R2dbcSession, CompletionStage[A]]): CompletionStage[A] = {
+    scaladsl.R2dbcSession.withSession(system, connectionFactory, configPath.toScala) { scaladslSession =>
+      val javadslSession = new R2dbcSession(scaladslSession.connection)(system.executionContext, system)
+      fun(javadslSession).asScala
+    }
+  }.asJava
+
 }
 
 final class R2dbcSession(val connection: Connection)(implicit ec: ExecutionContext, system: ActorSystem[_]) {

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/scaladsl/R2dbcSession.scala

@@ -16,6 +16,7 @@ import akka.persistence.r2dbc.ConnectionPoolSettings
 import akka.persistence.r2dbc.internal.R2dbcExecutor
 import akka.projection.r2dbc.R2dbcProjectionSettings
 import io.r2dbc.spi.Connection
+import io.r2dbc.spi.ConnectionFactory
 import io.r2dbc.spi.Row
 import io.r2dbc.spi.Statement
 import org.slf4j.LoggerFactory
@@ -37,7 +38,16 @@ object R2dbcSession {
       fun: R2dbcSession => Future[A]): Future[A] = {
     val connectionFactoryProvider = ConnectionFactoryProvider(system)
     val connectionFactory = connectionFactoryProvider.connectionFactoryFor(connectionFactoryConfigPath)
-    val poolSettings = new ConnectionPoolSettings(system.settings.config.getConfig(connectionFactoryConfigPath))
+    withSession(system, connectionFactory, Some(connectionFactoryConfigPath))(fun)
+  }
+
+  /**
+   * Provide a custom connectionFactory and an optional config path to load closeCallsExceeding from.
+   */
+  def withSession[A](system: ActorSystem[_], connectionFactory: ConnectionFactory, configPath: Option[String])(
+      fun: R2dbcSession => Future[A]): Future[A] = {
+    val poolConfig = system.settings.config.getConfig(configPath.getOrElse(connectionFactoryConfigPath(system)))
+    val poolSettings = new ConnectionPoolSettings(poolConfig)
     val r2dbcExecutor =
       new R2dbcExecutor(connectionFactory, log, logDbCallsDisabled, poolSettings.closeCallsExceeding)(
         system.executionContext,