feat(tx): FP-friendly resource-block API [closes #99]

Author: ivan-klass

scalasql/core/src/DbApi.scala

@@ -194,12 +194,12 @@ object DbApi {
   trait Txn extends DbApi {
 
     /**
-     * Creates a SQL Savepoint that is active within the given block; automatically
+     * Returns a ResourceBlock that creates a SQL Savepoint that is active within the given block; automatically
      * releases the savepoint if the block completes successfully and rolls it back
      * if the block terminates with an exception, and allows you to roll back the
      * savepoint manually via the [[DbApi.Savepoint]] parameter passed to that block
      */
-    def savepoint[T](block: DbApi.Savepoint => T): T
+    def savepoint: UseBlock[DbApi.Savepoint]
 
     /**
      * Rolls back any active Savepoints and then rolls back this Transaction
@@ -575,31 +575,28 @@ object DbApi {
       DbApi.renderSql(query, config, dialect.withCastParams(castParams))
     }
 
-    val savepointStack = collection.mutable.ArrayDeque.empty[java.sql.Savepoint]
-
-    def savepoint[T](block: DbApi.Savepoint => T): T = {
-      val savepoint = connection.setSavepoint()
-      savepointStack.append(savepoint)
-
-      try {
-        val res = block(new DbApi.SavepointImpl(savepoint, () => rollbackSavepoint(savepoint)))
-        if (dialect.supportSavepointRelease && savepointStack.lastOption.exists(_ eq savepoint)) {
-          // Only release if this savepoint has not been rolled back,
-          // directly or indirectly
-          connection.releaseSavepoint(savepoint)
-        }
-        res
-      } catch {
-        case e: Throwable =>
-          rollbackSavepoint(savepoint)
-          throw e
+    private val savepointStack = collection.mutable.ArrayDeque.empty[java.sql.Savepoint]
+
+    lazy val savepoint: UseBlock[DbApi.Savepoint] = UseBlockImpl {
+      val jSavepoint = connection.setSavepoint()
+      savepointStack.append(jSavepoint)
+      jSavepoint
+    }((jSp, error) => {
+      error match {
+        case None =>
+          if (dialect.supportSavepointRelease && savepointStack.lastOption.exists(_ eq jSp)) {
+            // Only release if this savepoint has not been rolled back,
+            // directly or indirectly
+            connection.releaseSavepoint(jSp)
+          }
+        case Some(_) => rollbackSavepoint(jSp)
       }
-    }
+    }).map(jSp => new DbApi.SavepointImpl(jSp, () => rollbackSavepoint(jSp)))
 
     // Make sure we keep track of what savepoints are active on the stack, so we do
     // not release or rollback the same savepoint multiple times even in the case of
     // exceptions or explicit rollbacks
-    def rollbackSavepoint(savepoint: java.sql.Savepoint) = {
+    private def rollbackSavepoint(savepoint: java.sql.Savepoint) = {
       savepointStack.indexOf(savepoint) match {
         case -1 => // do nothing
         case savepointIndex =>
@@ -608,7 +605,7 @@ object DbApi {
       }
     }
 
-    def rollback() = {
+    def rollback(): Unit = {
       try {
         notifyListeners(listeners)(_.beforeRollback())
       } finally {
@@ -618,6 +615,11 @@ object DbApi {
       }
     }
 
+    /** Attempts rollback, adding any exceptions as suppressed to the cause */
+    def rollbackCause(cause: Throwable): Unit =
+      try rollback()
+      catch { case e: Throwable => cause.addSuppressed(e) }
+
     private def cast[T](t: Any): T = t.asInstanceOf[T]
 
     private def flattenParamPuts[T](flattened: SqlStr.Flattened) = {

scalasql/core/src/DbClient.scala

@@ -1,7 +1,5 @@
 package scalasql.core
 
-import scalasql.core.DialectConfig
-
 /**
  * A database client. Primarily allows you to access the database within a [[transaction]]
  * block or via [[getAutoCommitClientConnection]]
@@ -14,12 +12,12 @@ trait DbClient {
   def renderSql[Q, R](query: Q, castParams: Boolean = false)(implicit qr: Queryable[Q, R]): String
 
   /**
-   * Opens a database transaction within the given [[block]], automatically committing it
+   * Returns a [[UseBlock]] for database transaction, automatically committing it
    * if the block returns successfully and rolling it back if the blow fails with an uncaught
    * exception. Within the block, you provides a [[DbApi.Txn]] you can use to run queries, create
    * savepoints, or roll back the transaction.
    */
-  def transaction[T](block: DbApi.Txn => T): T
+  def transaction: UseBlock[DbApi.Txn]
 
   /**
    * Provides a [[DbApi]] that you can use to run queries in "auto-commit" mode, such
@@ -71,39 +69,29 @@ object DbClient {
       DbApi.renderSql(query, config, dialect.withCastParams(castParams))
     }
 
-    def transaction[T](block: DbApi.Txn => T): T = {
+    lazy val transaction: UseBlock[DbApi.Txn] = UseBlockImpl[DbApi.Impl] {
       connection.setAutoCommit(false)
       val txn = new DbApi.Impl(connection, config, dialect, listeners, autoCommit = false)
-      var rolledBack = false
-      try {
-        notifyListeners(txn.listeners)(_.begin())
-        val result = block(txn)
-        notifyListeners(txn.listeners)(_.beforeCommit())
-        result
-      } catch {
-        case e: Throwable =>
-          rolledBack = true
+      notifyListeners(txn.listeners)(_.begin())
+      txn
+    }((txn, error) => {
+      error match {
+        case None =>
           try {
-            notifyListeners(txn.listeners)(_.beforeRollback())
+            notifyListeners(txn.listeners)(_.beforeCommit())
           } catch {
-            case e2: Throwable => e.addSuppressed(e2)
-          } finally {
-            connection.rollback()
-            try {
-              notifyListeners(txn.listeners)(_.afterRollback())
-            } catch {
-              case e3: Throwable => e.addSuppressed(e3)
-            }
+            case beforeCommitHookErr: Throwable =>
+              txn.rollbackCause(beforeCommitHookErr)
+              throw beforeCommitHookErr
           }
-          throw e
-      } finally {
-        // this commits uncommitted operations, if any
-        connection.setAutoCommit(true)
-        if (!rolledBack) {
+          // this commits uncommitted operations, if any
+          connection.setAutoCommit(true)
+          // afterCommit exceptions just propagate - commit already done
           notifyListeners(txn.listeners)(_.afterCommit())
-        }
+
+        case Some(useError) => txn.rollbackCause(useError)
       }
-    }
+    })
 
     def getAutoCommitClientConnection: DbApi = {
       connection.setAutoCommit(true)
@@ -130,13 +118,13 @@ object DbClient {
       DbApi.renderSql(query, config, dialect.withCastParams(castParams))
     }
 
-    private def withConnection[T](f: DbClient.Connection => T): T = {
-      val connection = dataSource.getConnection
-      try f(new DbClient.Connection(connection, config, listeners))
-      finally connection.close()
-    }
+    private lazy val withConnectionImpl: UseBlockImpl[Connection] = UseBlockImpl
+      .autoCloseable(dataSource.getConnection)
+      .map(new Connection(_, config, listeners))
+
+    lazy val withConnection: UseBlock[Connection] = withConnectionImpl
 
-    def transaction[T](block: DbApi.Txn => T): T = withConnection(_.transaction(block))
+    lazy val transaction: UseBlock[DbApi.Txn] = withConnectionImpl.flatMap(_.transaction)
 
     def getAutoCommitClientConnection: DbApi = {
       val connection = dataSource.getConnection

scalasql/core/src/UseBlock.scala

@@ -0,0 +1,87 @@
+package scalasql.core
+
+/**
+ * A block abstraction that wraps resource usage with proper acquiring and close/release.
+ * Exposes lifecycle operations separately, which is useful for integration with FP effect libraries.
+ * Should not be implemented outside of library code, so sealed.
+ *
+ * @tparam A The resource type provided during the `use` phase
+ */
+sealed trait UseBlock[+A] {
+
+  /**
+   * Acquires the resource. Called once at the start of the block.
+   * @return The acquired resource and release function.
+   */
+  def allocate(): (A, Option[Throwable] => Unit)
+
+  /**
+   * Combines acquire, use, and release into a single operation.
+   * Makes it friendly to traditional block-style API.
+   */
+  final def apply[T](use: A => T): T = {
+    val (resource, release) = allocate()
+    var usedOk = false
+    try {
+      val result = use(resource)
+      usedOk = true
+      release(None)
+      result
+    } catch {
+      case e: Throwable =>
+        if (!usedOk) {
+          release(Some(e))
+        } // else - we had an error in `release(None)`, so just propagating
+        throw e
+    }
+  }
+}
+
+/** Package-private implementation for internal composition */
+private[core] class UseBlockImpl[+A](alloc: () => (A, Option[Throwable] => Unit))
+    extends UseBlock[A] {
+
+  def allocate(): (A, Option[Throwable] => Unit) = alloc()
+
+  def map[B](f: A => B): UseBlockImpl[B] = new UseBlockImpl[B](() => {
+    val (a, release) = alloc()
+    (f(a), release)
+  })
+
+  def flatMap[B](f: A => UseBlock[B]): UseBlockImpl[B] = new UseBlockImpl[B](() => {
+    val (outerResource, outerRelease) = alloc()
+    val (innerResource, innerRelease) =
+      try {
+        f(outerResource).allocate()
+      } catch {
+        case e: Throwable =>
+          outerRelease(Some(e))
+          throw e
+      }
+    def combinedRelease(errOpt: Option[Throwable]): Unit = {
+      var errorForOuter = errOpt
+      try {
+        innerRelease(errOpt)
+      } catch {
+        case e: Throwable =>
+          errorForOuter = Some(e)
+          throw e
+      } finally {
+        outerRelease(errorForOuter)
+      }
+    }
+    (innerResource, combinedRelease)
+  })
+}
+
+private[core] object UseBlockImpl {
+
+  def apply[A](acquire: => A)(release: (A, Option[Throwable]) => Unit): UseBlockImpl[A] =
+    new UseBlockImpl[A](() => {
+      val a = acquire
+      (a, errOpt => release(a, errOpt))
+    })
+
+  def autoCloseable[A <: AutoCloseable](acquire: => A): UseBlockImpl[A] =
+    apply(acquire)((a, _) => a.close())
+}