nio encoder/decoder

Author: tanner0101

Sources/PostgreSQL/Connection/PostgreSQLConnection+TCP.swift

@@ -5,23 +5,39 @@ extension PostgreSQLConnection {
     /// Connects to a Redis server using a TCP socket.
     public static func connect(
         hostname: String = "localhost",
-        port: UInt16 = 5432,
+        port: Int = 5432,
         on worker: Worker,
         onError: @escaping (Error) -> ()
-    ) throws -> PostgreSQLConnection {
-        let handler = HTTPClientHandler()
-        let bootstrap = ClientBootstrap(group: group)
+    ) throws -> Future<PostgreSQLConnection> {
+        let handler = QueueHandler<PostgreSQLMessage, PostgreSQLMessage>(on: worker)
+        let bootstrap = ClientBootstrap(group: worker.eventLoop)
             // Enable SO_REUSEADDR.
             .channelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
             .channelInitializer { channel in
-                return channel.pipeline.addHTTPClientHandlers().then {
+                return channel.pipeline.addPostgreSQLClientHandlers().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 bootstrap.connect(host: hostname, port: port).map(to: PostgreSQLConnection.self) { channel in
+            return .init(queue: handler, channel: channel)
         }
-        return PostgreSQLConnection(stream: stream, on: worker)
+    }
+}
+
+extension ChannelPipeline {
+    func addPostgreSQLClientHandlers(first: Bool = false) -> EventLoopFuture<Void> {
+        return addHandlers(PostgreSQLMessageEncoder(), PostgreSQLMessageDecoder(), first: first)
+    }
+
+    /// Adds the provided channel handlers to the pipeline in the order given, taking account
+    /// of the behaviour of `ChannelHandler.add(first:)`.
+    private func addHandlers(_ handlers: ChannelHandler..., first: Bool) -> EventLoopFuture<Void> {
+        var handlers = handlers
+        if first {
+            handlers = handlers.reversed()
+        }
+
+        return EventLoopFuture<Void>.andAll(handlers.map { add(handler: $0) }, eventLoop: eventLoop)
     }
 }

Sources/PostgreSQL/Connection/PostgreSQLConnection.swift

@@ -5,32 +5,24 @@ import NIO
 /// A PostgreSQL frontend client.
 public final class PostgreSQLConnection {
     /// Handles enqueued redis commands and responses.
-    private let queueStream: QueueStream<PostgreSQLMessage, PostgreSQLMessage>
+    private let queue: QueueHandler<PostgreSQLMessage, PostgreSQLMessage>
+
+    /// The channel
+    private let channel: Channel
 
     /// If non-nil, will log queries.
     public var logger: PostgreSQLLogger?
 
     /// 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)
-        let parserStream = PostgreSQLMessageParser().stream(on: worker)
-
-        stream.stream(to: parserStream)
-            .stream(to: queueStream)
-            .stream(to: serializerStream)
-            .output(to: stream)
-
-        self.queueStream = queueStream
+    init(queue: QueueHandler<PostgreSQLMessage, PostgreSQLMessage>, channel: Channel) {
+        self.queue = queue
+        self.channel = channel
     }
 
     /// Sends `PostgreSQLMessage` to the server.
     func send(_ messages: [PostgreSQLMessage], onResponse: @escaping (PostgreSQLMessage) throws -> ()) -> Future<Void> {
         var error: Error?
-        return queueStream.enqueue(messages) { message in
+        return queue.enqueue(messages) { message in
             switch message {
             case .readyForQuery:
                 if let e = error { throw e }
@@ -60,7 +52,7 @@ public final class PostgreSQLConnection {
             "database": database ?? username
         ])
         var authRequest: PostgreSQLAuthenticationRequest?
-        return queueStream.enqueue([.startupMessage(startup)]) { message in
+        return queue.enqueue([.startupMessage(startup)]) { message in
             switch message {
             case .authenticationRequest(let a):
                 authRequest = a
@@ -114,7 +106,7 @@ public final class PostgreSQLConnection {
                 input = [.password(passwordMessage)]
             }
 
-            return self.queueStream.enqueue(input) { message in
+            return self.queue.enqueue(input) { message in
                 switch message {
                 case .error(let error): throw error
                 case .readyForQuery: return true
@@ -128,6 +120,6 @@ public final class PostgreSQLConnection {
 
     /// Closes this client.
     public func close() {
-        queueStream.close()
+        channel.close(promise: nil)
     }
 }

Sources/PostgreSQL/Database/PostgreSQLDatabase.swift

@@ -15,18 +15,18 @@ public final class PostgreSQLDatabase: Database {
 
     /// See `Database.makeConnection()`
     public func makeConnection(on worker: Worker) -> Future<PostgreSQLConnection> {
-        do {
-            let client = try PostgreSQLConnection.connect(hostname: config.hostname, port: config.port, on: worker) { _, error in
+        let config = self.config
+        return Future.flatMap(on: worker) {
+            return try PostgreSQLConnection.connect(hostname: config.hostname, port: config.port, on: worker) { error in
                 print("[PostgreSQL] \(error)")
+            }.flatMap(to: PostgreSQLConnection.self) { client in
+                client.logger = self.logger
+                return client.authenticate(
+                    username: config.username,
+                    database: config.database,
+                    password: config.password
+                ).transform(to: client)
             }
-            client.logger = logger
-            return client.authenticate(
-                username: config.username, 
-                database: config.database,
-                password: config.password
-            ).transform(to: client)
-        } catch {
-            return Future(error: error)
         }
     }
 }

Sources/PostgreSQL/Database/PostgreSQLDatabaseConfig.swift

@@ -9,7 +9,7 @@ public struct PostgreSQLDatabaseConfig {
     public let hostname: String
 
     /// Destination port.
-    public let port: UInt16
+    public let port: Int
 
     /// Username to authenticate.
     public let username: String
@@ -22,7 +22,7 @@ public struct PostgreSQLDatabaseConfig {
     public let password: String?
 
     /// Creates a new `PostgreSQLDatabaseConfig`.
-    public init(hostname: String, port: UInt16, username: String, database: String? = nil, password: String? = nil) {
+    public init(hostname: String, port: Int, username: String, database: String? = nil, password: String? = nil) {
         self.hostname = hostname
         self.port = port
         self.username = username