Repository: quarkusio/quarkus
Issue: Resteasy Reactive Rest Client fails to re-construct large chunks of streamed json (stream+json) and fails deserialization

### Describe the bug

Note: This is related to #30040 in that both result in `JsonMappingException`s, but this is a different issue.

I am trying to use `RestMediaType.APPLICATION_STREAM_JSON` to stream a `Multi` that contains a few larger objects (think ~25 kB each) from my reactive java server to a reactive java client, using the rest client.

However, the deserialization of the objects fails with `Unexpected end-of-input` errors, either at the end of `1943` bytes or `8192` bytes. (For the `1943` case,  I think there is potentially header info on the first chunk that is cutting into the `8192` total, since when I try to collect the chunks as Strings instead of reading them as my custom objects, all but the first String are 8192 chars long)

Through my own investigation I have found the workaround is to increase the Vertx `HttpClientOptions` for `DEFAULT_MAX_CHUNK_SIZE` and `DEFAULT_RECEIVE_BUFFER_SIZE`, both of which are defaulted to `8192`

Maybe that is the final solution, but I would think that the rest client should be able to use the newline-delimited json to determine when it has received an entire object, and buffer/hold in memory up to that point before attempting to deserialize. 

### Expected behavior

Rest client can properly receive and deserialize a stream of larger json objects, without having to fiddle with the http client options myself

### Actual behavior

Errors in the client:

```
com.fasterxml.jackson.databind.JsonMappingException: Unexpected end-of-input in field name
 at [Source: (ByteArrayInputStream); line: 1, column: 1943]
...
com.fasterxml.jackson.core.io.JsonEOFException: Unexpected end-of-input in VALUE_STRING
 at [Source: (ByteArrayInputStream); line: 1, column: 8192]
```

### How to Reproduce?

Reproducer: `service3` (client) and `service4` (server) in https://github.com/fleckware/quarkus-test

1. in one terminal `./gradlew service4:quarkusDev`
2. in a separate terminal `./gradlew service3:quarkusDev`
3. in a separate terminal `curl localhost:8081/service4/test/large > output.json`
4. observe that when hitting the streaming endpoint on the server directly, there is no issue
5. now try to hit it using the rest client `curl localhost:8080/service3/test/large`
6. observe the json mapping error in the `service3` terminal

### Output of `uname -a` or `ver`

_No response_

### Output of `java -version`

openjdk 11.0.10

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

2.15.0

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

gradle 7.4.2

### Additional information

_No response_

Changed files:
extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/Message.java
extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/SseTestCase.java
extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/SseResource.java
independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java

