nio 2 updates

Author: tanner0101

Package.swift

@@ -8,7 +8,7 @@ let package = Package(
         .executable(name: "NIOPostgresBenchmark", targets: ["NIOPostgresBenchmark"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/apple/swift-nio.git", from: "1.0.0"),
+        .package(url: "https://github.com/apple/swift-nio.git", .branch("master")),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "1.0.0"),
     ],
     targets: [

Sources/NIOPostgres/Connection/PostgresConnection+Connect.swift

@@ -8,7 +8,7 @@ extension PostgresConnection {
         return bootstrap.connect(to: socketAddress).then { channel in
             let handler = PostgresConnection.InboundHandler(channel)
             return channel.pipeline.addHandlers([
-                PostgresMessage.InboundHandler(),
+                ByteToMessageHandler(PostgresMessage.InboundHandler()),
                 PostgresMessage.OutboundHandler(),
                 handler,
             ], first: false).map {

Sources/NIOPostgres/Connection/PostgresConnection+InboundHandler.swift

@@ -21,7 +21,7 @@ extension PostgresConnection {
 
         func send(_ messages: [PostgresMessage], _ callback: @escaping (PostgresMessage) throws -> Bool) -> EventLoopFuture<Void> {
             // print("PostgresConnection.ChannelHandler.send(\(messages))")
-            let promise: EventLoopPromise<Void> = channel.eventLoop.newPromise()
+            let promise: EventLoopPromise<Void> = channel.eventLoop.makePromise()
             waiters.append(Request(promise: promise, callback: callback))
             messages.forEach { channel.write(wrapOutboundOut($0)).cascadeFailure(promise: promise) }
             channel.flush()

Sources/NIOPostgres/Connection/PostgresConnection+Query.swift

@@ -11,7 +11,7 @@ extension PostgresConnection {
         do {
             data = try binds.serialize(allocator: self.handler.channel.allocator)
         } catch {
-            return self.eventLoop.newFailedFuture(error: error)
+            return self.eventLoop.makeFailedFuture(error: error)
         }
         print("[NIOPostgres] \(string) \(data)")
         let parse = PostgresMessage.Parse(

Sources/NIOPostgres/Connection/PostgresConnection.swift

@@ -49,7 +49,7 @@ extension PostgresConnection {
                 let handler = try! OpenSSLClientHandler(context: sslContext)
                 return self.handler.channel.pipeline.add(handler: handler, first: true).map { true }
             case .unsupported:
-                return self.eventLoop.newSucceededFuture(result: false)
+                return self.eventLoop.makeSucceededFuture(result: false)
             }
         }
     }

Sources/NIOPostgres/Message/PostgresMessage+InboundHandler.swift

@@ -20,39 +20,37 @@ extension PostgresMessage {
         /// See `ByteToMessageDecoder`.
         func decode(ctx: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState {
             // special check for SSL response
+            var sslBuffer = buffer
             if
                 !self.hasRequestedAuthentication,
-                let sslResponse = buffer.getInteger(
-                    at: buffer.readerIndex, as: UInt8.self
-                ).flatMap(PostgresMessage.SSLResponse.init)
+                let sslResponse = sslBuffer.readInteger(as: UInt8.self)
+                    .flatMap(PostgresMessage.SSLResponse.init)
             {
-                buffer.moveReaderIndex(forwardBy: 1)
+                buffer = sslBuffer
                 ctx.fireChannelRead(wrapInboundOut(.sslResponse(sslResponse)))
                 return .continue
             }
 
+            var peekBuffer = buffer
             // peek at the message identifier
             // the message identifier is always the first byte of a message
-            guard let messageIdentifier = buffer.getInteger(at: buffer.readerIndex, as: UInt8.self).map(PostgresMessage.Identifier.init) else {
+            guard let messageIdentifier = peekBuffer.readInteger(as: UInt8.self).map(PostgresMessage.Identifier.init) else {
                 return .needMoreData
             }
-            // print("PostgresMessage.ChannelDecoder.decode(\(messageIdentifier))")
-            
-            #warning("check for TLS support")
 
             // peek at the message size
             // the message size is always a 4 byte integer appearing immediately after the message identifier
-            guard let messageSize = buffer.getInteger(at: buffer.readerIndex + 1, as: Int32.self).flatMap(Int.init) else {
+            guard let messageSize = peekBuffer.readInteger(as: Int32.self).flatMap(Int.init) else {
                 return .needMoreData
             }
 
             // ensure message is large enough (skipping message type) or reject
-            guard buffer.readableBytes - 1 >= messageSize else {
+            guard peekBuffer.readableBytes >= messageSize - 4 else {
                 return .needMoreData
             }
 
-            // skip message identifier and message size
-            buffer.moveReaderIndex(forwardBy: 1 + 4)
+            // there is sufficient data, use this buffer
+            buffer = peekBuffer
 
             let message: PostgresMessage
             switch messageIdentifier {

Tests/NIOPostgresTests/NIOPostgresTests.swift

@@ -241,7 +241,7 @@ final class NIOPostgresTests: XCTestCase {
         defer { try! elg.syncShutdownGracefully() }
 
         let conn = try PostgresConnection.connect(
-            to: SocketAddress.newAddressResolving(host: "elmer.db.elephantsql.com", port: 5432),
+            to: SocketAddress.makeAddressResolvingHost("elmer.db.elephantsql.com", port: 5432),
             on: elg.next()
         ).wait()
         let upgraded = try conn.requestTLS(using: .forClient(certificateVerification: .none)).wait()

Tests/NIOPostgresTests/Utilities.swift

@@ -16,7 +16,7 @@ extension PostgresConnection {
                     .map { conn }
             }
         } catch {
-            return eventLoop.newFailedFuture(error: error)
+            return eventLoop.makeFailedFuture(error: error)
         }
     }
 }