Do not hold on to pending channels after close.

Motivation:

We temporarily store channel creation requests in NIOSSHHandler before
passing them to the multiplexer while we wait for the handshake to
complete. This is fine, but if for any reason we get removed from the
pipeline while we're still holding some, they must all be cancelled.

Modifications:

- Correctly cancel creation requests on handlerRemoved.
- Verify behaviour in tests.

Result:

Fewer hangs on channel creation.

Sources/NIOSSH/NIOSSHHandler.swift

@@ -113,6 +113,9 @@ extension NIOSSHHandler: ChannelDuplexHandler {
 
         self.dropAllPendingGlobalRequests(ChannelError.eof)
         self.dropUnsatisfiedGlobalRequests(ChannelError.eof)
+        while let next = self.pendingChannelInitializations.popFirst() {
+            next.promise?.fail(ChannelError.eof)
+        }
     }
 
     public func channelActive(context: ChannelHandlerContext) {

Tests/NIOSSHTests/ChildChannelMultiplexerTests.swift

@@ -61,6 +61,16 @@ final class ErrorLoggingHandler: ChannelInboundHandler {
     }
 }
 
+final class ErrorClosingHandler: ChannelInboundHandler {
+    typealias InboundIn = Any
+    typealias InboundOut = Any
+
+    func errorCaught(context: ChannelHandlerContext, error: Error) {
+        context.close(promise: nil)
+        context.fireErrorCaught(error)
+    }
+}
+
 final class ReadCountingHandler: ChannelOutboundHandler {
     typealias OutboundIn = Any
     typealias OutboundOut = Any

Tests/NIOSSHTests/EndToEndTests.swift

@@ -527,4 +527,41 @@ class EndToEndTests: XCTestCase {
         XCTAssertEqual(errorCatcher.errors.count, 1)
         XCTAssertEqual(errorCatcher.errors.first as? TestError, .bang)
     }
+
+    func testCreateChannelBeforeIncompleteHandshakeFails() throws {
+        enum TestError: Error {
+            case bang
+        }
+
+        struct RejectDelegate: NIOSSHClientServerAuthenticationDelegate {
+            func validateHostKey(hostKey: NIOSSHPublicKey, validationCompletePromise: EventLoopPromise<Void>) {
+                validationCompletePromise.fail(TestError.bang)
+            }
+        }
+
+        var harness = TestHarness()
+        harness.clientServerAuthDelegate = RejectDelegate()
+
+        XCTAssertNoThrow(try self.channel.configureWithHarness(harness))
+        XCTAssertNoThrow(try self.channel.client.pipeline.addHandler(ErrorClosingHandler()).wait())
+
+        // Get an early ref to the handler and try to create a child channel.
+        let handler = self.channel.clientSSHHandler
+
+        var err: Error?
+        let promise = self.channel.client.eventLoop.makePromise(of: Channel.self)
+        promise.futureResult.whenFailure { error in err = error }
+        handler!.createChannel(promise, channelType: .session) { channel, _ in
+            channel.eventLoop.makeSucceededFuture(())
+        }
+        XCTAssertNil(err)
+
+        // Activation errors.
+        XCTAssertNoThrow(try self.channel.activate())
+        XCTAssertThrowsError(try self.channel.interactInMemory()) { error in
+            XCTAssertEqual(error as? TestError, .bang)
+        }
+        self.channel.run()
+        XCTAssertEqual(err as? ChannelError?, .eof)
+    }
 }