Repository: apple/swift-nio-http2
Pull request #36 (2018-12-29 23:49:12)

Initialize HTTP2Parser handler in the handlerAdded event hook.
By initializing the handler in the handlerAdded hook, HTTP2Parser supports
being added dynamically in a channelRead event, after the channel has sent
the channelActive event.

swift-nio-bot: Can one of the admins verify this patch?
swift-nio-bot: Can one of the admins verify this patch?
swift-nio-bot: Can one of the admins verify this patch?
sergiocampama: Sure, I can do that, but for illustrative purposes, let me explain the use case, because I don't follow the "writing to channels before they're active isn't a good call" comment, since I'm not writing to a channel/ It's a server and a handler modifies the pipeline during a `channelRead`.

The architecture I have is this initial pipeline configuration.

`channel.pipeline = [Http1_2Multiplexer(I)]`

`Http1_2Multiplexer` peeks into the first line of the first `channelRead`, and sees whether it's HTTP1 or HTTP2. If it's HTTP1, I modify the pipeline with `NIOHTTP1`'s `configureHTTPServerPipeline`, and after that I append my HTTP1 handler, becoming:

`channel.pipeline = [Http1_2Multiplexer(I), HTTPRequestDecoder(IO), HTTPResponseEncoder(IO), MYHTTP1Handler(I)]`

If the request is HTTP2 instead, it becomes:

`channel.pipeline = [Http1_2Multiplexer(I), HTTP2Parser(IO), HTTP2Multiplexer(IO), HTTP2ToHTTP1ClientCodec(IO), MYHTTP1Handler(I)]`

When I add `HTTP2Parser` to the pipeline, it seems like `channelActive` is not getting called, resulting in the `session` being `nil` and failing at runtime when the first `channelRead` is forwarded from `Http1_2Multiplexer` after modifying the pipeline.

So as I understand it, when the initial request comes, the channel becomes active, I then add HTTP2Parser, and it doesn't get the `channelActive` event. This is well before I write anything to the channel, which is why I'm not following the comment :)

Should I instead trigger a `fireChannelActive` event after I modify the pipeline to ensure that the new pipeline is active?
sergiocampama: Just checked, if I trigger a `fireChannelActive` in the first `channelRead` from `Http1_2Multiplexer` after I modify the pipeline and before trigger the `fireChannelRead`call, `HTTPParser2` works as expected.

Is this expected? It feels dirty having to trigger the `channelActive` event this way.
Lukasa: > because I don't follow the "writing to channels before they're active isn't a good call" comment, since I'm not writing to a channel/ 

You aren't, but `HTTP2Parser` *does* in the body of your `handlerAdded` method, as it writes out the HTTP/2 server preamble.

> When I add `HTTP2Parser` to the pipeline, it seems like `channelActive` is not getting called

This is expected behaviour, and you're right that right now `HTTP2Parser` does not tolerate it well.

> Should I instead trigger a `fireChannelActive` event after I modify the pipeline to ensure that the new pipeline is active?

No. What I'm suggesting is that you add the following function to `HTTP2Parser` in this PR:

```swift
func handlerAdded(ctx: ChannelHandlerContext) {
    if ctx.channel.isActive {
        self.channelActive(ctx: ctx)
    }
}
```

Alternatively, if you prefer, both `channelActive` and `handlerAdded` could call into a private function that does what `channelActive` currently does.
sergiocampama: Gotcha, thanks for the pointers! I failed to notice that the `flushPreamble` was actually writing to the channel. I went on the private method route, that way I don't have the side effect of triggering the `channelActive` event on subsequent handlers, which could create some confusion.

I also needed to change the test server/client handling, as without the `connect`s, the `isActive` state was not getting enabled.

Lukasa: @sergiocampama Ah, do you want to retarget this PR against the branch called `nghttp2-support-branch`? That branch is where new releases of the 0.x cycle will be cut from: the `master` branch is being used for 1.0 development, which will not have nghttp2. I'll forward port this PR to `master`, but the CI is currently busted so I don't want to block this PR on it.
Lukasa: @swift-nio-bot test this please
weissi: @swift-nio-bot test this please
sergiocampama: fixed the latest test error (missing empty line at the end)
Lukasa: @swift-nio-bot test this please

