Add Combine publishers for reading, writing, and executing queries in a transaction (#23)

Author: atdrendel

Package.resolved

@@ -12,7 +12,16 @@ let package = Package(
             targets: ["SQLite"]),
     ],
     dependencies: [
-        .package(name: "Synchronized", url: "https://github.com/shareup/synchronized.git", from: "2.1.0")
+        .package(
+            name: "Synchronized",
+            url: "https://github.com/shareup/synchronized.git",
+            from: "2.2.0"
+        ),
+        .package(
+            name: "CombineExtensions",
+            url: "https://github.com/shareup/combine-extensions.git",
+            from: "2.3.0"
+        )
     ],
     targets: [
         .target(
@@ -23,6 +32,7 @@ let package = Package(
         .testTarget(
             name: "SQLiteTests",
             dependencies: [
+                .product(name: "CombineTestExtensions", package: "CombineExtensions"),
                 "SQLite",
             ]
         ),

Package.swift

@@ -30,7 +30,13 @@ public final class SQLiteDatabase {
     private var _isOpen: Bool
 
     private lazy var _queue: DispatchQueue = {
-        let queue = DispatchQueue(label: "Database._queue")
+        let queue = DispatchQueue(
+            label: "app.shareup.sqlite.sqlitedatabase.queue",
+            qos: .default,
+            attributes: [],
+            autoreleaseFrequency: .workItem,
+            target: .global()
+        )
         queue.setSpecific(key: self._queueKey, value: self._queueContext)
         return queue
     }()
@@ -61,7 +67,103 @@ public final class SQLiteDatabase {
             SQLiteDatabase.close(_connection)
         }
     }
+}
+
+// MARK: -
+// MARK: Asynchronous queries
+
+extension SQLiteDatabase {
+    public func inTransactionPublisher<T>(
+        _ block: @escaping (SQLiteDatabase) throws -> T
+    ) -> AnyPublisher<T, SQLiteError> {
+        SQLiteFuture { [_queue] (promise) in
+            _queue.async { [weak self] in
+                guard let self = self else {
+                    promise(.failure(.onExecuteQueryAfterDeallocating))
+                    return
+                }
+
+                do {
+                    let result = try self.inTransaction(block)
+                    promise(.success(result))
+                } catch let error as SQLiteError {
+                    promise(.failure(error))
+                } catch {
+                    promise(.failure(.onInternalError(error.localizedDescription)))
+                }
+            }
+        }
+        .eraseToAnyPublisher()
+    }
+
+    public func writePublisher(
+        _ sql: SQL,
+        arguments: SQLiteArguments = [:]
+    ) -> AnyPublisher<Void, SQLiteError> {
+        let prepareStatement = { [unowned self] () throws -> OpaquePointer in
+            try self.cachedStatement(for: sql)
+        }
+
+        let resetStatement = { (statement: OpaquePointer) -> Void in
+            statement.resetAndClearBindings()
+        }
 
+        return _executeAsync(
+            sql,
+            arguments: arguments,
+            prepareStatement: prepareStatement,
+            resetStatement: resetStatement
+        )
+        .tryMap { rows in throw SQLiteError.onWrite(rows) }
+        .mapError { (error) -> SQLiteError in
+            if let sqliteError = error as? SQLiteError {
+                return sqliteError
+            } else {
+                return .onInternalError(error.localizedDescription)
+            }
+        }
+        .eraseToAnyPublisher()
+    }
+
+    public func readPublisher(_ sql: SQL, arguments: SQLiteArguments = [:]) -> AnyPublisher<[SQLiteRow], SQLiteError> {
+        let prepareStatement = { [unowned self] () throws -> OpaquePointer in
+            try self.cachedStatement(for: sql)
+        }
+
+        let resetStatement = { (statement: OpaquePointer) -> Void in
+            statement.resetAndClearBindings()
+        }
+
+        return _executeAsync(
+            sql,
+            arguments: arguments,
+            prepareStatement: prepareStatement,
+            resetStatement: resetStatement
+        )
+        .eraseToAnyPublisher()
+    }
+
+    public func readPublisher<T: SQLiteTransformable>(
+        _ sql: SQL,
+        arguments: SQLiteArguments
+    ) -> AnyPublisher<[T], SQLiteError> {
+        readPublisher(sql, arguments: arguments)
+            .tryMap { try $0.map { try T.init(row: $0) } }
+            .mapError { (error) -> SQLiteError in
+                if let sqliteError = error as? SQLiteError {
+                    return sqliteError
+                } else {
+                    return .onInternalError(error.localizedDescription)
+                }
+            }
+            .eraseToAnyPublisher()
+    }
+}
+
+// MARK: -
+// MARK: Synchronous queries
+
+extension SQLiteDatabase {
     public func inTransaction<T>(_ block: (SQLiteDatabase) throws -> T) rethrows -> T {
         return try sync {
             _transactionCount += 1
@@ -125,6 +227,9 @@ public final class SQLiteDatabase {
     }
 }
 
+// MARK: -
+// MARK: Tables and columns
+
 extension SQLiteDatabase {
     public func tables() throws -> Array<String> {
         return try sync {
@@ -156,6 +261,9 @@ extension SQLiteDatabase {
     }
 }
 
+// MARK: -
+// MARK: Combine Publishers observing SQL queries
+
 extension SQLiteDatabase {
     public func publisher(
         _ sql: SQL,
@@ -191,6 +299,9 @@ extension SQLiteDatabase {
     }
 }
 
+// MARK: -
+// MARK: Block-based observation of SQL queries
+
 extension SQLiteDatabase {
     func observe(
         _ sql: SQL,
@@ -249,12 +360,18 @@ extension SQLiteDatabase {
     }
 }
 
+// MARK: -
+// MARK: Equatable
+
 extension SQLiteDatabase: Equatable {
     public static func == (lhs: SQLiteDatabase, rhs: SQLiteDatabase) -> Bool {
         return lhs._connection == rhs._connection
     }
 }
 
+// MARK: -
+// MARK: Compile-time SQLite options
+
 extension SQLiteDatabase {
     public var supportsJSON: Bool {
         return isCompileOptionEnabled("ENABLE_JSON1")
@@ -265,6 +382,9 @@ extension SQLiteDatabase {
     }
 }
 
+// MARK: -
+// MARK: Vacuuming
+
 extension SQLiteDatabase {
     public enum AutoVacuumMode: Int {
         case none = 0
@@ -313,6 +433,9 @@ extension SQLiteDatabase {
     }
 }
 
+// MARK: -
+// MARK: SQLite hooks
+
 extension SQLiteDatabase {
     func createUpdateHandler(_ block: @escaping (String) -> Void) {
         let updateBlock: UpdateHookCallback = { _, _, _, tableName, _ in
@@ -360,6 +483,9 @@ extension SQLiteDatabase {
     }
 }
 
+// MARK: -
+// MARK: Private
+
 extension SQLiteDatabase {
     private func _observe(
         _ sql: SQL,
@@ -389,10 +515,45 @@ extension SQLiteDatabase {
 }
 
 extension SQLiteDatabase {
-    private func _execute(_ sql: SQL, statement: OpaquePointer,
-                          arguments: SQLiteArguments) throws -> Array<SQLiteRow> {
+    private func _executeAsync(
+        _ sql: SQL,
+        arguments: SQLiteArguments,
+        prepareStatement: @escaping () throws -> OpaquePointer,
+        resetStatement: @escaping (OpaquePointer) -> Void
+    ) -> SQLiteFuture<[SQLiteRow]> {
+        SQLiteFuture { [_queue] (promise) in
+            _queue.async { [weak self] in
+                guard let self = self else {
+                    promise(.failure(.onExecuteQueryAfterDeallocating))
+                    return
+                }
+
+                do {
+                    let statement = try prepareStatement()
+                    defer { resetStatement(statement) }
+
+                    let result = try self._execute(
+                        sql,
+                        statement: statement,
+                        arguments: arguments
+                    )
+                    promise(.success(result))
+                } catch let error as SQLiteError {
+                    promise(.failure(error))
+                } catch {
+                    promise(.failure(.onInternalError(error.localizedDescription)))
+                }
+            }
+        }
+    }
+
+    private func _execute(
+        _ sql: SQL,
+        statement: OpaquePointer,
+        arguments: SQLiteArguments
+    ) throws -> [SQLiteRow] {
         assert(isOnDatabaseQueue)
-        guard _isOpen else { assertionFailure("Database is closed"); return [] }
+        guard _isOpen else { throw SQLiteError.onExecuteQueryWithoutOpenDatabase }
 
         try statement.bind(arguments: arguments)
         let (result, output) = try statement.evaluate()

Sources/SQLite/SQLiteDatabase.swift

@@ -1,7 +1,7 @@
 import Foundation
 import SQLite3
 
-public enum SQLiteError: Error {
+public enum SQLiteError: Error, Equatable {
     case onInternalError(String)
     case onOpen(Int32, String)
     case onClose(Int32)
@@ -11,6 +11,9 @@ public enum SQLiteError: Error {
     case onStep(Int32, String)
     case onWrite(Array<SQLiteRow>)
     case onGetColumnType(Int32)
+    case onBeginTransactionAfterDeallocating
+    case onExecuteQueryWithoutOpenDatabase
+    case onExecuteQueryAfterDeallocating
     case onCreateFunction(String, Int32)
     case onRemoveFunction(String, Int32)
     case onGetColumnInTable(String)
@@ -44,12 +47,18 @@ extension SQLiteError: CustomStringConvertible {
             return "Could not get index for '\(parameterName)'"
         case .onBindParameter(let code, let index, let value):
             return "Could not bind \(value) to \(index): \(string(for: code))"
-        case .onStep(let code, let sql):
-            return "Could not execute SQL '\(sql)': \(string(for: code))"
+        case .onBeginTransactionAfterDeallocating:
+            return "Tried to begin a transaction after deallocating"
+        case .onExecuteQueryWithoutOpenDatabase:
+            return "Tried to execute a query without an open database connection"
+        case .onExecuteQueryAfterDeallocating:
+            return "Tried to execute a query after deallocating"
         case .onWrite(let result):
             return "Write returned results: '\(result)'"
         case .onGetColumnType(let type):
             return "Invalid column type: \(type)"
+        case .onStep(let code, let sql):
+            return "Could not execute SQL '\(sql)': \(string(for: code))"
         case .onCreateFunction(let name, let code):
             return "Could not create function '\(name)': \(string(for: code))"
         case .onRemoveFunction(let name, let code):