Code review part 3

Author: fabianfett

Diff for: Sources/AsyncHTTPClient/ConnectionPool/HTTP2/HTTP2ClientRequestHandler.swift

@@ -17,7 +17,7 @@ import NIO
 import NIOHTTP1
 import NIOHTTP2
 
-class HTTP2ClientRequestHandler: ChannelDuplexHandler {
+final class HTTP2ClientRequestHandler: ChannelDuplexHandler {
     typealias OutboundIn = HTTPExecutableRequest
     typealias OutboundOut = HTTPClientRequestPart
     typealias InboundIn = HTTPClientResponsePart
@@ -35,10 +35,8 @@ class HTTP2ClientRequestHandler: ChannelDuplexHandler {
 
     private var request: HTTPExecutableRequest? {
         didSet {
-            if let newRequest = self.request {
-                if let idleReadTimeout = newRequest.idleReadTimeout {
-                    self.idleReadTimeoutStateMachine = .init(timeAmount: idleReadTimeout)
-                }
+            if let newRequest = self.request, let idleReadTimeout = newRequest.idleReadTimeout {
+                self.idleReadTimeoutStateMachine = .init(timeAmount: idleReadTimeout)
             } else {
                 self.idleReadTimeoutStateMachine = nil
             }
@@ -88,7 +86,7 @@ class HTTP2ClientRequestHandler: ChannelDuplexHandler {
             self.runTimeoutAction(timeoutAction, context: context)
         }
 
-        let action = self.state.channelRead(self.unwrapInboundIn(data))
+        let action = self.state.channelRead(httpPart)
         self.run(action, context: context)
     }
 

Diff for: Sources/AsyncHTTPClient/ConnectionPool/HTTP2/HTTP2Connection.swift

@@ -26,7 +26,7 @@ struct HTTP2PushNotSupportedError: Error {}
 
 struct HTTP2ReceivedGoAwayBeforeSettingsError: Error {}
 
-class HTTP2Connection {
+final class HTTP2Connection {
     let channel: Channel
     let multiplexer: HTTP2StreamMultiplexer
     let logger: Logger
@@ -106,7 +106,6 @@ class HTTP2Connection {
             outboundBufferSizeHighWatermark: 8196,
             outboundBufferSizeLowWatermark: 4092,
             inboundStreamInitializer: { (channel) -> EventLoopFuture<Void> in
-
                 channel.eventLoop.makeFailedFuture(HTTP2PushNotSupportedError())
             }
         )
@@ -116,7 +115,7 @@ class HTTP2Connection {
 
     deinit {
         guard case .closed = self.state else {
-            preconditionFailure("")
+            preconditionFailure("Connection must be closed, before we can deinit it")
         }
     }
 
@@ -136,20 +135,25 @@ class HTTP2Connection {
 
         self.multiplexer.createStreamChannel(promise: createStreamChannelPromise) { channel -> EventLoopFuture<Void> in
             do {
+                // We only support http/2 over an https connection – using the Application-Layer
+                // Protocol Negotiation (ALPN). For this reason it is save to fix this to `.https`.
                 let translate = HTTP2FramePayloadToHTTP1ClientCodec(httpProtocol: .https)
                 let handler = HTTP2ClientRequestHandler(eventLoop: channel.eventLoop)
 
                 try channel.pipeline.syncOperations.addHandler(translate)
                 try channel.pipeline.syncOperations.addHandler(handler)
-                channel.write(request, promise: nil)
 
+                // We must add the new channel to the list of open channels BEFORE we write the
+                // request to it. In case of an error, we are sure that the channel was added
+                // before.
                 let box = ChannelBox(channel)
                 self.openStreams.insert(box)
                 self.channel.closeFuture.whenComplete { _ in
                     self.openStreams.remove(box)
                 }
 
-                return channel.eventLoop.makeSucceededFuture(Void())
+                channel.write(request, promise: nil)
+                return channel.eventLoop.makeSucceededVoidFuture()
             } catch {
                 return channel.eventLoop.makeFailedFuture(error)
             }
@@ -256,8 +260,13 @@ class HTTP2Connection {
 
         self.state = .closing
 
+        // inform all open streams, that the currently running request should be cancelled.
         self.openStreams.forEach { box in
             box.channel.triggerUserOutboundEvent(HTTPConnectionEvent.cancelRequest, promise: nil)
         }
+
+        // inform the idle connection handler, that connection should be closed, once all streams
+        // are closed.
+        self.channel.triggerUserOutboundEvent(HTTPConnectionEvent.closeConnection, promise: nil)
     }
 }

Diff for: Sources/AsyncHTTPClient/ConnectionPool/HTTP2/HTTP2IdleHandler.swift

@@ -16,7 +16,8 @@ import Logging
 import NIO
 import NIOHTTP2
 
-internal final class HTTP2IdleHandler: ChannelDuplexHandler {
+// This is a `ChannelDuplexHandler` since we need to intercept outgoing user events.
+final class HTTP2IdleHandler: ChannelDuplexHandler {
     typealias InboundIn = HTTP2Frame
     typealias InboundOut = HTTP2Frame
     typealias OutboundIn = HTTP2Frame
@@ -25,7 +26,7 @@ internal final class HTTP2IdleHandler: ChannelDuplexHandler {
     let logger: Logger
     let connection: HTTP2Connection
 
-    var state: StateMachine = .init()
+    private var state: StateMachine = .init()
 
     init(connection: HTTP2Connection, logger: Logger) {
         self.connection = connection
@@ -66,7 +67,7 @@ internal final class HTTP2IdleHandler: ChannelDuplexHandler {
         context.fireChannelRead(data)
     }
 
-    func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) {
+    func triggerUserOutboundEvent(context: ChannelHandlerContext, event: Any, promise: EventLoopPromise<Void>?) {
         switch event {
         case HTTPConnectionEvent.closeConnection:
             let action = self.state.closeEventReceived()
@@ -140,6 +141,9 @@ extension HTTP2IdleHandler {
                 preconditionFailure("Invalid state: \(self.state)")
 
             case .connected:
+                // a settings frame might have multiple entries for `maxConcurrentStreams`. We are
+                // only interested in the last value! If no `maxConcurrentStreams` is set, we assume
+                // the http/2 default of 100.
                 let maxStreams = settings.last(where: { $0.parameter == .maxConcurrentStreams })?.value ?? 100
                 self.state = .active(openStreams: 0, maxStreams: maxStreams)
                 return .notifyConnectionNewSettings(settings)
@@ -197,28 +201,41 @@ extension HTTP2IdleHandler {
         }
 
         mutating func streamCreated() -> Action {
-            guard case .active(var openStreams, let maxStreams) = self.state else {
-                preconditionFailure("Invalid state")
-            }
+            switch self.state {
+            case .active(var openStreams, let maxStreams):
+                openStreams += 1
+                self.state = .active(openStreams: openStreams, maxStreams: maxStreams)
+                return .nothing
 
-            openStreams += 1
-            assert(openStreams <= maxStreams)
+            case .closing(var openStreams, let maxStreams):
+                openStreams += 1
+                self.state = .active(openStreams: openStreams, maxStreams: maxStreams)
+                return .nothing
 
-            self.state = .active(openStreams: openStreams, maxStreams: maxStreams)
-            return .nothing
+            case .initialized, .connected, .closed:
+                preconditionFailure("Invalid state: \(self.state)")
+            }
         }
 
         mutating func streamClosed() -> Action {
-            guard case .active(var openStreams, let maxStreams) = self.state else {
-                preconditionFailure("Invalid state")
-                // TODO: What happens, if we received a go away?!??!
-            }
+            switch self.state {
+            case .active(var openStreams, let maxStreams):
+                openStreams -= 1
+                self.state = .active(openStreams: openStreams, maxStreams: maxStreams)
+                return .notifyConnectionStreamClosed(currentlyAvailable: maxStreams - openStreams)
 
-            openStreams -= 1
-            assert(openStreams >= 0)
+            case .closing(var openStreams, let maxStreams):
+                openStreams -= 1
+                if openStreams == 0 {
+                    self.state = .closed
+                    return .close
+                }
+                self.state = .closing(openStreams: openStreams, maxStreams: maxStreams)
+                return .nothing
 
-            self.state = .active(openStreams: openStreams, maxStreams: maxStreams)
-            return .notifyConnectionStreamClosed(currentlyAvailable: maxStreams - openStreams)
+            case .initialized, .connected, .closed:
+                preconditionFailure("Invalid state: \(self.state)")
+            }
         }
     }
 }