Add withTransaction (COMMIT on success, ROLLBACK on failure) to conne…

…ction pool (#68)

Sources/OracleNIO/Pool/OracleClient.swift

@@ -189,6 +189,30 @@ public final class OracleClient: Sendable, Service {
         }
     }
 
+    /// Lease a connection in the context of an isolated transaction for the provided `closure`'s lifetime.
+    ///
+    /// - Note: If the closure does not fail, all changes will be committed via `COMMIT`.
+    ///         If the closure throws, a `ROLLBACK` will be issued, the original error rethrows.
+    ///
+    /// - Parameter closure: A closure that uses the passed `OracleConnection`. The closure **must not** capture
+    ///                      the provided `OracleConnection`.
+    /// - Returns: The closure's return value.
+    @discardableResult
+    public func withTransaction<Result>(
+        _ closure: (OracleConnection) async throws -> Result
+    ) async throws -> Result {
+        try await self.withConnection { connection in
+            do {
+                let result = try await closure(connection)
+                try await connection.commit()
+                return result
+            } catch {
+                try await connection.rollback()
+                throw error
+            }
+        }
+    }
+
 
     /// The client's run method. Users must call this function in order to start the client's background task processing
     /// like creating and destroying connections and running timers.

Tests/IntegrationTests/OracleClientTests.swift

@@ -94,6 +94,62 @@ final class OracleClientTests: XCTestCase {
         }
     }
 
+    func testTransactionSuccess() async throws {
+        let config = try OracleConnection.testConfig()
+        let client = OracleClient(configuration: config, drcp: false, backgroundLogger: .oracleTest)
+        let runTask = Task {
+            await client.run()
+        }
+        try await client.withTransaction { connection in
+            _ = try? await connection.execute("DROP TABLE test_pool_transaction_success")
+            try await connection.execute("CREATE TABLE test_pool_transaction_success (id NUMBER)")
+            let affectedRows = try await connection.executeBatch(
+                "INSERT INTO test_pool_transaction_success (id) VALUES (:1)",
+                binds: Array(1...10)
+            ).affectedRows
+            XCTAssertEqual(affectedRows, 10)
+        }
+        try await client.withConnection { connection in
+            let stream = try await connection.execute("SELECT id FROM test_pool_transaction_success")
+            var index = 0
+            for try await id in stream.decode(Int.self) {
+                index += 1
+                XCTAssertEqual(index, id)
+            }
+            XCTAssertEqual(index, 10)
+        }
+        runTask.cancel()
+    }
+
+    func testTransactionFailure() async throws {
+        let config = try OracleConnection.testConfig()
+        let client = OracleClient(configuration: config, drcp: false, backgroundLogger: .oracleTest)
+        let runTask = Task {
+            await client.run()
+        }
+        do {
+            try await client.withTransaction { connection in
+                _ = try? await connection.execute("DROP TABLE test_pool_transaction_failure")
+                try await connection.execute("CREATE TABLE test_pool_transaction_failure (id VARCHAR2(1 byte))")
+                try await connection.executeBatch(
+                    "INSERT INTO test_pool_transaction_failure (id) VALUES (:1)",
+                    binds: Array(1...10).map({ "\($0)" })
+                )
+            }
+        } catch let error as OracleSQLError {
+            XCTAssertEqual(error.serverInfo?.affectedRows, 9)
+        }
+        try await client.withConnection { connection in
+            let rows =
+                try await connection
+                .execute("SELECT id FROM test_pool_transaction_failure")
+                .collect()
+                .count
+            XCTAssertEqual(rows, 0)
+        }
+        runTask.cancel()
+    }
+
     @available(macOS 14.0, *)
     func testPingPong() async throws {
         let idleTimeout = Duration.seconds(20)