diff --git a/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/Message.java b/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/Message.java
similarity index 70%
rename from extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/Message.java
rename to extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/Message.java
index c09c92acb96..79109e22e20 100644
--- a/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/Message.java
+++ b/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/Message.java
@@ -1,4 +1,4 @@
-package io.quarkus.resteasy.reactive.jackson.deployment.test.sse;
+package io.quarkus.resteasy.reactive.jackson.deployment.test.streams;
 
 public class Message {
     public String name;
diff --git a/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/SseResource.java b/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamResource.java
similarity index 84%
rename from extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/SseResource.java
rename to extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamResource.java
index 3176a97a817..fd5c2b81a72 100644
--- a/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/SseResource.java
+++ b/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamResource.java
@@ -1,6 +1,9 @@
-package io.quarkus.resteasy.reactive.jackson.deployment.test.sse;
+package io.quarkus.resteasy.reactive.jackson.deployment.test.streams;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
 
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
@@ -16,8 +19,8 @@
 import io.smallrye.common.annotation.Blocking;
 import io.smallrye.mutiny.Multi;
 
-@Path("sse")
-public class SseResource {
+@Path("streams")
+public class StreamResource {
 
     @GET
     @Produces(MediaType.SERVER_SENT_EVENTS)
@@ -120,4 +123,21 @@ public Multi<Message> multiStreamJson() {
         return Multi.createFrom().items(new Message("hello"), new Message("stef"));
     }
 
+    /**
+     * Reproduce <a href="https://github.com/quarkusio/quarkus/issues/30044">#30044</a>.
+     */
+    @Path("stream-json/multi/fast")
+    @GET
+    @Produces(RestMediaType.APPLICATION_STREAM_JSON)
+    @RestStreamElementType(MediaType.APPLICATION_JSON)
+    public Multi<Message> multiStreamJsonFast() {
+        List<UUID> ids = new ArrayList<>(5000);
+        for (int i = 0; i < 5000; i++) {
+            ids.add(UUID.randomUUID());
+        }
+        return Multi.createFrom().items(ids::stream)
+                .onItem().transform(id -> new Message(id.toString()))
+                .onOverflow().buffer(81920);
+    }
+
 }
diff --git a/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/SseTestCase.java b/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamTestCase.java
similarity index 78%
rename from extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/SseTestCase.java
rename to extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamTestCase.java
index afb909111ba..bc82769dd1d 100644
--- a/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/sse/SseTestCase.java
+++ b/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamTestCase.java
@@ -1,4 +1,4 @@
-package io.quarkus.resteasy.reactive.jackson.deployment.test.sse;
+package io.quarkus.resteasy.reactive.jackson.deployment.test.streams;
 
 import static io.restassured.RestAssured.when;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -33,7 +33,7 @@
 import io.quarkus.test.common.http.TestHTTPResource;
 import io.smallrye.mutiny.Multi;
 
-public class SseTestCase {
+public class StreamTestCase {
 
     @TestHTTPResource
     URI uri;
@@ -41,16 +41,16 @@ public class SseTestCase {
     @RegisterExtension
     static final QuarkusUnitTest config = new QuarkusUnitTest()
             .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
-                    .addClasses(SseResource.class, Message.class));
+                    .addClasses(StreamResource.class, Message.class));
 
     @Test
     public void testSseFromSse() throws Exception {
-        testSse("sse");
+        testSse("streams");
     }
 
     @Test
     public void testSseFromMulti() throws Exception {
-        testSse("sse/multi");
+        testSse("streams/multi");
     }
 
     private void testSse(String path) throws Exception {
@@ -81,12 +81,12 @@ public void accept(Throwable throwable) {
 
     @Test
     public void testMultiFromSse() {
-        testMulti("sse");
+        testMulti("streams");
     }
 
     @Test
     public void testMultiFromMulti() {
-        testMulti("sse/multi");
+        testMulti("streams/multi");
     }
 
     private void testMulti(String path) {
@@ -99,24 +99,24 @@ private void testMulti(String path) {
 
     @Test
     public void testJsonMultiFromSse() {
-        testJsonMulti("sse/json");
-        testJsonMulti("sse/json2");
-        testJsonMulti("sse/blocking/json");
+        testJsonMulti("streams/json");
+        testJsonMulti("streams/json2");
+        testJsonMulti("streams/blocking/json");
     }
 
     @Test
     public void testJsonMultiFromMulti() {
-        testJsonMulti("sse/json/multi");
+        testJsonMulti("streams/json/multi");
     }
 
     @Test
     public void testJsonMultiFromMultiWithDefaultElementType() {
-        testJsonMulti("sse/json/multi2");
+        testJsonMulti("streams/json/multi2");
     }
 
     @Test
     public void testNdJsonMultiFromMulti() {
-        when().get(uri.toString() + "sse/ndjson/multi")
+        when().get(uri.toString() + "streams/ndjson/multi")
                 .then().statusCode(HttpStatus.SC_OK)
                 // @formatter:off
                 .body(is("{\\"name\\":\\"hello\\"}\\n"
@@ -127,7 +127,7 @@ public void testNdJsonMultiFromMulti() {
 
     @Test
     public void testStreamJsonMultiFromMulti() {
-        when().get(uri.toString() + "sse/stream-json/multi")
+        when().get(uri.toString() + "streams/stream-json/multi")
                 .then().statusCode(HttpStatus.SC_OK)
                 // @formatter:off
                 .body(is("{\\"name\\":\\"hello\\"}\\n"
@@ -143,4 +143,19 @@ private void testJsonMulti(String path) {
         List<Message> list = multi.collect().asList().await().atMost(Duration.ofSeconds(30));
         assertThat(list).extracting("name").containsExactly("hello", "stef");
     }
+
+    /**
+     * Reproduce <a href="https://github.com/quarkusio/quarkus/issues/30044">#30044</a>.
+     */
+    @Test
+    public void testStreamJsonMultiFromMultiFast() {
+        String payload = when().get(uri.toString() + "streams/stream-json/multi/fast")
+                .then().statusCode(HttpStatus.SC_OK)
+                .header(HttpHeaders.CONTENT_TYPE, containsString(RestMediaType.APPLICATION_STREAM_JSON))
+                .extract().response().asString();
+
+        // the payload include 5000 json objects
+        assertThat(payload.lines()).hasSize(5000)
+                .allSatisfy(s -> assertThat(s).matches("\\\\{\\"name\\":\\".*\\"}"));
+    }
 }
diff --git a/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java b/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java
index c6edb7584aa..aa4129bbd27 100644
--- a/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java
+++ b/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java
@@ -1,6 +1,7 @@
 package org.jboss.resteasy.reactive.client.impl;
 
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -18,6 +19,7 @@
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpClientResponse;
 import io.vertx.core.net.impl.ConnectionBase;
+import io.vertx.core.parsetools.RecordParser;
 
 public class MultiInvoker extends AbstractRxInvoker<Multi<?>> {
 
@@ -121,10 +123,13 @@ public <R> Multi<R> method(String name, Entity<?> entity, GenericType<R> respons
                 } else {
                     HttpClientResponse vertxResponse = restClientRequestContext.getVertxClientResponse();
                     if (!emitter.isCancelled()) {
-                        // FIXME: this is probably not good enough
                         if (response.getStatus() == 200
                                 && MediaType.SERVER_SENT_EVENTS_TYPE.isCompatible(response.getMediaType())) {
                             registerForSse(multiRequest, responseType, response, vertxResponse);
+                        } else if (response.getStatus() == 200
+                                && RestMediaType.APPLICATION_STREAM_JSON_TYPE.isCompatible(response.getMediaType())) {
+                            registerForJsonStream(multiRequest, restClientRequestContext, responseType, response,
+                                    vertxResponse);
                         } else {
                             // read stuff in chunks
                             registerForChunks(multiRequest, restClientRequestContext, responseType, response, vertxResponse);
@@ -267,4 +272,46 @@ private boolean startsWith(byte[] array, byte[] prefix) {
         });
     }
 
+    private <R> void registerForJsonStream(MultiRequest<? super R> multiRequest,
+            RestClientRequestContext restClientRequestContext,
+            GenericType<R> responseType,
+            ResponseImpl response,
+            HttpClientResponse vertxClientResponse) {
+        RecordParser parser = RecordParser.newDelimited("\\n");
+        parser.handler(new Handler<Buffer>() {
+            @Override
+            public void handle(Buffer chunk) {
+
+                ByteArrayInputStream in = new ByteArrayInputStream(chunk.getBytes());
+                try {
+                    R item = restClientRequestContext.readEntity(in, responseType, response.getMediaType(),
+                            response.getMetadata());
+                    multiRequest.emitter.emit(item);
+                } catch (IOException e) {
+                    multiRequest.emitter.fail(e);
+                }
+            }
+        });
+        vertxClientResponse.exceptionHandler(t -> {
+            if (t == ConnectionBase.CLOSED_EXCEPTION) {
+                // we can ignore this one since we registered a closeHandler
+            } else {
+                multiRequest.emitter.fail(t);
+            }
+        });
+        vertxClientResponse.endHandler(new Handler<Void>() {
+            @Override
+            public void handle(Void c) {
+                multiRequest.complete();
+            }
+        });
+
+        vertxClientResponse.handler(parser);
+
+        // watch for user cancelling
+        multiRequest.onCancel(() -> {
+            vertxClientResponse.request().connection().close();
+        });
+    }
+
 }