nio

Author: tanner0101

Diff for: Package.swift

@@ -7,26 +7,23 @@ let package = Package(
         .library(name: "PostgreSQL", targets: ["PostgreSQL"]),
     ],
     dependencies: [
-        // ⏱ Promises and reactive-streams in Swift built for high-performance and scalability.
-        .package(url: "https://github.com/vapor/async.git", from: "1.0.0-rc"),
-        
         // 🌎 Utility package containing tools for byte manipulation, Codable, OS APIs, and debugging.
-        .package(url: "https://github.com/vapor/core.git", from: "3.0.0-rc"),
+        .package(url: "https://github.com/vapor/core.git", .branch("nio")),
 
         // 🔑 Hashing (BCrypt, SHA, HMAC, etc), encryption, and randomness.
-        .package(url: "https://github.com/vapor/crypto.git", from: "3.0.0-rc"),
+        .package(url: "https://github.com/vapor/crypto.git", .branch("nio")),
 
         // 🗄 Core services for creating database integrations.
-        .package(url: "https://github.com/vapor/database-kit.git", from: "1.0.0-rc"),
+        .package(url: "https://github.com/vapor/database-kit.git", .branch("nio")),
 
         // 📦 Dependency injection / inversion of control framework.
-        .package(url: "https://github.com/vapor/service.git", from: "1.0.0-rc"),
-
-        // 🔌 Non-blocking TCP socket layer, with event-driven server and client.
-        .package(url: "https://github.com/vapor/sockets.git", from: "3.0.0-rc"),
+        .package(url: "https://github.com/vapor/service.git", .branch("nio")),
+        
+        // Event-driven network application framework for high performance protocol servers & clients, non-blocking.
+        .package(url: "https://github.com/apple/swift-nio.git", from: "1.0.0"),
     ],
     targets: [
-        .target(name: "PostgreSQL", dependencies: ["Async", "Bits", "Crypto", "DatabaseKit", "Service", "TCP"]),
+        .target(name: "PostgreSQL", dependencies: ["Async", "Bits", "Crypto", "DatabaseKit", "NIO", "Service"]),
         .testTarget(name: "PostgreSQLTests", dependencies: ["PostgreSQL"]),
     ]
 )

Diff for: Sources/PostgreSQL/Connection/PostgreSQLConnection+TCP.swift

@@ -1,18 +1,27 @@
 import Async
-import TCP
+import NIO
 
 extension PostgreSQLConnection {
     /// Connects to a Redis server using a TCP socket.
     public static func connect(
         hostname: String = "localhost",
         port: UInt16 = 5432,
         on worker: Worker,
-        onError: @escaping TCPSocketSink.ErrorHandler
+        onError: @escaping (Error) -> ()
     ) throws -> PostgreSQLConnection {
-        let socket = try TCPSocket(isNonBlocking: true)
-        let client = try TCPClient(socket: socket)
-        try client.connect(hostname: hostname, port: port)
-        let stream = socket.stream(on: worker, onError: onError)
+        let handler = HTTPClientHandler()
+        let bootstrap = ClientBootstrap(group: group)
+            // Enable SO_REUSEADDR.
+            .channelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
+            .channelInitializer { channel in
+                return channel.pipeline.addHTTPClientHandlers().then {
+                    channel.pipeline.add(handler: handler)
+                }
+        }
+
+        return bootstrap.connect(host: hostname, port: port).map(to: HTTPClient.self) { _ in
+            return .init(handler: handler, bootstrap: bootstrap)
+        }
         return PostgreSQLConnection(stream: stream, on: worker)
     }
 }

Diff for: Sources/PostgreSQL/Connection/PostgreSQLConnection.swift

@@ -1,5 +1,6 @@
 import Async
 import Crypto
+import NIO
 
 /// A PostgreSQL frontend client.
 public final class PostgreSQLConnection {
@@ -11,6 +12,8 @@ public final class PostgreSQLConnection {
 
     /// Creates a new Redis client on the provided data source and sink.
     init<Stream>(stream: Stream, on worker: Worker) where Stream: ByteStream {
+        
+
         let queueStream = QueueStream<PostgreSQLMessage, PostgreSQLMessage>()
 
         let serializerStream = PostgreSQLMessageSerializer().stream(on: worker)

Diff for: Sources/PostgreSQL/PostgreSQLMessageDecoder.swift

@@ -0,0 +1,8 @@
+//
+//  PostgreSQLMessageDecoder.swift
+//  PostgreSQL
+//
+//  Created by Tanner on 3/2/18.
+//
+
+import Foundation

Diff for: Sources/PostgreSQL/QueueHandler.swift

@@ -0,0 +1,66 @@
+import Async
+import NIO
+
+struct InputContext<In> {
+    var promise: Promise<Void>
+    var onInput: (In) throws -> Bool
+}
+
+final class QueueHandler<In, Out>: ChannelInboundHandler {
+    typealias InboundIn = In
+    typealias OutboundOut = Out
+
+    private var inputQueue: [InputContext<InboundIn>]
+    private var outputQueue: [OutboundOut]
+    private let eventLoop: EventLoop
+    private var waitingCtx: ChannelHandlerContext?
+
+    init(on worker: Worker) {
+        self.inputQueue = []
+        self.outputQueue = []
+    }
+
+    func enqueue(_ output: [OutboundOut], onInput: @escaping (InboundIn) throws -> Bool) -> Future<Void> {
+        outputQueue.insert(contentsOf: output.reversed(), at: 0)
+        let promise = eventLoop.newPromise(Void.self)
+        let context = InputContext<InboundIn>(promise: promise, onInput: onInput)
+        inputQueue.insert(context, at: 0)
+        if let ctx = waitingCtx {
+            sendOutput(ctx: ctx)
+        }
+        return promise.futureResult
+    }
+
+    func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
+        let input = unwrapInboundIn(data)
+        guard let current = inputQueue.last else {
+            assert(false, "Empty input queue.")
+            return
+        }
+        do {
+            if try current.onInput(input) {
+                current.promise.succeed()
+                assert(inputQueue.popLast() != nil)
+            }
+        } catch {
+            current.promise.fail(error: error)
+            assert(inputQueue.popLast() != nil)
+        }
+    }
+
+    func channelActive(ctx: ChannelHandlerContext) {
+        sendOutput(ctx: ctx)
+    }
+
+    func sendOutput(ctx: ChannelHandlerContext) {
+        if let next = outputQueue.popLast() {
+            ctx.write(wrapOutboundOut(next)).do {
+                self.sendOutput(ctx: ctx)
+            }.catch { error in
+                fatalError("\(error)")
+            }
+        } else {
+            waitingCtx = ctx
+        }
+    }
+}