[HTTP2] Forward HTTP2 events to StateMachine (#466)

Author: dnadoba

Sources/AsyncHTTPClient/ConnectionPool/HTTP2/HTTP2Connection.swift

@@ -37,7 +37,7 @@ final class HTTP2Connection {
 
     enum State {
         case initialized
-        case starting(EventLoopPromise<Void>)
+        case starting(EventLoopPromise<Int>)
         case active(maxStreams: Int)
         case closing
         case closed
@@ -117,9 +117,9 @@ final class HTTP2Connection {
         delegate: HTTP2ConnectionDelegate,
         configuration: HTTPClient.Configuration,
         logger: Logger
-    ) -> EventLoopFuture<HTTP2Connection> {
+    ) -> EventLoopFuture<(HTTP2Connection, Int)> {
         let connection = HTTP2Connection(channel: channel, connectionID: connectionID, delegate: delegate, logger: logger)
-        return connection.start().map { _ in connection }
+        return connection.start().map { maxStreams in (connection, maxStreams) }
     }
 
     func executeRequest(_ request: HTTPExecutableRequest) {
@@ -154,10 +154,10 @@ final class HTTP2Connection {
         return promise.futureResult
     }
 
-    private func start() -> EventLoopFuture<Void> {
+    private func start() -> EventLoopFuture<Int> {
         self.channel.eventLoop.assertInEventLoop()
 
-        let readyToAcceptConnectionsPromise = self.channel.eventLoop.makePromise(of: Void.self)
+        let readyToAcceptConnectionsPromise = self.channel.eventLoop.makePromise(of: Int.self)
 
         self.state = .starting(readyToAcceptConnectionsPromise)
         self.channel.closeFuture.whenComplete { _ in
@@ -266,7 +266,7 @@ extension HTTP2Connection: HTTP2IdleHandlerDelegate {
 
         case .starting(let promise):
             self.state = .active(maxStreams: maxStreams)
-            promise.succeed(())
+            promise.succeed(maxStreams)
 
         case .active:
             self.state = .active(maxStreams: maxStreams)

Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool+Factory.swift

@@ -86,8 +86,8 @@ extension HTTPConnectionPool.ConnectionFactory {
                     logger: logger
                 ).whenComplete { result in
                     switch result {
-                    case .success(let connection):
-                        requester.http2ConnectionCreated(connection, maximumStreams: 0)
+                    case .success((let connection, let maximumStreams)):
+                        requester.http2ConnectionCreated(connection, maximumStreams: maximumStreams)
                     case .failure(let error):
                         requester.failedToCreateHTTPConnection(connectionID, error: error)
                     }

Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift

@@ -448,18 +448,14 @@ extension HTTPConnectionPool: HTTPConnectionRequester {
     }
 
     func http2ConnectionCreated(_ connection: HTTP2Connection, maximumStreams: Int) {
-        preconditionFailure("Did not expect http/2 connections right now.")
-//        let action = self.stateLock.withLock { () -> StateMachine.Action in
-//            if let settings = connection.settings {
-//                return self._state.newHTTP2ConnectionCreated(.http2(connection), settings: settings)
-//            } else {
-//                // immidiate connection closure before we can register with state machine
-//                // is the only reason we don't have settings
-//                struct ImmidiateConnectionClose: Error {}
-//                return self._state.failedToCreateNewConnection(ImmidiateConnectionClose(), connectionID: connection.id)
-//            }
-//        }
-//        self.run(action: action)
+        self.logger.trace("successfully created connection", metadata: [
+            "ahc-connection-id": "\(connection.id)",
+            "ahc-http-version": "http/2",
+            "ahc-max-streams": "\(maximumStreams)",
+        ])
+        self.modifyStateAndRunActions {
+            $0.newHTTP2ConnectionCreated(.http2(connection), maxConcurrentStreams: maximumStreams)
+        }
     }
 
     func failedToCreateHTTPConnection(_ connectionID: HTTPConnectionPool.Connection.ID, error: Error) {
@@ -497,27 +493,44 @@ extension HTTPConnectionPool: HTTP1ConnectionDelegate {
 
 extension HTTPConnectionPool: HTTP2ConnectionDelegate {
     func http2Connection(_ connection: HTTP2Connection, newMaxStreamSetting: Int) {
-        // ignore for now
+        self.logger.debug("new max stream setting", metadata: [
+            "ahc-connection-id": "\(connection.id)",
+            "ahc-http-version": "http/2",
+            "ahc-max-streams": "\(newMaxStreamSetting)",
+        ])
+        self.modifyStateAndRunActions {
+            $0.newHTTP2MaxConcurrentStreamsReceived(connection.id, newMaxStreams: newMaxStreamSetting)
+        }
     }
 
-    func http2ConnectionGoAwayReceived(_: HTTP2Connection) {
-        // ignore for now
+    func http2ConnectionGoAwayReceived(_ connection: HTTP2Connection) {
+        self.logger.debug("connection go away received", metadata: [
+            "ahc-connection-id": "\(connection.id)",
+            "ahc-http-version": "http/2",
+        ])
+        self.modifyStateAndRunActions {
+            $0.http2ConnectionGoAwayReceived(connection.id)
+        }
     }
 
-    func http2ConnectionClosed(_: HTTP2Connection) {
-        // ignore for now
-//        let action = self.stateLock.withLock {
-//            self._state.connectionClosed(connection.id)
-//        }
-//        self.run(action: action)
+    func http2ConnectionClosed(_ connection: HTTP2Connection) {
+        self.logger.debug("connection closed", metadata: [
+            "ahc-connection-id": "\(connection.id)",
+            "ahc-http-version": "http/2",
+        ])
+        self.modifyStateAndRunActions {
+            $0.http2ConnectionClosed(connection.id)
+        }
     }
 
     func http2ConnectionStreamClosed(_ connection: HTTP2Connection, availableStreams: Int) {
-        // ignore for now
-//        let action = self.stateLock.withLock {
-//            self._state.http2ConnectionStreamClosed(connection.id, availableStreams: availableStreams)
-//        }
-//        self.run(action: action)
+        self.logger.trace("stream closed", metadata: [
+            "ahc-connection-id": "\(connection.id)",
+            "ahc-http-version": "http/2",
+        ])
+        self.modifyStateAndRunActions {
+            $0.http2ConnectionStreamClosed(connection.id)
+        }
     }
 }