Merge pull request #169 from vapor/dbkit-gm

dbkit 1.0.0 gm

Author: tanner0101

Package.swift

@@ -14,7 +14,7 @@ let package = Package(
         .package(url: "https://github.com/vapor/crypto.git", from: "3.0.0-rc.2"),
 
         // 🗄 Core services for creating database integrations.
-        .package(url: "https://github.com/vapor/database-kit.git", from: "1.0.0-rc.2"),
+        .package(url: "https://github.com/vapor/database-kit.git", from: "1.0.0"),
 
         // 📦 Dependency injection / inversion of control framework.
         .package(url: "https://github.com/vapor/service.git", from: "1.0.0-rc.2"),

Sources/MySQL/Connection/MySQLConnection+Authenticate.swift

@@ -22,10 +22,13 @@ extension MySQLConnection {
             guard let handshake = handshake else {
                 throw MySQLError(identifier: "handshake", reason: "Handshake required for auth response.", source: .capture())
             }
-            let authPlugin = handshake.authPluginName ?? "none"
+            let authPlugin = handshake.authPluginName
             let authResponse: Data
             switch authPlugin {
-            case "mysql_native_password":
+            case .some("mysql_native_password"), .none:
+                guard handshake.capabilities.get(CLIENT_SECURE_CONNECTION) else {
+                    throw MySQLError(identifier: "authproto", reason: "Pre-4.1 auth protocol is not supported or safe.", source: .capture())
+                }
                 guard let password = password else {
                     throw MySQLError(identifier: "password", reason: "Password required for auth plugin.", source: .capture())
                 }
@@ -40,7 +43,7 @@ extension MySQLConnection {
                     hash[i] = hash[i] ^ passwordHash[i]
                 }
                 authResponse = hash
-            default: throw MySQLError(identifier: "authPlugin", reason: "Unsupported auth plugin: \(authPlugin)", source: .capture())
+            default: throw MySQLError(identifier: "authPlugin", reason: "Unsupported auth plugin: \(authPlugin ?? "<none>")", source: .capture())
             }
             let response = MySQLHandshakeResponse41(
                 capabilities: [
@@ -55,7 +58,7 @@ extension MySQLConnection {
                 username: username,
                 authResponse: authResponse,
                 database: database,
-                authPluginName: authPlugin
+                authPluginName: authPlugin ?? "<none>"
             )
             return self.send([.handshakeResponse41(response)]) { message in
                 switch message {

Sources/MySQL/Connection/MySQLConnection.swift

@@ -7,11 +7,17 @@ import Service
 
 /// A MySQL frontend client.
 public final class MySQLConnection: BasicWorker, DatabaseConnection {
-    /// See `Worker.eventLoop`
+    /// See `Worker`.
     public var eventLoop: EventLoop {
         return channel.eventLoop
     }
 
+    /// See `DatabaseConnection`.
+    public var isClosed: Bool
+
+    /// See `Extendable`
+    public var extend: Extend
+
     /// Handles enqueued redis commands and responses.
     private let queue: QueueHandler<MySQLPacket, MySQLPacket>
 
@@ -24,39 +30,68 @@ public final class MySQLConnection: BasicWorker, DatabaseConnection {
     /// The current query running, if one exists.
     private var pipeline: Future<Void>
 
-    /// See `Extendable.extend`
-    public var extend: Extend
+    /// Currently running `send(...)`.
+    private var currentSend: Promise<Void>?
 
     /// Creates a new MySQL client with the provided MySQL packet queue and channel.
     init(queue: QueueHandler<MySQLPacket, MySQLPacket>, channel: Channel) {
         self.queue = queue
         self.channel = channel
         self.pipeline = Future.map(on: channel.eventLoop) { }
         self.extend = [:]
+        self.isClosed = false
+
+        // when the channel closes, set isClosed to true and fail any
+        // currently running calls to `send(...)`.
+        channel.closeFuture.always {
+            self.isClosed = true
+            if let current = self.currentSend {
+                current.fail(error: closeError)
+            }
+        }
     }
 
     /// Sends `MySQLPacket` to the server.
     internal func send(_ messages: [MySQLPacket], onResponse: @escaping (MySQLPacket) throws -> Bool) -> Future<Void> {
-        return queue.enqueue(messages) { message in
+        // if currentSend is not nil, previous send has not completed
+        assert(currentSend == nil, "Attempting to call `send(...)` again before previous invocation has completed.")
+
+        // if the connection is closed, fail immidiately
+        guard !isClosed else {
+            return eventLoop.newFailedFuture(error: closeError)
+        }
+
+        // create a new promise and store it
+        let promise = eventLoop.newPromise(Void.self)
+        currentSend = promise
+
+        // cascade this enqueue to the newly created promise
+        queue.enqueue(messages) { message in
             switch message {
             case .err(let err): throw err.makeError(source: .capture())
             default: return try onResponse(message)
             }
-        }
+        }.cascade(promise: promise)
+
+        // when the promise completes, remove the reference to it
+        promise.futureResult.always { self.currentSend = nil }
+
+        // return the promise's future result (same as `queue.enqueue`)
+        return promise.futureResult
     }
 
     /// Submits an async task to be pipelined.
     internal func operation(_ work: @escaping () -> Future<Void>) -> Future<Void> {
-        /// perform this work when the current pipeline future is completed
+        // perform this work when the current pipeline future is completed
         let new = pipeline.then(work)
 
-        /// append this work to the pipeline, discarding errors as the pipeline
-        //// does not care about them
+        // append this work to the pipeline, discarding errors as the pipeline
+        // does not care about them
         pipeline = new.catchMap { err in
             return ()
         }
 
-        /// return the newly enqueued work's future result
+        // return the newly enqueued work's future result
         return new
     }
 
@@ -65,3 +100,6 @@ public final class MySQLConnection: BasicWorker, DatabaseConnection {
         channel.close(promise: nil)
     }
 }
+
+/// Error to throw if the connection has closed.
+private let closeError = MySQLError(identifier: "closed", reason: "Connection is closed.", source: .capture())

Sources/MySQL/Connection/MySQLData.swift

@@ -54,10 +54,16 @@ public struct MySQLData: Equatable {
         let storage: MySQLBinaryDataStorage?
 
         if let integer = integer {
-            if I.isSigned {
-                storage = .integer8(numericCast(integer))
-            } else {
-                storage = .uinteger8(numericCast(integer))
+            switch (I.bitWidth, I.isSigned) {
+                case ( 8, true):  storage = .integer1(numericCast(integer))
+                case ( 8, false): storage = .uinteger1(numericCast(integer))
+                case (16, true):  storage = .integer2(numericCast(integer))
+                case (16, false): storage = .uinteger2(numericCast(integer))
+                case (32, true):  storage = .integer4(numericCast(integer))
+                case (32, false): storage = .uinteger4(numericCast(integer))
+                case (64, true):  storage = .integer8(numericCast(integer))
+                case (64, false): storage = .uinteger8(numericCast(integer))
+                default: fatalError("Unsupported bit-width: \(I.bitWidth)")
             }
         } else {
             storage = nil

Sources/MySQL/Database/MySQLDatabase.swift

@@ -11,8 +11,8 @@ public final class MySQLDatabase: Database {
         self.config = config
     }
 
-    /// See `Database.makeConnection()`
-    public func makeConnection(on worker: Worker) -> Future<MySQLConnection> {
+    /// See `Database`
+    public func newConnection(on worker: Worker) -> Future<MySQLConnection> {
         let config = self.config
         return Future.flatMap(on: worker) {
             return try MySQLConnection.connect(hostname: config.hostname, port: config.port, on: worker) { error in

Sources/MySQL/Database/MySQLLogger.swift

@@ -7,10 +7,7 @@ public protocol MySQLLogger {
 }
 
 extension DatabaseLogger: MySQLLogger {
-    /// See MySQLLogger.log
     public func log(query: String) {
-        let log = DatabaseLog(query: query)
-        record(log: log)
+        record(query: query, values: [])
     }
 }
-

Sources/MySQL/Database/MySQLProvider.swift

@@ -13,7 +13,7 @@ public final class MySQLProvider: Provider {
         try services.register(DatabaseKitProvider())
         services.register(MySQLDatabaseConfig.self)
         services.register(MySQLDatabase.self)
-        var databases = DatabaseConfig()
+        var databases = DatabasesConfig()
         databases.add(database: MySQLDatabase.self, as: .mysql)
         services.register(databases)
     }

Sources/MySQL/Pipeline/MySQLConnectionSession.swift

@@ -80,6 +80,7 @@ enum MySQLStatementProtocolState {
 
     case waitingExecute
     case rowColumns(columns: [MySQLColumnDefinition41], remaining: Int)
+    case rowColumnsDone(columns: [MySQLColumnDefinition41])
     /// ProtocolBinary::ResultsetRow until eof
     case rows(columns: [MySQLColumnDefinition41])
 }

Sources/MySQL/Pipeline/MySQLPacketDecoder.swift

@@ -17,6 +17,11 @@ final class MySQLPacketDecoder: ByteToMessageDecoder {
         self.session = session
     }
 
+    func channelInactive(ctx: ChannelHandlerContext) {
+        cumulationBuffer = nil
+        ctx.fireChannelInactive()
+    }
+
     /// Decode from a `ByteBuffer`. This method will be called till either the input
     /// `ByteBuffer` has nothing to read left or `DecodingState.needMoreData` is returned.
     ///
@@ -52,7 +57,7 @@ final class MySQLPacketDecoder: ByteToMessageDecoder {
     }
 
     /// Decode's an OK, ERR, or EOF packet
-    func decodeBasicPacket(ctx: ChannelHandlerContext, buffer: inout ByteBuffer, capabilities: MySQLCapabilities) throws -> DecodingState {
+    func decodeBasicPacket(ctx: ChannelHandlerContext, buffer: inout ByteBuffer, capabilities: MySQLCapabilities, forwarding: Bool = true) throws -> DecodingState {
         guard let length = try buffer.checkPacketLength(source: .capture()) else {
             return .needMoreData
         }
@@ -84,7 +89,9 @@ final class MySQLPacketDecoder: ByteToMessageDecoder {
         }
 
         session.incrementSequenceID()
-        ctx.fireChannelRead(wrapInboundOut(packet))
+        if forwarding {
+            ctx.fireChannelRead(wrapInboundOut(packet))
+        }
 
         return .continue
     }
@@ -96,6 +103,15 @@ final class MySQLPacketDecoder: ByteToMessageDecoder {
         textState: MySQLTextProtocolState,
         capabilities: MySQLCapabilities
     ) throws -> DecodingState {
+        if !capabilities.get(CLIENT_DEPRECATE_EOF) {
+            // check for error or OK packet
+            let peek = buffer.peekInteger(as: Byte.self, skipping: 4)
+            switch peek {
+            case 0xFE: return try decodeBasicPacket(ctx: ctx, buffer: &buffer, capabilities: capabilities, forwarding: false)
+            default: break
+            }
+        }
+        
         switch textState {
         case .waiting:
             // check for error or OK packet
@@ -200,7 +216,7 @@ final class MySQLPacketDecoder: ByteToMessageDecoder {
             }
 
             if !capabilities.get(CLIENT_DEPRECATE_EOF) {
-                return try decodeBasicPacket(ctx: ctx, buffer: &buffer, capabilities: capabilities)
+                return try decodeBasicPacket(ctx: ctx, buffer: &buffer, capabilities: capabilities, forwarding: false)
             }
         case .columns(var remaining):
             guard let _ = try buffer.checkPacketLength(source: .capture()) else {
@@ -219,7 +235,7 @@ final class MySQLPacketDecoder: ByteToMessageDecoder {
             }
         case .columnsDone:
             if !capabilities.get(CLIENT_DEPRECATE_EOF) {
-                return try decodeBasicPacket(ctx: ctx, buffer: &buffer, capabilities: capabilities)
+                return try decodeBasicPacket(ctx: ctx, buffer: &buffer, capabilities: capabilities, forwarding: false)
             }
         case .waitingExecute:
             // check for error or OK packet
@@ -247,10 +263,17 @@ final class MySQLPacketDecoder: ByteToMessageDecoder {
             ctx.fireChannelRead(wrapInboundOut(.columnDefinition41(column)))
             remaining -= 1
             if remaining == 0 {
-                session.connectionState = .statement(.rows(columns: columns))
+                session.connectionState = .statement(.rowColumnsDone(columns: columns))
             } else {
                 session.connectionState = .statement(.rowColumns(columns: columns, remaining: remaining))
             }
+        case .rowColumnsDone(let columns):
+            if !capabilities.get(CLIENT_DEPRECATE_EOF) {
+                let result = try decodeBasicPacket(ctx: ctx, buffer: &buffer, capabilities: capabilities, forwarding: false)
+                session.connectionState = .statement(.rows(columns: columns))
+                return result
+            }
+            session.connectionState = .statement(.rows(columns: columns))
         case .rows(let columns):
             if buffer.peekInteger(as: Byte.self, skipping: 4) == 0xFE {
                 session.connectionState = .none

Sources/MySQL/Protocol/MySQLHandshakeV10.swift

@@ -71,11 +71,17 @@ struct MySQLHandshakeV10 {
             let reserved = try bytes.requireBytes(length: 10, source: .capture())
             assert(reserved == [0, 0, 0, 0, 0, 0, 0, 0, 0, 0])
 
-            if capabilities.get(CLIENT_SECURE_CONNECTION), authPluginDataLength > 0 {
-                let len = max(13, authPluginDataLength - 8)
-                let authPluginDataPart2 = try bytes.requireBytes(length: numericCast(len), source: .capture())
-
-                self.authPluginData = Data(authPluginDataPart1 + authPluginDataPart2)
+            if capabilities.get(CLIENT_SECURE_CONNECTION) {
+                if capabilities.get(CLIENT_PLUGIN_AUTH) {
+                    let len = max(13, authPluginDataLength - 8)
+                    let authPluginDataPart2 = try bytes.requireBytes(length: numericCast(len), source: .capture())
+                    self.authPluginData = Data(authPluginDataPart1 + authPluginDataPart2)
+                } else {
+                    let authPluginDataPart2 = try bytes.requireBytes(length: 12, source: .capture())
+                    self.authPluginData = Data(authPluginDataPart1 + authPluginDataPart2)
+                    let filler: Byte = try bytes.requireInteger(source: .capture())
+                    assert(filler == 0x00)
+                }
             } else {
                 self.authPluginData = Data(authPluginDataPart1)
             }