Set up HTTP2Parser in handlerAdded if the channel had been previously activated. (#36)

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

Author: sergiocampama

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))

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),

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()

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)