Close stream channels when the parent channel becomes inactive (#131)

Motivation:

Stream channels don't get closed when the parent channel closes. This
can lead to some unexpected behaviour for users.

Modifications:

Call receiveStreamClosed on stream channels in the multiplexer when
reciving channelInactive. Added a test to verify the stream is closed.

Result:

Stream channels will be closed when the parent channel is closed.

Sources/NIOHTTP2/HTTP2StreamMultiplexer.swift

@@ -97,6 +97,13 @@ public final class HTTP2StreamMultiplexer: ChannelInboundHandler, ChannelOutboun
         context.fireChannelActive()
     }
 
+    public func channelInactive(context: ChannelHandlerContext) {
+        for channel in self.streams.values {
+            channel.receiveStreamClosed(nil)
+        }
+        context.fireChannelInactive()
+    }
+
     public func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) {
         switch event {
         case let evt as StreamClosedEvent:

Tests/NIOHTTP2Tests/ConfiguringPipelineTests+XCTest.swift

@@ -29,6 +29,7 @@ extension ConfiguringPipelineTests {
                 ("testBasicPipelineCommunicates", testBasicPipelineCommunicates),
                 ("testPipelineRespectsPositionRequest", testPipelineRespectsPositionRequest),
                 ("testPreambleGetsWrittenOnce", testPreambleGetsWrittenOnce),
+                ("testClosingParentChannelClosesStreamChannel", testClosingParentChannelClosesStreamChannel),
            ]
    }
 }

Tests/NIOHTTP2Tests/ConfiguringPipelineTests.swift

@@ -138,4 +138,47 @@ class ConfiguringPipelineTests: XCTestCase {
         // We don't expect anything else at this point.
         XCTAssertNil(try self.serverChannel.readOutbound(as: IOData.self))
     }
+
+    func testClosingParentChannelClosesStreamChannel() throws {
+        /// A channel handler that succeeds a promise when the channel becomes inactive.
+        final class InactiveHandler: ChannelInboundHandler {
+            typealias InboundIn = Any
+
+            let inactivePromise: EventLoopPromise<Void>
+
+            init(inactivePromise: EventLoopPromise<Void>) {
+                self.inactivePromise = inactivePromise
+            }
+
+            func channelInactive(context: ChannelHandlerContext) {
+                inactivePromise.succeed(())
+            }
+        }
+
+        let clientMultiplexer = try assertNoThrowWithValue(self.clientChannel.configureHTTP2Pipeline(mode: .client).wait())
+        XCTAssertNoThrow(try self.serverChannel.configureHTTP2Pipeline(mode: .server).wait())
+
+        XCTAssertNoThrow(try self.assertDoHandshake(client: self.clientChannel, server: self.serverChannel))
+
+        let inactivePromise: EventLoopPromise<Void> = self.clientChannel.eventLoop.makePromise()
+        let streamChannelPromise: EventLoopPromise<Channel> = self.clientChannel.eventLoop.makePromise()
+
+        clientMultiplexer.createStreamChannel(promise: streamChannelPromise) { channel, _ in
+            return channel.pipeline.addHandler(InactiveHandler(inactivePromise: inactivePromise))
+        }
+
+        (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
+
+        let streamChannel = try assertNoThrowWithValue(try streamChannelPromise.futureResult.wait())
+        // Close the parent channel, not the stream channel.
+        XCTAssertNoThrow(try self.clientChannel.close().wait())
+
+        XCTAssertNoThrow(try inactivePromise.futureResult.wait())
+        XCTAssertFalse(streamChannel.isActive)
+
+        XCTAssertThrowsError(try self.clientChannel.finish()) { error in
+            XCTAssertEqual(error as? ChannelError, .alreadyClosed)
+        }
+        XCTAssertNoThrow(try self.serverChannel.finish())
+    }
 }