Correctly activate child channels. (#27)

Motivation:

Child channels should not activate before their parents.

Modifications:

Delay activation if the parent channel is not currently active.
Activate during channelActive, if needed.

Result:

Easier to start HTTP/2 requests

Author: Lukasa

Sources/NIOHTTP2/HTTP2StreamChannel.swift

@@ -106,12 +106,11 @@ final class HTTP2StreamChannel: Channel, ChannelCore {
             self.autoRead = autoRead
             return initializer?(self, self.streamID) ?? self.eventLoop.newSucceededFuture(result: ())
         }.map {
-            self.state.activate()
-            self.pipeline.fireChannelActive()
-            if self.autoRead {
-                self.read0()
+            // This force unwrap is safe as parent is assigned in the initializer, and never unassigned.
+            // If parent is not active, we expect to receive a channelActive later.
+            if self.parent!.isActive {
+                self.performActivation()
             }
-            self.deliverPendingWrites()
         }
 
         f.whenFailure { (error: Error) in
@@ -124,6 +123,17 @@ final class HTTP2StreamChannel: Channel, ChannelCore {
         return f
     }
 
+    /// Activates this channel.
+    internal func performActivation() {
+        precondition(self.parent?.isActive ?? false, "Parent must be active to activate the child")
+        self.state.activate()
+        self.pipeline.fireChannelActive()
+        if self.autoRead {
+            self.read0()
+        }
+        self.deliverPendingWrites()
+    }
+
     private var _pipeline: ChannelPipeline!
 
     public let allocator: ByteBufferAllocator

Sources/NIOHTTP2/HTTP2StreamMultiplexer.swift

@@ -75,6 +75,17 @@ public final class HTTP2StreamMultiplexer: ChannelInboundHandler, ChannelOutboun
         ctx.write(data, promise: promise)
     }
 
+    public func channelActive(ctx: ChannelHandlerContext) {
+        // We just got channelActive. Any previously existing channels may be marked active.
+        for channel in self.streams.values {
+            // We double-check the channel activity here, because it's possible action taken during
+            // the activation of one of the child channels will cause the parent to close!
+            if ctx.channel.isActive {
+                channel.performActivation()
+            }
+        }
+    }
+
     public func userInboundEventTriggered(ctx: ChannelHandlerContext, event: Any) {
         // The only event we care about right now is StreamClosedEvent, and in particular
         // we only care about it if we still have the stream channel for the stream

Tests/NIOHTTP2Tests/HTTP2StreamMultiplexerTests+XCTest.swift

@@ -54,6 +54,9 @@ extension HTTP2StreamMultiplexerTests {
                 ("testWritesOnCreatedChannelAreDelayed", testWritesOnCreatedChannelAreDelayed),
                 ("testWritesAreCancelledOnFailingInitializer", testWritesAreCancelledOnFailingInitializer),
                 ("testFailingInitializerDoesNotWrite", testFailingInitializerDoesNotWrite),
+                ("testCreatedChildChannelDoesNotActivateEarly", testCreatedChildChannelDoesNotActivateEarly),
+                ("testCreatedChildChannelActivatesIfParentIsActive", testCreatedChildChannelActivatesIfParentIsActive),
+                ("testInitiatedChildChannelActivates", testInitiatedChildChannelActivates),
            ]
    }
 }

Tests/NIOHTTP2Tests/HTTP2StreamMultiplexerTests.swift

@@ -126,6 +126,28 @@ final class HandlerRemovedHandler: ChannelInboundHandler {
 }
 
 
+/// A channel handler that succeeds a promise when its channel becomes active.
+final class ActiveHandler: ChannelInboundHandler {
+    typealias InboundIn = Any
+
+    let activatedPromise: EventLoopPromise<Void>
+
+    init(activatedPromise: EventLoopPromise<Void>) {
+        self.activatedPromise = activatedPromise
+    }
+
+    func handlerAdded(ctx: ChannelHandlerContext) {
+        if ctx.channel.isActive {
+            self.activatedPromise.succeed(result: ())
+        }
+    }
+
+    func channelActive(ctx: ChannelHandlerContext) {
+        self.activatedPromise.succeed(result: ())
+    }
+}
+
+
 final class HTTP2StreamMultiplexerTests: XCTestCase {
     var channel: EmbeddedChannel!
 
@@ -199,6 +221,8 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
     func testChannelsCloseAfterResetStreamFrameFirstThenEvent() throws {
         var closeError: Error? = nil
 
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // First, set up the frames we want to send/receive.
         let streamID = HTTP2StreamID(knownID: Int32(1))
         var frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -233,6 +257,8 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
     func testChannelsCloseAfterGoawayFrameFirstThenEvent() throws {
         var closeError: Error? = nil
 
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // First, set up the frames we want to send/receive.
         let streamID = HTTP2StreamID(knownID: Int32(1))
         var frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -345,6 +371,9 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: frameReceiver).wait())
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
 
+
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // Let's send a headers frame to open the stream.
         let streamID = HTTP2StreamID(knownID: 1)
         let frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -372,6 +401,9 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: frameReceiver).wait())
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
 
+
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // Let's send a headers frame to open the stream.
         let streamID = HTTP2StreamID(knownID: 1)
         let frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -406,6 +438,9 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: frameReceiver).wait())
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
 
