event loop preference

Package.swift

@@ -10,10 +10,11 @@ let package = Package(
     ],
     dependencies: [
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.0.0"),
+        .package(url: "https://github.com/apple/swift-log.git", from: "1.0.0"),
         .package(url: "https://github.com/vapor/sql-kit.git", from: "3.0.0-alpha"),
     ],
     targets: [
-        .target(name: "FluentKit", dependencies: ["NIO"]),
+        .target(name: "FluentKit", dependencies: ["NIO", "Logging"]),
         .target(name: "FluentBenchmark", dependencies: ["FluentKit"]),
         .target(name: "FluentSQL", dependencies: ["FluentKit", "SQLKit"]),
         .testTarget(name: "FluentKitTests", dependencies: ["FluentBenchmark", "FluentSQL"]),

Sources/FluentBenchmark/FluentBenchmarker.swift

@@ -317,12 +317,13 @@ public final class FluentBenchmarker {
             migrations.add(GalaxyMigration())
             migrations.add(PlanetMigration())
 
-            var databases = Databases(on: self.database.eventLoop)
-            databases.add(self.database, as: .init(string: "main"))
+            let databases = Databases()
+            databases.add(self.database.driver, as: .init(string: "main"))
 
             var migrator = Migrator(
                 databases: databases,
                 migrations: migrations,
+                logger: .init(label: "codes.vapor.fluent.test"),
                 on: self.database.eventLoop
             )
             try migrator.setupIfNeeded().wait()
@@ -348,12 +349,13 @@ public final class FluentBenchmarker {
             migrations.add(ErrorMigration())
             migrations.add(PlanetMigration())
 
-            var databases = Databases(on: self.database.eventLoop)
-            databases.add(self.database, as: .init(string: "main"))
+            let databases = Databases()
+            databases.add(self.database.driver, as: .init(string: "main"))
 
             let migrator = Migrator(
                 databases: databases,
                 migrations: migrations,
+                logger: .init(label: "codes.vapor.fluent.test"),
                 on: self.database.eventLoop
             )
             try migrator.setupIfNeeded().wait()
@@ -606,20 +608,19 @@ public final class FluentBenchmarker {
         try runTest(#function, [
             GalaxyMigration(),
         ]) {
-            var fetched64: [Galaxy] = []
-            var fetched2047: [Galaxy] = []
+            var fetched64: [Result<Galaxy, Error>] = []
+            var fetched2047: [Result<Galaxy, Error>] = []
 
-            try self.database.withConnection { database -> EventLoopFuture<Void> in
-                let saves = (1...512).map { i -> EventLoopFuture<Void> in
-                    return Galaxy(name: "Milky Way \(i)")
-                        .save(on: database)
-                }
-                return .andAllSucceed(saves, on: database.eventLoop)
-            }.wait()
+            let saves = (1...512).map { i -> EventLoopFuture<Void> in
+                return Galaxy(name: "Milky Way \(i)")
+                    .save(on: self.database)
+            }
+            try EventLoopFuture<Void>.andAllSucceed(saves, on: self.database.eventLoop).wait()
 
             try Galaxy.query(on: self.database).chunk(max: 64) { chunk in
                 guard chunk.count == 64 else {
-                    throw Failure("bad chunk count")
+                    XCTFail("bad chunk count")
+                    return
                 }
                 fetched64 += chunk
             }.wait()
@@ -630,7 +631,8 @@ public final class FluentBenchmarker {
 
             try Galaxy.query(on: self.database).chunk(max: 511) { chunk in
                 guard chunk.count == 511 || chunk.count == 1 else {
-                    throw Failure("bad chunk count")
+                    XCTFail("bad chunk count")
+                    return
                 }
                 fetched2047 += chunk
             }.wait()
@@ -1322,7 +1324,16 @@ public final class FluentBenchmarker {
                 default:
                     XCTFail("unexpected name: \(user.name)")
                 }
-            }        
+            }
+
+            // test query with no ids
+            // https://github.com/vapor/fluent-kit/issues/85
+            let users2 = try User.query(on: self.database)
+                .with(\.$bestFriend)
+                .filter(\.$bestFriend == nil)
+                .all().wait()
+            XCTAssertEqual(users2.count, 1)
+            XCTAssert(users2.first?.bestFriend == nil)
         }
     }
 

Sources/FluentKit/Database/Database.swift

@@ -1,16 +1,71 @@
+public enum EventLoopPreference {
+    case indifferent
+    case delegate(on: EventLoop)
+}
+
 public protocol Database {
-    var eventLoop: EventLoop { get }
+    var driver: DatabaseDriver { get }
+    var logger: Logger { get }
+    var eventLoopPreference: EventLoopPreference { get }
+}
+
+private struct DriverOverrideDatabase: Database {
+    var logger: Logger {
+        return self.base.logger
+    }
 
-    func execute(
-        _ query: DatabaseQuery,
-        _ onOutput: @escaping (DatabaseOutput) throws -> ()
-    ) -> EventLoopFuture<Void>
+    var eventLoopPreference: EventLoopPreference {
+        return self.base.eventLoopPreference
+    }
 
-    func execute(_ schema: DatabaseSchema) -> EventLoopFuture<Void>
+    let base: Database
+    let driver: DatabaseDriver
 
-    func close() -> EventLoopFuture<Void>
+    init(base: Database, driver: DatabaseDriver) {
+        self.base = base
+        self.driver = driver
+    }
+}
+
+extension Database {
+    public var eventLoop: EventLoop {
+        switch self.eventLoopPreference {
+        case .indifferent:
+            return self.driver.eventLoopGroup.next()
+        case .delegate(let eventLoop):
+            return eventLoop
+        }
+    }
 
-    func withConnection<T>(_ closure: @escaping (Database) -> EventLoopFuture<T>) -> EventLoopFuture<T>
+    var hopEventLoop: EventLoop? {
+        switch self.eventLoopPreference {
+        case .delegate(let eventLoop):
+            if !eventLoop.inEventLoop {
+                return eventLoop
+            } else {
+                return nil
+            }
+        case .indifferent:
+            return nil
+        }
+    }
+}
+
+public protocol DatabaseDriver {
+    var eventLoopGroup: EventLoopGroup { get }
+
+    func execute(
+        query: DatabaseQuery,
+        database: Database,
+        onRow: @escaping (DatabaseRow) -> ()
+    ) -> EventLoopFuture<Void>
+
+    func execute(
+        schema: DatabaseSchema,
+        database: Database
+    ) -> EventLoopFuture<Void>
+
+    func shutdown()
 }
 
 public protocol DatabaseError {

Sources/FluentKit/Database/DatabaseOutput.swift

@@ -1,57 +1,82 @@
-public protocol DatabaseOutput: CustomStringConvertible {
+public struct DatabaseOutput {
+    public let database: Database
+    public let row: DatabaseRow
+
+    public func contains(_ field: String) -> Bool {
+        return self.row.contains(field: field)
+    }
+
+    public func decode<T>(_ field: String, as: T.Type = T.self) throws -> T
+        where T: Decodable
+    {
+        return try self.row.decode(field: field, as: T.self, for: self.database)
+    }
+}
+
+public protocol DatabaseRow: CustomStringConvertible {
     func contains(field: String) -> Bool
-    func decode<T>(field: String, as type: T.Type) throws -> T
+    func decode<T>(
+        field: String,
+        as type: T.Type,
+        for database: Database
+    ) throws -> T
         where T: Decodable
 }
 
-extension DatabaseOutput {
-    func prefixed(by string: String) -> DatabaseOutput {
-        return PrefixingOutput(self, prefix: string)
+extension DatabaseRow {
+    public func output(for database: Database) -> DatabaseOutput {
+        return .init(database: database, row: self)
     }
 }
 
-private struct PrefixingOutput: DatabaseOutput {
-    let wrapped: DatabaseOutput
+extension DatabaseRow {
+    func prefixed(by string: String) -> DatabaseRow {
+        return PrefixingOutput(wrapped: self, prefix: string)
+    }
+}
 
+private struct PrefixingOutput: DatabaseRow {
+    let wrapped: DatabaseRow
     let prefix: String
 
     var description: String {
         return self.wrapped.description
     }
 
-    init(_ wrapped: DatabaseOutput, prefix: String) {
-        self.wrapped = wrapped
-        self.prefix = prefix
-    }
-
     func contains(field: String) -> Bool {
         return self.wrapped.contains(field: self.prefix + field)
     }
 
-    func decode<T>(field: String, as type: T.Type) throws -> T where T : Decodable {
-        return try self.wrapped.decode(field: self.prefix + field, as: T.self)
+    func decode<T>(
+        field: String,
+        as type: T.Type,
+        for database: Database
+    ) throws -> T where T : Decodable {
+        return try self.wrapped.decode(field: self.prefix + field, as: T.self, for: database)
     }
 }
 
-extension DatabaseOutput {
-    func cascading(to output: DatabaseOutput) -> DatabaseOutput {
+extension DatabaseRow {
+    func cascading(to output: DatabaseRow) -> DatabaseRow {
         return CombinedOutput(first: self, second: output)
     }
 }
 
-private struct CombinedOutput: DatabaseOutput {
-    var first: DatabaseOutput
-    var second: DatabaseOutput
+private struct CombinedOutput: DatabaseRow {
+    var first: DatabaseRow
+    var second: DatabaseRow
 
     func contains(field: String) -> Bool {
         return self.first.contains(field: field) || self.second.contains(field: field)
     }
 
-    func decode<T>(field: String, as type: T.Type) throws -> T where T : Decodable {
+    func decode<T>(field: String, as type: T.Type, for database: Database) throws -> T
+        where T : Decodable
+    {
         if self.first.contains(field: field) {
-            return try self.first.decode(field: field, as: T.self)
+            return try self.first.decode(field: field, as: T.self, for: database)
         } else if self.second.contains(field: field) {
-            return try self.second.decode(field: field, as: T.self)
+            return try self.second.decode(field: field, as: T.self, for: database)
         } else {
             throw FluentError.missingField(name: field)
         }

Sources/FluentKit/Database/Databases.swift

@@ -1,21 +1,23 @@
 import Foundation
 
-public struct Databases {
+public final class Databases {
     private var storage: [DatabaseID: Database]
-
     private var _default: Database?
 
-    public let eventLoop: EventLoop
-
-    public init(on eventLoop: EventLoop) {
+    public init() {
         self.storage = [:]
-        self.eventLoop = eventLoop
     }
 
-    public mutating func add(_ database: Database, as id: DatabaseID, isDefault: Bool = true) {
-        self.storage[id] = database
+    public func add(
+        _ driver: DatabaseDriver,
+        logger: Logger = .init(label: "codes.vapor.db"),
+        as id: DatabaseID,
+        isDefault: Bool = true
+    ) {
+        let db = BasicDatabase(driver: driver, logger: logger)
+        self.storage[id] = db
         if isDefault {
-            self._default = database
+            self._default = db
         }
     }
 
@@ -27,7 +29,18 @@ public struct Databases {
         return self._default!
     }
 
-    public func close() -> EventLoopFuture<Void> {
-        return .andAllSucceed(self.storage.values.map { $0.close() }, on: self.eventLoop)
+    public func shutdown() {
+        for db in self.storage.values {
+            db.driver.shutdown()
+        }
     }
 }
+
+private struct BasicDatabase: Database {
+    var eventLoopPreference: EventLoopPreference {
+        return .indifferent
+    }
+
+    let driver: DatabaseDriver
+    let logger: Logger
+}