HTTP/2 prior knowledge support (#10761)

When accepting HTTP/2 plaintext connections, also accept prior knowledge HTTP/2. This actually already mostly worked for the new Http2ServerHandler, the changes are only necessary for the old multiplex handler.

Also fixed some test flakiness.

Fixes #6301

http-server-netty/src/main/java/io/micronaut/http/server/netty/HttpPipelineBuilder.java

@@ -506,12 +506,22 @@ void configureForH2cSupport() {
 
             final Http2FrameCodec frameCodec;
             final Http2ConnectionHandler connectionHandler;
+            Http2MultiplexHandler multiplexHandler;
             if (server.getServerConfiguration().isLegacyMultiplexHandlers()) {
                 frameCodec = createHttp2FrameCodec();
                 connectionHandler = frameCodec;
+                multiplexHandler = new Http2MultiplexHandler(new ChannelInitializer<Http2StreamChannel>() {
+                    @Override
+                    protected void initChannel(@NonNull Http2StreamChannel ch) {
+                        StreamPipeline streamPipeline = new StreamPipeline(ch, sslHandler, connectionCustomizer.specializeForChannel(ch, NettyServerCustomizer.ChannelRole.REQUEST_STREAM));
+                        streamPipeline.insertHttp2FrameHandlers();
+                        streamPipeline.streamCustomizer.onStreamPipelineBuilt();
+                    }
+                });
             } else {
                 connectionHandler = createHttp2ServerHandler(false);
                 frameCodec = null;
+                multiplexHandler = null;
             }
             final String fallbackHandlerName = "http1-fallback-handler";
             HttpServerUpgradeHandler.UpgradeCodecFactory upgradeCodecFactory = protocol -> {
@@ -537,14 +547,7 @@ public void upgradeTo(ChannelHandlerContext ctx, FullHttpRequest upgradeRequest)
                     if (frameCodec == null) {
                         return new Http2ServerUpgradeCodecImpl(connectionHandler);
                     } else {
-                        return new Http2ServerUpgradeCodecImpl(frameCodec, new Http2MultiplexHandler(new ChannelInitializer<Http2StreamChannel>() {
-                            @Override
-                            protected void initChannel(@NonNull Http2StreamChannel ch) {
-                                StreamPipeline streamPipeline = new StreamPipeline(ch, sslHandler, connectionCustomizer.specializeForChannel(ch, NettyServerCustomizer.ChannelRole.REQUEST_STREAM));
-                                streamPipeline.insertHttp2FrameHandlers();
-                                streamPipeline.streamCustomizer.onStreamPipelineBuilt();
-                            }
-                        }));
+                        return new Http2ServerUpgradeCodecImpl(frameCodec, multiplexHandler);
                     }
                 } else {
                     return null;
@@ -557,8 +560,14 @@ protected void initChannel(@NonNull Http2StreamChannel ch) {
                     upgradeCodecFactory,
                     server.getServerConfiguration().getMaxH2cUpgradeRequestSize()
             );
+            ChannelHandler priorKnowledgeHandler = frameCodec == null ? connectionHandler : new ChannelInitializer<>() {
+                @Override
+                protected void initChannel(@NonNull Channel ch) {
+                    ch.pipeline().addLast(connectionHandler, multiplexHandler);
+                }
+            };
             final CleartextHttp2ServerUpgradeHandler cleartextHttp2ServerUpgradeHandler =
-                    new CleartextHttp2ServerUpgradeHandler(sourceCodec, upgradeHandler, connectionHandler);
+                    new CleartextHttp2ServerUpgradeHandler(sourceCodec, upgradeHandler, priorKnowledgeHandler);
 
             pipeline.addLast(cleartextHttp2ServerUpgradeHandler);
             pipeline.addLast(fallbackHandlerName, new SimpleChannelInboundHandler<HttpMessage>() {

http-server-netty/src/main/java/io/micronaut/http/server/netty/NettyHttpRequest.java

@@ -295,7 +295,8 @@ public HttpVersion getHttpVersion() {
         if (pipeline != null) {
             return pipeline.httpVersion;
         }
-        return HttpVersion.HTTP_1_1;
+        // Http2ServerHandler case
+        return findConnectionHandler() == null ? HttpVersion.HTTP_1_1 : HttpVersion.HTTP_2_0;
     }
 
     @Override

http-server-netty/src/test/groovy/io/micronaut/http/server/netty/Http2CompressionSpec.groovy

@@ -9,6 +9,7 @@ class Http2CompressionSpec extends CompressionSpec {
 
                 'micronaut.server.http-version': '2.0',
                 'micronaut.server.ssl.enabled': true,
+                'micronaut.server.ssl.port': 0,
                 'micronaut.server.ssl.build-self-signed': true,
         ] as Map<String, Object>
     }

http-server-netty/src/test/groovy/io/micronaut/http/server/netty/context/ContextURISpec.groovy

@@ -23,11 +23,11 @@ class ContextURISpec extends Specification {
     void "test getContextURI returns the base URI when context path is not set"() {
         when:
         EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer, [
-                'micronaut.server.port': 60006
+                'micronaut.server.port': 60007
         ])
 
         then:
-        embeddedServer.getContextURI().toString() == 'http://localhost:60006'
+        embeddedServer.getContextURI().toString() == 'http://localhost:60007'
 
         cleanup:
         embeddedServer.close()
@@ -37,11 +37,11 @@ class ContextURISpec extends Specification {
         when:
         EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer, [
                 'micronaut.server.context-path': '',
-                'micronaut.server.port': 60006
+                'micronaut.server.port': 60008
         ])
 
         then:
-        embeddedServer.getContextURI().toString() == 'http://localhost:60006'
+        embeddedServer.getContextURI().toString() == 'http://localhost:60008'
 
         cleanup:
         embeddedServer.close()

http-server-netty/src/test/groovy/io/micronaut/http/server/netty/http2/H2cSpec.groovy

@@ -229,6 +229,72 @@ class H2cSpec extends Specification {
         content.release()
     }
 
+    def 'prior knowledge'() {
+        given:
+        def responseFuture = new CompletableFuture()
+
+        def group = new NioEventLoopGroup(1)
+        def bootstrap = new Bootstrap()
+                .remoteAddress(embeddedServer.host, embeddedServer.port)
+                .group(group)
+                .channel(NioSocketChannel.class)
+                .handler(new ChannelInitializer<SocketChannel>() {
+                    @Override
+                    protected void initChannel(@NonNull SocketChannel ch) throws Exception {
+                        def http2Connection = new DefaultHttp2Connection(false)
+                        def inboundAdapter = new InboundHttp2ToHttpAdapterBuilder(http2Connection)
+                                .maxContentLength(1000000)
+                                .validateHttpHeaders(true)
+                                .propagateSettings(true)
+                                .build()
+                        def connectionHandler = new HttpToHttp2ConnectionHandlerBuilder()
+                                .connection(http2Connection)
+                                .frameListener(new DelegatingDecompressorFrameListener(http2Connection, inboundAdapter))
+                                .build()
+
+                        ch.pipeline()
+                                .addLast(connectionHandler)
+                                .addLast(new ChannelInboundHandlerAdapter() {
+                                    @Override
+                                    void channelRead(@NonNull ChannelHandlerContext ctx, @NonNull Object msg) throws Exception {
+                                        ctx.read()
+                                        if (msg instanceof HttpMessage) {
+                                            if (msg.headers().getInt(HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text(), -1) != 3) {
+                                                responseFuture.completeExceptionally(new AssertionError("Response must be on stream 3"));
+                                            }
+                                            responseFuture.complete(ReferenceCountUtil.retain(msg))
+                                        }
+                                        super.channelRead(ctx, msg)
+                                    }
+
+                                    @Override
+                                    void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+                                        super.exceptionCaught(ctx, cause)
+                                        cause.printStackTrace()
+                                        responseFuture.completeExceptionally(cause)
+                                    }
+                                })
+
+                    }
+                })
+
+        def channel = (SocketChannel) bootstrap.connect().await().channel()
+
+        def request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, '/h2c/test')
+        request.headers().set(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text(), "http")
+        channel.writeAndFlush(request)
+        channel.read()
+
+        expect:
+        def resp = responseFuture.get(10, TimeUnit.SECONDS)
+        resp != null
+
+        cleanup:
+        channel.close()
+        resp.release()
+        group.shutdownGracefully()
+    }
+
     @Controller("/h2c")
     @Requires(property = "spec.name", value = "H2cSpec")
     static class TestController {