+
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // Let's send a headers frame to open the stream.
         let streamID = HTTP2StreamID(knownID: 1)
         let frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -459,6 +494,8 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: frameReceiver).wait())
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
 
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // Let's send a headers frame to open the stream.
         let streamID = HTTP2StreamID(knownID: 1)
         let frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -494,6 +531,9 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         }
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
 
+
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // Let's send a headers frame to open the stream.
         let streamID = HTTP2StreamID(knownID: 1)
         let frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -608,6 +648,8 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: writeTracker).wait())
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
 
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // Let's open two streams.
         let firstStreamID = HTTP2StreamID(knownID: 1)
         let secondStreamID = HTTP2StreamID(knownID: 3)
@@ -751,6 +793,9 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         }
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
 
+
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // Let's open a stream.
         let streamID = HTTP2StreamID(knownID: 1)
         let frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -797,6 +842,8 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         }
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
 
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // Let's open two streams.
         for streamID in [firstStreamID, secondStreamID] {
             let frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -836,6 +883,8 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         }
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
 
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // Let's open a stream.
         let streamID = HTTP2StreamID(knownID: 1)
         let frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -884,6 +933,8 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: readCounter).wait())
         XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
 
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         // Let's open a stream.
         let streamID = HTTP2StreamID(knownID: 1)
         let frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
@@ -1039,6 +1090,8 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
         var childChannel: Channel? = nil
         var childStreamID: HTTP2StreamID? = nil
 
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
         let multiplexer = HTTP2StreamMultiplexer { (_, _) in
             XCTFail("Must not be called")
             return self.channel.eventLoop.newFailedFuture(error: MyError())
@@ -1112,4 +1165,91 @@ final class HTTP2StreamMultiplexerTests: XCTestCase {
 
         XCTAssertNoThrow(try self.channel.finish())
     }
+
+    func testCreatedChildChannelDoesNotActivateEarly() throws {
+        var activated = false
+
+        let activePromise: EventLoopPromise<Void> = self.channel.eventLoop.newPromise()
+        let activeRecorder = ActiveHandler(activatedPromise: activePromise)
+        activePromise.futureResult.map {
+            activated = true
+        }.whenFailure { (_: Error) in
+            XCTFail("Activation promise must not fail")
+        }
+
+        let multiplexer = HTTP2StreamMultiplexer { (_, _) in
+            XCTFail("Must not be called")
+            return self.channel.eventLoop.newFailedFuture(error: MyError())
+        }
+        XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
+        multiplexer.createStreamChannel(promise: nil) { (channel, streamID) in
+            return channel.pipeline.add(handler: activeRecorder)
+        }
+        (self.channel.eventLoop as! EmbeddedEventLoop).run()
+        XCTAssertFalse(activated)
+
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
+        XCTAssertTrue(activated)
+
+        XCTAssertNoThrow(try self.channel.finish())
+    }
+
+    func testCreatedChildChannelActivatesIfParentIsActive() throws {
+        var activated = false
+
+        let activePromise: EventLoopPromise<Void> = self.channel.eventLoop.newPromise()
+        let activeRecorder = ActiveHandler(activatedPromise: activePromise)
+        activePromise.futureResult.map {
+            activated = true
+        }.whenFailure { (_: Error) in
+            XCTFail("Activation promise must not fail")
+        }
+
+        let multiplexer = HTTP2StreamMultiplexer { (_, _) in
+            XCTFail("Must not be called")
+            return self.channel.eventLoop.newFailedFuture(error: MyError())
+        }
+        XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
+
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(ipAddress: "127.0.0.1", port: 8765)).wait())
+        XCTAssertFalse(activated)
+
+        multiplexer.createStreamChannel(promise: nil) { (channel, streamID) in
+            return channel.pipeline.add(handler: activeRecorder)
+        }
+        (self.channel.eventLoop as! EmbeddedEventLoop).run()
+        XCTAssertTrue(activated)
+
+        XCTAssertNoThrow(try self.channel.finish())
+    }
+
+    func testInitiatedChildChannelActivates() throws {
+        XCTAssertNoThrow(try self.channel.connect(to: SocketAddress(unixDomainSocketPath: "/whatever"), promise: nil))
+
+        var activated = false
+
+        let activePromise: EventLoopPromise<Void> = self.channel.eventLoop.newPromise()
+        let activeRecorder = ActiveHandler(activatedPromise: activePromise)
+        activePromise.futureResult.map {
+            activated = true
+        }.whenFailure { (_: Error) in
+            XCTFail("Activation promise must not fail")
+        }
+
+        let multiplexer = HTTP2StreamMultiplexer { (channel, _) in
+            return channel.pipeline.add(handler: activeRecorder)
+        }
+        XCTAssertNoThrow(try self.channel.pipeline.add(handler: multiplexer).wait())
+        self.channel.pipeline.fireChannelActive()
+
+        // Open a new stream.
+        XCTAssertFalse(activated)
+        let streamID = HTTP2StreamID(knownID: 1)
+        let frame = HTTP2Frame(streamID: streamID, payload: .headers(HTTPHeaders()))
+        XCTAssertNoThrow(try self.channel.writeInbound(frame))
+        XCTAssertTrue(activated)
+
+        XCTAssertNoThrow(try self.channel.finish())
+    }
 }