Patch:
diff --git a/Sources/NIOHTTP2/HTTP2Parser.swift b/Sources/NIOHTTP2/HTTP2Parser.swift
index f4eb6e0..9fe5c15 100644
--- a/Sources/NIOHTTP2/HTTP2Parser.swift
+++ b/Sources/NIOHTTP2/HTTP2Parser.swift
@@ -185,17 +185,19 @@ public final class HTTP2Parser: ChannelInboundHandler, ChannelOutboundHandler {
     }
 
     public func channelActive(ctx: ChannelHandlerContext) {
-        self.session = NGHTTP2Session(mode: self.mode,
-                                      allocator: ctx.channel.allocator,
-                                      maxCachedStreamIDs: self.maxCachedClosedStreams,
-                                      frameReceivedHandler: { ctx.fireChannelRead(self.wrapInboundOut($0)) },
-                                      sendFunction: { ctx.write(self.wrapOutboundOut($0), promise: $1) },
-                                      userEventFunction: { ctx.fireUserInboundEventTriggered($0) })
-
-        self.flushPreamble(ctx: ctx)
+        self.initializeSession(ctx: ctx)
         ctx.fireChannelActive()
     }
 
+    public func handlerAdded(ctx: ChannelHandlerContext) {
+        // Check if the channel is already active when this handler is being added,
+        // in case the handler was added dynamically into the pipeline. If so,
+        // call channelActive to set up the handler.
+        if ctx.channel.isActive {
+            self.initializeSession(ctx: ctx)
+        }
+    }
+
     public func handlerRemoved(ctx: ChannelHandlerContext) {
         self.session = nil
     }
@@ -235,6 +237,19 @@ public final class HTTP2Parser: ChannelInboundHandler, ChannelOutboundHandler {
         self.reentrancyManager.markFlushPoint()
         self.reentrancyManager.process(ctx: ctx, self.process)
     }
