Use sync pipeline operations (#152)

Author: fabianfett

Package.swift

@@ -13,11 +13,11 @@ let package = Package(
         .library(name: "PostgresNIO", targets: ["PostgresNIO"]),
     ],
     dependencies: [
+        .package(url: "https://github.com/apple/swift-nio.git", from: "2.28.0"),
+        .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.12.0"),
         .package(url: "https://github.com/apple/swift-crypto.git", from: "1.0.0"),
-        .package(url: "https://github.com/apple/swift-nio.git", from: "2.0.0"),
-        .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.0.0"),
         .package(url: "https://github.com/apple/swift-metrics.git", from: "2.0.0"),
-        .package(url: "https://github.com/apple/swift-log.git", from: "1.0.0"),
+        .package(url: "https://github.com/apple/swift-log.git", from: "1.4.0"),
     ],
     targets: [
         .target(name: "PostgresNIO", dependencies: [

Sources/PostgresNIO/New/PSQLChannelHandler.swift

@@ -20,18 +20,18 @@ final class PSQLChannelHandler: ChannelDuplexHandler {
     }
     private var currentQuery: PSQLRows?
     private let authentificationConfiguration: PSQLConnection.Configuration.Authentication?
-    private let enableSSLCallback: ((Channel) -> EventLoopFuture<Void>)?
+    private let configureSSLCallback: ((Channel) throws -> Void)?
 
     /// this delegate should only be accessed on the connections `EventLoop`
     weak var notificationDelegate: PSQLChannelHandlerNotificationDelegate?
 
     init(authentification: PSQLConnection.Configuration.Authentication?,
          logger: Logger,
-         enableSSLCallback: ((Channel) -> EventLoopFuture<Void>)? = nil)
+         configureSSLCallback: ((Channel) throws -> Void)?)
     {
         self.state = ConnectionStateMachine()
         self.authentificationConfiguration = authentification
-        self.enableSSLCallback = enableSSLCallback
+        self.configureSSLCallback = configureSSLCallback
         self.logger = logger
     }
 
@@ -40,11 +40,11 @@ final class PSQLChannelHandler: ChannelDuplexHandler {
     init(authentification: PSQLConnection.Configuration.Authentication?,
          state: ConnectionStateMachine = .init(.initialized),
          logger: Logger = .psqlNoOpLogger,
-         enableSSLCallback: ((Channel) -> EventLoopFuture<Void>)? = nil)
+         configureSSLCallback: ((Channel) throws -> Void)?)
     {
         self.state = state
         self.authentificationConfiguration = authentification
-        self.enableSSLCallback = enableSSLCallback
+        self.configureSSLCallback = configureSSLCallback
         self.logger = logger
     }
     #endif
@@ -302,23 +302,21 @@ final class PSQLChannelHandler: ChannelDuplexHandler {
     // MARK: - Private Methods -
 
     private func connected(context: ChannelHandlerContext) {
-        let action = self.state.connected(requireTLS: self.enableSSLCallback != nil)
+        let action = self.state.connected(requireTLS: self.configureSSLCallback != nil)
 
         self.run(action, with: context)
     }
 
     private func establishSSLConnection(context: ChannelHandlerContext) {
         // This method must only be called, if we signalized the StateMachine before that we are
         // able to setup a SSL connection.
-        self.enableSSLCallback!(context.channel).whenComplete { result in
-            switch result {
-            case .success:
-                let action = self.state.sslHandlerAdded()
-                self.run(action, with: context)
-            case .failure(let error):
-                let action = self.state.errorHappened(.failedToAddSSLHandler(underlying: error))
-                self.run(action, with: context)
-            }
+        do {
+            try self.configureSSLCallback!(context.channel)
+            let action = self.state.sslHandlerAdded()
+            self.run(action, with: context)
+        } catch {
+            let action = self.state.errorHappened(.failedToAddSSLHandler(underlying: error))
+            self.run(action, with: context)
         }
     }
 

Sources/PostgresNIO/New/PSQLConnection.swift

@@ -215,17 +215,16 @@ final class PSQLConnection {
                     .channelInitializer { channel in
                         let decoder = ByteToMessageHandler(PSQLBackendMessage.Decoder())
 
-                        var enableSSLCallback: ((Channel) -> EventLoopFuture<Void>)? = nil
+                        var configureSSLCallback: ((Channel) throws -> ())? = nil
                         if let tlsConfiguration = configuration.tlsConfiguration {
-                            enableSSLCallback = { channel in
-                                channel.eventLoop.submit {
-                                    let sslContext = try NIOSSLContext(configuration: tlsConfiguration)
-                                    return try NIOSSLClientHandler(
-                                        context: sslContext,
-                                        serverHostname: configuration.sslServerHostname)
-                                }.flatMap { sslHandler in
-                                    channel.pipeline.addHandler(sslHandler, position: .before(decoder))
-                                }
+                            configureSSLCallback = { channel in
+                                channel.eventLoop.assertInEventLoop()
+                                
+                                let sslContext = try NIOSSLContext(configuration: tlsConfiguration)
+                                let sslHandler = try NIOSSLClientHandler(
+                                    context: sslContext,
+                                    serverHostname: configuration.sslServerHostname)
+                                try channel.pipeline.syncOperations.addHandler(sslHandler, position: .before(decoder))
                             }
                         }
 
@@ -235,7 +234,7 @@ final class PSQLConnection {
                             PSQLChannelHandler(
                                 authentification: configuration.authentication,
                                 logger: logger,
-                                enableSSLCallback: enableSSLCallback),
+                                configureSSLCallback: configureSSLCallback),
                             PSQLEventsHandler(logger: logger)
                         ])
                     }

Tests/PostgresNIOTests/New/PSQLChannelHandlerTests.swift

@@ -1,5 +1,6 @@
 import XCTest
 import NIO
+import NIOTLS
 @testable import PostgresNIO
 
 class PSQLChannelHandlerTests: XCTestCase {
@@ -8,7 +9,7 @@ class PSQLChannelHandlerTests: XCTestCase {
 
     func testHandlerAddedWithoutSSL() {
         let config = self.testConnectionConfiguration()
-        let handler = PSQLChannelHandler(authentification: config.authentication)
+        let handler = PSQLChannelHandler(authentification: config.authentication, configureSSLCallback: nil)
         let embedded = EmbeddedChannel(handler: handler)
         defer { XCTAssertNoThrow(try embedded.finish()) }
 
@@ -35,7 +36,6 @@ class PSQLChannelHandlerTests: XCTestCase {
         var addSSLCallbackIsHit = false
         let handler = PSQLChannelHandler(authentification: config.authentication) { channel in
             addSSLCallbackIsHit = true
-            return channel.eventLoop.makeSucceededFuture(())
         }
         let embedded = EmbeddedChannel(handler: handler)
 
@@ -48,14 +48,24 @@ class PSQLChannelHandlerTests: XCTestCase {
 
         XCTAssertEqual(request.code, 80877103)
 
-        // first we need to add an encoder, because NIOSSLHandler can only
-        // operate on ByteBuffer
-        let future = embedded.pipeline.addHandlers(MessageToByteHandler(PSQLFrontendMessage.Encoder.forTests), position: .first)
-        XCTAssertNoThrow(try future.wait())
         XCTAssertNoThrow(try embedded.writeInbound(PSQLBackendMessage.sslSupported))
 
         // a NIOSSLHandler has been added, after it SSL had been negotiated
         XCTAssertTrue(addSSLCallbackIsHit)
+        
+        // signal that the ssl connection has been established
+        embedded.pipeline.fireUserInboundEventTriggered(TLSUserEvent.handshakeCompleted(negotiatedProtocol: ""))
+        
+        // startup message should be issued
+        var maybeStartupMessage: PSQLFrontendMessage?
+        XCTAssertNoThrow(maybeStartupMessage = try embedded.readOutbound(as: PSQLFrontendMessage.self))
+        guard case .startup(let startupMessage) = maybeStartupMessage else {
+            return XCTFail("Unexpected message")
+        }
+        
+        XCTAssertEqual(startupMessage.parameters.user, config.authentication?.username)
+        XCTAssertEqual(startupMessage.parameters.database, config.authentication?.database)
+        XCTAssertEqual(startupMessage.parameters.replication, .false)
     }
 
     func testSSLUnsupportedClosesConnection() {
@@ -64,7 +74,7 @@ class PSQLChannelHandlerTests: XCTestCase {
 
         let handler = PSQLChannelHandler(authentification: config.authentication) { channel in
             XCTFail("This callback should never be exectuded")
-            return channel.eventLoop.makeFailedFuture(PSQLError.sslUnsupported)
+            throw PSQLError.sslUnsupported
         }
         let embedded = EmbeddedChannel(handler: handler)
         let eventHandler = TestEventHandler()
@@ -94,7 +104,7 @@ class PSQLChannelHandlerTests: XCTestCase {
             database: config.authentication?.database
         )
         let state = ConnectionStateMachine(.waitingToStartAuthentication)
-        let handler = PSQLChannelHandler(authentification: config.authentication, state: state)
+        let handler = PSQLChannelHandler(authentification: config.authentication, state: state, configureSSLCallback: nil)
         let embedded = EmbeddedChannel(handler: handler)
 
         embedded.triggerUserOutboundEvent(PSQLOutgoingEvent.authenticate(authContext), promise: nil)
@@ -119,7 +129,7 @@ class PSQLChannelHandlerTests: XCTestCase {
             database: config.authentication?.database
         )
         let state = ConnectionStateMachine(.waitingToStartAuthentication)
-        let handler = PSQLChannelHandler(authentification: config.authentication, state: state)
+        let handler = PSQLChannelHandler(authentification: config.authentication, state: state, configureSSLCallback: nil)
         let embedded = EmbeddedChannel(handler: handler)
 
         embedded.triggerUserOutboundEvent(PSQLOutgoingEvent.authenticate(authContext), promise: nil)