Fix InputStream handling as Response entity in RESTEasy Classic

Fixes: quarkusio#21602

extensions/resteasy-classic/resteasy/deployment/src/test/java/io/quarkus/resteasy/test/InputStreamResponseLargePayloadWithRemainderTest.java

@@ -0,0 +1,43 @@
+package io.quarkus.resteasy.test;
+
+import static io.restassured.RestAssured.when;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Response;
+
+import org.hamcrest.Matchers;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+
+public class InputStreamResponseLargePayloadWithRemainderTest {
+
+    @RegisterExtension
+    static QuarkusUnitTest runner = new QuarkusUnitTest()
+            .withApplicationRoot((jar) -> jar
+                    .addClasses(RootResource.class)
+                    .addAsResource(new StringAsset("quarkus.resteasy.vertx.response-buffer-size=11\n"),
+                            "application.properties"));
+
+    @Test
+    public void test() {
+        when().get("/test").then().body(Matchers.is("Hello World!"));
+    }
+
+    @Path("test")
+    public static class TestResource {
+
+        @Produces("text/plain")
+        @GET
+        public Response test() {
+            return Response.ok(new ByteArrayInputStream("Hello World!".getBytes(StandardCharsets.UTF_8))).build();
+        }
+    }
+}

extensions/resteasy-classic/resteasy/deployment/src/test/java/io/quarkus/resteasy/test/InputStreamResponseLargePayloadWithoutRemainderTest.java

@@ -0,0 +1,43 @@
+package io.quarkus.resteasy.test;
+
+import static io.restassured.RestAssured.when;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Response;
+
+import org.hamcrest.Matchers;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+
+public class InputStreamResponseLargePayloadWithoutRemainderTest {
+
+    @RegisterExtension
+    static QuarkusUnitTest runner = new QuarkusUnitTest()
+            .withApplicationRoot((jar) -> jar
+                    .addClasses(RootResource.class)
+                    .addAsResource(new StringAsset("quarkus.resteasy.vertx.response-buffer-size=3\n"),
+                            "application.properties"));
+
+    @Test
+    public void test() {
+        when().get("/test").then().body(Matchers.is("Hello World!"));
+    }
+
+    @Path("test")
+    public static class TestResource {
+
+        @Produces("text/plain")
+        @GET
+        public Response test() {
+            return Response.ok(new ByteArrayInputStream("Hello World!".getBytes(StandardCharsets.UTF_8))).build();
+        }
+    }
+}

extensions/resteasy-classic/resteasy/deployment/src/test/java/io/quarkus/resteasy/test/InputStreamResponseTest.java

@@ -0,0 +1,40 @@
+package io.quarkus.resteasy.test;
+
+import static io.restassured.RestAssured.when;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Response;
+
+import org.hamcrest.Matchers;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+
+public class InputStreamResponseTest {
+
+    @RegisterExtension
+    static QuarkusUnitTest runner = new QuarkusUnitTest()
+            .withApplicationRoot((jar) -> jar
+                    .addClasses(RootResource.class));
+
+    @Test
+    public void test() {
+        when().get("/test").then().body(Matchers.is("Hello World"));
+    }
+
+    @Path("test")
+    public static class TestResource {
+
+        @Produces("text/plain")
+        @GET
+        public Response test() {
+            return Response.ok(new ByteArrayInputStream("Hello World".getBytes(StandardCharsets.UTF_8))).build();
+        }
+    }
+}

extensions/resteasy-classic/resteasy/runtime/src/main/java/io/quarkus/resteasy/runtime/standalone/VertxOutputStream.java

@@ -167,13 +167,21 @@ public CompletionStage<Void> asyncWrite(final byte[] b, final int off, final int
         }
 
         if (bufferCount == 1) {
-            pooledBuffer = allocator.allocateBuffer();
-            pooledBuffer.writeBytes(b);
+            if (pooledBuffer == null) {
+                pooledBuffer = allocator.allocateBuffer();
+            }
+            pooledBuffer.writeBytes(b, 0, len);
         } else {
             for (int i = 0; i < bufferCount - 1; i++) {
                 int bufferIndex = i;
                 ret = ret.thenCompose(v -> {
-                    ByteBuf tmpBuf = allocator.allocateBuffer();
+                    ByteBuf tmpBuf = null;
+                    if ((bufferIndex == 0) && ((pooledBuffer != null))) {
+                        tmpBuf = pooledBuffer;
+                    }
+                    if (tmpBuf == null) {
+                        tmpBuf = allocator.allocateBuffer();
+                    }
                     tmpBuf.writeBytes(b, bufferIndex * bufferSize, bufferSize);
                     return response.writeNonBlocking(tmpBuf, false);
                 });