+  
+    private func initializeSession(ctx: ChannelHandlerContext) {
+      if self.session == nil {
+          self.session = NGHTTP2Session(mode: self.mode,
+                                        allocator: ctx.channel.allocator,
+                                        maxCachedStreamIDs: self.maxCachedClosedStreams,
+                                        frameReceivedHandler: { ctx.fireChannelRead(self.wrapInboundOut($0)) },
+                                        sendFunction: { ctx.write(self.wrapOutboundOut($0), promise: $1) },
+                                        userEventFunction: { ctx.fireUserInboundEventTriggered($0) })
+      
+          self.flushPreamble(ctx: ctx)
+      }
+    }
 
     private func flushPreamble(ctx: ChannelHandlerContext) {
         let frame = HTTP2Frame(streamID: .rootStream, payload: .settings(self.initialSettings))


Test patch:
diff --git a/Tests/NIOHTTP2Tests/SimpleClientServerTests+XCTest.swift b/Tests/NIOHTTP2Tests/SimpleClientServerTests+XCTest.swift
index ef1bf20..5948638 100644
--- a/Tests/NIOHTTP2Tests/SimpleClientServerTests+XCTest.swift
+++ b/Tests/NIOHTTP2Tests/SimpleClientServerTests+XCTest.swift
@@ -27,6 +27,7 @@ extension SimpleClientServerTests {
    static var allTests : [(String, (SimpleClientServerTests) -> () throws -> Void)] {
       return [
                 ("testBasicRequestResponse", testBasicRequestResponse),
+                ("testBasicRequestResponseWithDynamicPipeline", testBasicRequestResponseWithDynamicPipeline),
                 ("testManyRequestsAtOnce", testManyRequestsAtOnce),
                 ("testNothingButGoaway", testNothingButGoaway),
                 ("testGoAwayWithStreamsUpQuiescing", testGoAwayWithStreamsUpQuiescing),
diff --git a/Tests/NIOHTTP2Tests/SimpleClientServerTests.swift b/Tests/NIOHTTP2Tests/SimpleClientServerTests.swift
index 796d622..6b22644 100644
--- a/Tests/NIOHTTP2Tests/SimpleClientServerTests.swift
+++ b/Tests/NIOHTTP2Tests/SimpleClientServerTests.swift
@@ -169,6 +169,26 @@ final class ClosedEventVsFrameOrderingHandler: ChannelInboundHandler {
     }
 }
 
+/// A simple channel handler that adds the HTTP2Parser handler dynamically
+/// after a read event has been triggered.
+class HTTP2ParserProxyHandler: ChannelInboundHandler {
+    typealias InboundIn = ByteBuffer
+
+    private let maxCachedClosedStreams: Int
+
+    init(maxCachedClosedStreams: Int) {
+        self.maxCachedClosedStreams = maxCachedClosedStreams
+    }
+
+    func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
+        XCTAssertNoThrow(try ctx.pipeline.add(
+            handler: HTTP2Parser(mode: .server, maxCachedClosedStreams: maxCachedClosedStreams
+        )).wait())
+        ctx.fireChannelRead(data)
+        _ = ctx.pipeline.remove(ctx: ctx)
+    }
+}
+
 class SimpleClientServerTests: XCTestCase {
     var clientChannel: EmbeddedChannel!
     var serverChannel: EmbeddedChannel!
@@ -190,6 +210,13 @@ class SimpleClientServerTests: XCTestCase {
         try self.assertDoHandshake(client: self.clientChannel, server: self.serverChannel)
     }
 
+    /// Establish a basic HTTP/2 connection where the HTTP2Parser handler is added after the channel has been activated.
+    func basicHTTP2DynamicPipelineConnection(maxCachedClosedStreams: Int = 1024) throws {
+        XCTAssertNoThrow(try self.clientChannel.pipeline.add(handler: HTTP2Parser(mode: .client, maxCachedClosedStreams: maxCachedClosedStreams)).wait())
+        XCTAssertNoThrow(try self.serverChannel.pipeline.add(handler: HTTP2ParserProxyHandler(maxCachedClosedStreams: maxCachedClosedStreams)).wait())
+        try self.assertDoHandshake(client: self.clientChannel, server: self.serverChannel)
+    }
+
     func testBasicRequestResponse() throws {
         // Begin by getting the connection up.
         try self.basicHTTP2Connection()
@@ -216,6 +243,32 @@ class SimpleClientServerTests: XCTestCase {
         XCTAssertNoThrow(try self.serverChannel.finish())
     }
 
+    func testBasicRequestResponseWithDynamicPipeline() throws {
+        // Begin by getting the connection up.
+        try self.basicHTTP2DynamicPipelineConnection()
+
+        // We're now going to try to send a request from the client to the server.
+        let headers = HTTPHeaders([(":path", "/"), (":method", "POST"), (":scheme", "https"), (":authority", "localhost")])
+        var requestBody = self.clientChannel.allocator.buffer(capacity: 128)
+        requestBody.write(staticString: "A simple HTTP/2 request.")
+
+        let clientStreamID = HTTP2StreamID()
+        let reqFrame = HTTP2Frame(streamID: clientStreamID, payload: .headers(headers))
+        var reqBodyFrame = HTTP2Frame(streamID: clientStreamID, payload: .data(.byteBuffer(requestBody)))
+        reqBodyFrame.flags.insert(.endStream)
+
+        let serverStreamID = try self.assertFramesRoundTrip(frames: [reqFrame, reqBodyFrame], sender: self.clientChannel, receiver: self.serverChannel).first!.streamID
+
+        // Let's send a quick response back.
+        let responseHeaders = HTTPHeaders([(":status", "200"), ("content-length", "0")])
+        var respFrame = HTTP2Frame(streamID: serverStreamID, payload: .headers(responseHeaders))
+        respFrame.flags.insert(.endStream)
+        try self.assertFramesRoundTrip(frames: [respFrame], sender: self.serverChannel, receiver: self.clientChannel)
+
+        XCTAssertNoThrow(try self.clientChannel.finish())
+        XCTAssertNoThrow(try self.serverChannel.finish())
+    }
+
     func testManyRequestsAtOnce() throws {
         // Begin by getting the connection up.
         try self.basicHTTP2Connection()
diff --git a/Tests/NIOHTTP2Tests/TestUtilities.swift b/Tests/NIOHTTP2Tests/TestUtilities.swift
index 876e795..658c365 100644
--- a/Tests/NIOHTTP2Tests/TestUtilities.swift
+++ b/Tests/NIOHTTP2Tests/TestUtilities.swift
@@ -79,8 +79,13 @@ extension XCTestCase {
     func assertDoHandshake(client: EmbeddedChannel, server: EmbeddedChannel,
                            clientSettings: [HTTP2Setting] = nioDefaultSettings, serverSettings: [HTTP2Setting] = nioDefaultSettings,
                            file: StaticString = #file, line: UInt = #line) throws {
-        client.pipeline.fireChannelActive()
-        server.pipeline.fireChannelActive()
+        // This connects are not semantically right, but are required in order to activate the
+        // channels.
+        //! FIXME: Replace with registerAlreadyConfigured0 once EmbeddedChannel propagates this
+        //         call to its channelcore.
+        let socket = try SocketAddress(unixDomainSocketPath: "/fake")
+        _ = try client.connect(to: socket).wait()
+        _ = try server.connect(to: socket).wait()
 
         // First the channels need to interact.
         self.interactInMemory(client, server, file: file, line: line)