Finish multiplexer's inbound streams in more cases

Author: gjcairo

Sources/NIOHTTP2/HTTP2ChannelHandler+InboundStreamMultiplexer.swift

@@ -146,6 +146,24 @@ extension NIOHTTP2Handler.InboundStreamMultiplexer {
         }
     }
 
+    func errorCaughtReceived(_ error: any Error) {
+        switch self {
+        case .inline(let inlineStreamMultiplexer):
+            inlineStreamMultiplexer.propagateErrorCaught(error)
+        case .legacy:
+            break  // do nothing
+        }
+    }
+
+    func handlerRemovedReceived() {
+        switch self {
+        case .inline(let inlineStreamMultiplexer):
+            inlineStreamMultiplexer.propagateHandlerRemoved()
+        case .legacy:
+            break  // do nothing
+        }
+    }
+
     func processedFrame(_ frame: HTTP2Frame) {
         switch self {
         case .inline(let inlineStreamMultiplexer):

Sources/NIOHTTP2/HTTP2ChannelHandler+InlineStreamMultiplexer.swift

@@ -150,6 +150,14 @@ extension InlineStreamMultiplexer {
         }
     }
 
+    internal func propagateErrorCaught(_ error: any Error) {
+        self._commonStreamMultiplexer.propagateErrorCaught(error)
+    }
+
+    internal func propagateHandlerRemoved() {
+        self._commonStreamMultiplexer.propagateHandlerRemoved()
+    }
+
     internal func processedFrame(frame: HTTP2Frame) {
         self._commonStreamMultiplexer.processedFrame(streamID: frame.streamID, size: frame.payload.flowControlledSize)
     }

Sources/NIOHTTP2/HTTP2ChannelHandler.swift

@@ -414,6 +414,7 @@ public final class NIOHTTP2Handler: ChannelDuplexHandler {
     public func handlerRemoved(context: ChannelHandlerContext) {
         // Any frames we're buffering need to be dropped.
         self.outboundBuffer.invalidateBuffer()
+        self.inboundStreamMultiplexer?.handlerRemovedReceived()
         self.inboundStreamMultiplexerState = .deinitialized
     }
 
@@ -550,6 +551,11 @@ public final class NIOHTTP2Handler: ChannelDuplexHandler {
         self.inboundStreamMultiplexer?.channelWritabilityChangedReceived()
         context.fireChannelWritabilityChanged()
     }
+
+    public func errorCaught(context: ChannelHandlerContext, error: any Error) {
+        self.inboundStreamMultiplexer?.errorCaughtReceived(error)
+        context.fireErrorCaught(error)
+    }
 }
 
 /// Inbound frame handling.

Sources/NIOHTTP2/HTTP2CommonInboundStreamMultiplexer.swift

@@ -213,6 +213,30 @@ extension HTTP2CommonInboundStreamMultiplexer {
         self.streamChannelContinuation?.finish()
     }
 
+    internal func propagateErrorCaught(_ error: any Error) {
+        for channel in self.streams.values {
+            channel.receiveStreamClosed(nil)
+        }
+
+        for channel in self._pendingStreams.values {
+            channel.receiveStreamClosed(nil)
+        }
+
+        self.streamChannelContinuation?.finish(throwing: error)
+    }
+
+    internal func propagateHandlerRemoved() {
+        for channel in self.streams.values {
+            channel.receiveStreamClosed(nil)
+        }
+
+        for channel in self._pendingStreams.values {
+            channel.receiveStreamClosed(nil)
+        }
+
+        self.streamChannelContinuation?.finish()
+    }
+
     internal func selectivelyPropagateUserInboundEvent(context: ChannelHandlerContext, event: Any) {
         func propagateEvent(_ event: Any) {
             for channel in self.streams.values {