Fix ByteBody stream writing for HTTP/2 (#11261)

The writer wrote ByteBufs directly instead of HttpContent, which works for HTTP/1 but not HTTP/2. Also improved error handling.

Fixes #11260

---------

Co-authored-by: Alan Shijun Wang <[email protected]>

Author: yawkat

http-client-jdk/src/test/groovy/io/micronaut/http/client/jdk/Http2Spec.groovy

@@ -4,8 +4,10 @@ import io.micronaut.context.annotation.Property
 import io.micronaut.context.annotation.Requires
 import io.micronaut.http.HttpRequest
 import io.micronaut.http.MediaType
+import io.micronaut.http.annotation.Consumes
 import io.micronaut.http.annotation.Controller
 import io.micronaut.http.annotation.Get
+import io.micronaut.http.annotation.Post
 import io.micronaut.http.annotation.Produces
 import io.micronaut.http.client.HttpClient
 import io.micronaut.http.client.HttpVersionSelection
@@ -20,6 +22,7 @@ import spock.lang.Specification
 @Property(name = "micronaut.server.ssl.build-self-signed", value = "true")
 @Property(name = "micronaut.server.ssl.enabled", value = "true")
 @Property(name = "micronaut.http.client.ssl.insecure-trust-all-certificates", value = "true")
+@Property(name = "micronaut.server.max-request-size", value = "16384")
 class Http2Spec extends Specification {
 
     @Inject
@@ -34,6 +37,18 @@ class Http2Spec extends Specification {
         response == "hello"
     }
 
+    def "test http2 post long string"() {
+        when:
+        def body = Map.of("q", "a" * 8097)
+
+        def response = client.toBlocking().retrieve(HttpRequest.POST("/http2", body)
+                .contentType(MediaType.APPLICATION_FORM_URLENCODED_TYPE))
+
+        then:
+        response == "hello"
+    }
+
+
     @Controller("/http2")
     @Requires(property = "spec.name", value = "Http2Spec")
     static class SpecController {
@@ -43,5 +58,12 @@ class Http2Spec extends Specification {
         String get() {
             "hello"
         }
+
+        @Post
+        @Consumes(MediaType.APPLICATION_FORM_URLENCODED)
+        @Produces(MediaType.TEXT_PLAIN)
+        String post() {
+            "hello"
+        }
     }
 }

http-client/src/main/java/io/micronaut/http/client/netty/StreamWriter.java

@@ -23,6 +23,7 @@
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.http.DefaultHttpContent;
 import io.netty.handler.codec.http.LastHttpContent;
 
 import java.util.function.Consumer;
@@ -99,14 +100,16 @@ private void add0(ByteBuf buf) {
         }
 
         int readable = buf.readableBytes();
-        ctx.writeAndFlush(buf).addListener((ChannelFutureListener) future -> {
+        ctx.writeAndFlush(new DefaultHttpContent(buf)).addListener((ChannelFutureListener) future -> {
             assert ctx.executor().inEventLoop();
             if (future.isSuccess()) {
                 if (ctx.channel().isWritable()) {
                     upstream.onBytesConsumed(readable);
                 } else {
                     unwritten += readable;
                 }
+            } else {
+                error(future.cause());
             }
         });
     }

http-client/src/test/groovy/io/micronaut/http/client/netty/Http2Spec.groovy

@@ -0,0 +1,66 @@
+package io.micronaut.http.client.netty
+
+import io.micronaut.context.annotation.Property
+import io.micronaut.context.annotation.Requires
+import io.micronaut.http.HttpRequest
+import io.micronaut.http.MediaType
+import io.micronaut.http.annotation.*
+import io.micronaut.http.client.HttpClient
+import io.micronaut.http.client.HttpVersionSelection
+import io.micronaut.http.client.annotation.Client
+import io.micronaut.test.extensions.spock.annotation.MicronautTest
+import jakarta.inject.Inject
+import spock.lang.Specification
+
+@MicronautTest
+@Property(name = "spec.name", value = "Http2Spec")
+@Property(name = "micronaut.server.http-version", value = "HTTP_2_0")
+@Property(name = "micronaut.server.ssl.build-self-signed", value = "true")
+@Property(name = "micronaut.server.ssl.enabled", value = "true")
+@Property(name = "micronaut.http.client.ssl.insecure-trust-all-certificates", value = "true")
+@Property(name = "micronaut.server.max-request-size", value = "16384")
+class Http2Spec extends Specification {
+
+    @Inject
+    @Client(value = "/", alpnModes = HttpVersionSelection.ALPN_HTTP_2)
+    HttpClient client
+
+    def "test http2"() {
+        when:
+        def response = client.toBlocking().retrieve(HttpRequest.GET("/http2"))
+
+        then:
+        response == "hello"
+    }
+
+    def "test http2 post long string"() {
+        when:
+        // Body must be >8096 bytes, according to io.netty.handler.codec.http.multipart.HttpPostBodyUtil#chunkSize
+        def body = Map.of("q", "a" * 8097)
+
+        def response = client.toBlocking().retrieve(HttpRequest.POST("/http2", body)
+                .contentType(MediaType.APPLICATION_FORM_URLENCODED_TYPE))
+
+        then:
+        response == "hello"
+    }
+
+
+    @Controller("/http2")
+    @Requires(property = "spec.name", value = "Http2Spec")
+    static class SpecController {
+
+        @Get
+        @Produces(MediaType.TEXT_PLAIN)
+        String get() {
+            "hello"
+        }
+
+        @Post
+        @Consumes(MediaType.APPLICATION_FORM_URLENCODED)
+        @Produces(MediaType.TEXT_PLAIN)
+        String post(@Body Map<String, String> ignored) {
+            "hello"
+        }
+    }
+}