Merge pull request #50 from dpgao/master

Make each `SQLiteConnection` have its own SQLite handle

Sources/SQLite/Database/SQLiteConnection.swift

@@ -23,10 +23,9 @@ public final class SQLiteConnection: BasicWorker, DatabaseConnection, DatabaseQu
     public typealias Database = SQLiteDatabase
 
     /// See `DatabaseConnection`.
-    public var isClosed: Bool
-
-    /// See `BasicWorker`.
-    public let eventLoop: EventLoop
+    public var isClosed: Bool {
+        return handle == nil
+    }
 
     /// See `DatabaseConnection`.
     public var extend: Extend
@@ -36,32 +35,39 @@ public final class SQLiteConnection: BasicWorker, DatabaseConnection, DatabaseQu
 
     /// Reference to parent `SQLiteDatabase` that created this connection.
     /// This reference will ensure the DB stays alive since this connection uses
-    /// it's C pointer handle.
-    internal let database: SQLiteDatabase
+    /// it's thread pool.
+    private let database: SQLiteDatabase
+
+    /// Internal SQLite database handle.
+    internal private(set) var handle: OpaquePointer!
+
+    /// See `BasicWorker`.
+    public let eventLoop: EventLoop
 
     /// Create a new SQLite conncetion.
-    internal init(database: SQLiteDatabase, on worker: Worker) {
+    internal init(database: SQLiteDatabase, on worker: Worker) throws {
+        self.extend = [:]
         self.database = database
+        self.handle = try database.openConnection()
         self.eventLoop = worker.eventLoop
-        self.extend = [:]
-        self.isClosed = false
     }
 
     /// Returns an identifier for the last inserted row.
     public var lastAutoincrementID: Int64? {
-        return sqlite3_last_insert_rowid(database.handle)
+        return sqlite3_last_insert_rowid(handle)
     }
 
     /// Returns the last error message, if one exists.
     internal var errorMessage: String? {
-        guard let raw = sqlite3_errmsg(database.handle) else {
+        if let raw = sqlite3_errmsg(handle) {
+            return String(cString: raw)
+        } else {
             return nil
         }
-        return String(cString: raw)
     }
 
     /// See `SQLConnection`.
-    public func decode<D>(_ type: D.Type, from row: [SQLiteColumn : SQLiteData], table: GenericSQLTableIdentifier<SQLiteIdentifier>?) throws -> D where D : Decodable {
+    public func decode<D>(_ type: D.Type, from row: [SQLiteColumn: SQLiteData], table: GenericSQLTableIdentifier<SQLiteIdentifier>?) throws -> D where D : Decodable {
         return try SQLiteRowDecoder().decode(D.self, from: row, table: table)
     }
 
@@ -75,14 +81,16 @@ public final class SQLiteConnection: BasicWorker, DatabaseConnection, DatabaseQu
     ///     - query: `SQLiteQuery` to execute.
     ///     - onRow: Callback for handling each row.
     /// - returns: A `Future` that signals completion of the query.
-    public func query(_ query: SQLiteQuery, _ onRow: @escaping ([SQLiteColumn: SQLiteData]) throws -> ()) -> Future<Void> {
+    public func query(_ query: SQLiteQuery, _ onRow: @escaping ([SQLiteColumn: SQLiteData]) throws -> Void) -> Future<Void> {
         var binds: [Encodable] = []
         let sql = query.serialize(&binds)
-        let promise = eventLoop.newPromise(Void.self)
-        let data = try! binds.map { try SQLiteDataEncoder().encode($0) }
+        let data = try! binds.map(SQLiteDataEncoder().encode)
+
         // log before anything happens, in case there's an error
-        logger?.record(query: sql, values: data.map { $0.description })
-        database.blockingIO.submit { state in
+        logger?.record(query: sql, values: data.map(String.init(describing:)))
+
+        let promise = eventLoop.newPromise(Void.self)
+        database.blockingIO.submit { _ in
             do {
                 let statement = try SQLiteStatement(query: sql, on: self)
                 try statement.bind(data)
@@ -97,16 +105,21 @@ public final class SQLiteConnection: BasicWorker, DatabaseConnection, DatabaseQu
                         }
                     }
                 }
-                return promise.succeed(result: ())
+                self.eventLoop.execute {
+                    promise.succeed()
+                }
             } catch {
-                return promise.fail(error: error)
+                self.eventLoop.execute {
+                    promise.fail(error: error)
+                }
             }
         }
         return promise.futureResult
     }
 
     /// See `DatabaseConnection`.
     public func close() {
-        isClosed = true
+        sqlite3_close(handle)
+        handle = nil
     }
 }

Sources/SQLite/Database/SQLiteDatabase.swift

@@ -20,8 +20,9 @@ public final class SQLiteDatabase: Database, LogSupporting {
     /// Thread pool for performing blocking IO work. See `BlockingIOThreadPool`.
     internal let blockingIO: BlockingIOThreadPool
 
-    /// Internal SQLite database handle.
-    internal let handle: OpaquePointer
+    /// If the database uses in-memory storage, this property will be set to
+    /// keep the database alive when there is no `SQLiteConnection` to it.
+    private var handle: OpaquePointer?
 
     /// Create a new SQLite database.
     ///
@@ -34,34 +35,53 @@ public final class SQLiteDatabase: Database, LogSupporting {
     public init(storage: SQLiteStorage = .memory, threadPool: BlockingIOThreadPool? = nil) throws {
         self.storage = storage
         self.blockingIO = threadPool ?? BlockingIOThreadPool(numberOfThreads: 2)
-        self.blockingIO.start()
-        // make connection
-        let options = SQLITE_OPEN_CREATE | SQLITE_OPEN_READWRITE | SQLITE_OPEN_FULLMUTEX
+        if case .memory = storage {
+            self.handle = try openConnection()
+        }
+    }
+
+    // Make database connection
+    internal func openConnection() throws -> OpaquePointer {
+        let path: String
+        switch storage {
+        case .memory:
+            path = "file:\(ObjectIdentifier(self))?mode=memory&cache=shared"
+        case .file(let file):
+            path = file
+        }
         var handle: OpaquePointer?
-        guard sqlite3_open_v2(self.storage.path, &handle, options, nil) == SQLITE_OK, let c = handle else {
-            throw SQLiteError(problem: .error, reason: "Could not open database.", source: .capture())
+        let options = SQLITE_OPEN_CREATE | SQLITE_OPEN_READWRITE | SQLITE_OPEN_FULLMUTEX
+        guard sqlite3_open_v2(path, &handle, options, nil) == SQLITE_OK,
+            let c = handle,
+            sqlite3_busy_handler(c, { _, _ in 1 }, nil) == SQLITE_OK else {
+                throw SQLiteError(problem: .error, reason: "Could not open database.", source: .capture())
         }
-        self.handle = c
+        return c
     }
 
     /// See `Database`.
     public func newConnection(on worker: Worker) -> Future<SQLiteConnection> {
-        let conn = SQLiteConnection(database: self, on: worker)
-        return worker.future(conn)
+        do {
+            let conn = try SQLiteConnection(database: self, on: worker)
+            return worker.future(conn)
+        } catch {
+            return worker.future(error: error)
+        }
     }
 
     /// See `LogSupporting`.
     public static func enableLogging(_ logger: DatabaseLogger, on conn: SQLiteConnection) {
         conn.logger = logger
     }
 
-    /// Closes the open SQLite handle on deinit.
     deinit {
-        sqlite3_close(handle)
-        self.blockingIO.shutdownGracefully { error in
+        self.blockingIO.shutdownGracefully { [handle] error in
             if let error = error {
                 print("[SQLite] [ERROR] Could not shutdown BlockingIOThreadPool: \(error)")
             }
+            if let handle = handle, sqlite3_close(handle) != SQLITE_OK {
+                print("[SQLite] [ERROR] Could not close database.")
+            }
         }
     }
 }

Sources/SQLite/Database/SQLiteStatement.swift

@@ -5,17 +5,17 @@ import SQLite3
 #endif
 
 internal struct SQLiteStatement {
-    internal let connection: SQLiteConnection
-    internal let c: OpaquePointer
+    private let c: OpaquePointer
+    private let connection: SQLiteConnection
 
     internal init(query: String, on connection: SQLiteConnection) throws {
         var handle: OpaquePointer?
-        let ret = sqlite3_prepare_v2(connection.database.handle, query, -1, &handle, nil)
+        let ret = sqlite3_prepare_v2(connection.handle, query, -1, &handle, nil)
         guard ret == SQLITE_OK, let c = handle else {
             throw SQLiteError(statusCode: ret, connection: connection, source: .capture())
         }
-        self.connection = connection
         self.c = c
+        self.connection = connection
     }
 
     internal func bind(_ binds: [SQLiteData]) throws {
@@ -52,7 +52,6 @@ internal struct SQLiteStatement {
                 }
             }
         }
-
     }
 
     internal func getColumns() throws -> [SQLiteColumn]? {
@@ -85,15 +84,13 @@ internal struct SQLiteStatement {
         case SQLITE_ROW: break
         default: throw SQLiteError(statusCode: step, connection: connection, source: .capture())
         }
-
 
         var row: [SQLiteColumn: SQLiteData] = [:]
 
         // iterator over column count again and create a field
         // for each column. Use the column we have already initialized.
-        for i in 0..<Int32(columns.count) {
-            let col = columns[Int(i)]
-            row[col] = try data(at: i)
+        for (i, col) in columns.enumerated() {
+            row[col] = try data(at: Int32(i))
         }
 
         // return to event loop

Sources/SQLite/Database/SQLiteStorage.swift

@@ -6,11 +6,4 @@ public enum SQLiteStorage {
 
     /// File-based storage, persisted between application launches.
     case file(path: String)
-
-    internal var path: String {
-        switch self {
-        case .memory: return ":memory:"
-        case .file(let path): return path
-        }
-    }
 }