Update to a newer version of NIO HTTP/2 (#1904)

Motivation:

swift-nio-http2 1.32.0 contains new API we need for configuring an async
http/2 pipeline with a stream delegate.

Modifications:

- Bump the minimum version
- Wire through the stream delegate
- Fix a test bug where the server didn't send the final status

Result:

Stream delegates are wired up

Package.swift

@@ -36,7 +36,7 @@ let packageDependencies: [Package.Dependency] = [
   ),
   .package(
     url: "https://github.com/apple/swift-nio-http2.git",
-    from: "1.31.0"
+    from: "1.32.0"
   ),
   .package(
     url: "https://github.com/apple/swift-nio-transport-services.git",

Sources/GRPCHTTP2Core/Client/Connection/ClientConnectionHandler.swift

@@ -247,8 +247,8 @@ final class ClientConnectionHandler: ChannelInboundHandler, ChannelOutboundHandl
         switch self.state.beginGracefulShutdown(promise: promise) {
         case .sendGoAway(let close):
           context.fireChannelRead(self.wrapInboundOut(.closing(.initiatedLocally)))
-          // Clients should send GOAWAYs when closing a connection.
-          self.writeAndFlushGoAway(context: context, errorCode: .noError)
+          // The client could send a GOAWAY at this point but it's not really necessary, the server
+          // can't open streams anyway, the client will just close the connection when it's done.
           if close {
             context.close(promise: nil)
           }

Sources/GRPCHTTP2Core/Internal/NIOChannelPipeline+GRPC.swift

@@ -69,6 +69,7 @@ extension ChannelPipeline.SynchronousOperations {
 
     let streamMultiplexer = try self.configureAsyncHTTP2Pipeline(
       mode: .server,
+      streamDelegate: serverConnectionHandler.http2StreamDelegate,
       configuration: NIOHTTP2Handler.Configuration(
         connection: http2HandlerConnectionConfiguration,
         stream: http2HandlerStreamConfiguration
@@ -137,8 +138,17 @@ extension ChannelPipeline.SynchronousOperations {
       HTTP2Setting(parameter: .maxHeaderListSize, value: HPACKDecoder.defaultMaxHeaderListSize),
     ]
 
+    let connectionHandler = ClientConnectionHandler(
+      eventLoop: self.eventLoop,
+      maxIdleTime: config.idle.map { TimeAmount($0.maxTime) },
+      keepaliveTime: config.keepalive.map { TimeAmount($0.time) },
+      keepaliveTimeout: config.keepalive.map { TimeAmount($0.timeout) },
+      keepaliveWithoutCalls: config.keepalive?.permitWithoutCalls ?? false
+    )
+
     let multiplexer = try self.configureAsyncHTTP2Pipeline(
       mode: .client,
+      streamDelegate: connectionHandler.http2StreamDelegate,
       configuration: http2
     ) { stream in
       // Shouldn't happen, push-promises are disabled so the server shouldn't be able to
@@ -146,14 +156,6 @@ extension ChannelPipeline.SynchronousOperations {
       stream.close()
     }
 
-    let connectionHandler = ClientConnectionHandler(
-      eventLoop: self.eventLoop,
-      maxIdleTime: config.idle.map { TimeAmount($0.maxTime) },
-      keepaliveTime: config.keepalive.map { TimeAmount($0.time) },
-      keepaliveTimeout: config.keepalive.map { TimeAmount($0.timeout) },
-      keepaliveWithoutCalls: config.keepalive?.permitWithoutCalls ?? false
-    )
-
     try self.addHandler(connectionHandler)
 
     let connection = try NIOAsyncChannel(

Tests/GRPCHTTP2CoreTests/Client/Connection/GRPCChannelTests.swift

@@ -329,8 +329,6 @@ final class GRPCChannelTests: XCTestCase {
   }
 
   func testCloseWhenRPCsAreInProgress() async throws {
-    try XCTSkipIf(true, "https://github.com/apple/swift-nio-http2/pull/439")
-
     // Verify that closing the channel while there are RPCs in progress allows the RPCs to finish
     // gracefully.
 

Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift

@@ -386,8 +386,6 @@ final class SubchannelTests: XCTestCase {
   }
 
   func testConnectionDropWithOpenStreams() async throws {
-    try XCTSkipIf(true, "HTTP/2 stream delegate API isn't currently exposed")
-
     let server = TestServer(eventLoopGroup: .singletonMultiThreadedEventLoopGroup)
     let address = try await server.bind()
     let subchannel = self.makeSubchannel(address: address, connector: .posix())
@@ -439,6 +437,8 @@ final class SubchannelTests: XCTestCase {
         .connectivityStateChanged(.connecting),
         .connectivityStateChanged(.ready),
         .connectivityStateChanged(.transientFailure),
+        .requiresNameResolution,
+        .connectivityStateChanged(.connecting),
         .connectivityStateChanged(.shutdown),
       ]
 

Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/HTTP2Connectors.swift

@@ -115,15 +115,6 @@ struct NIOPosixConnector: HTTP2Connector {
       channel.eventLoop.makeCompletedFuture {
         let sync = channel.pipeline.syncOperations
 
-        let multiplexer = try sync.configureAsyncHTTP2Pipeline(mode: .client) { stream in
-          // Server shouldn't be opening streams.
-          stream.close()
-        }
-
-        if self.dropPingAcks {
-          try sync.addHandler(PingAckDropper())
-        }
-
         let connectionHandler = ClientConnectionHandler(
           eventLoop: channel.eventLoop,
           maxIdleTime: self.maxIdleTime,
@@ -132,6 +123,18 @@ struct NIOPosixConnector: HTTP2Connector {
           keepaliveWithoutCalls: self.keepaliveWithoutCalls
         )
 
+        let multiplexer = try sync.configureAsyncHTTP2Pipeline(
+          mode: .client,
+          streamDelegate: connectionHandler.http2StreamDelegate
+        ) { stream in
+          // Server shouldn't be opening streams.
+          stream.close()
+        }
+
+        if self.dropPingAcks {
+          try sync.addHandler(PingAckDropper())
+        }
+
         try sync.addHandler(connectionHandler)
 
         let asyncChannel = try NIOAsyncChannel<ClientConnectionEvent, Void>(

Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/TestServer.swift

@@ -164,6 +164,7 @@ extension TestServer {
             try await outbound.write(.message(bytes))
           }
         }
+        try await outbound.write(.status(Status(code: .ok, message: ""), [:]))
       }
 
     case .never: