Repository: quarkusio/quarkus
Issue: Json stream item is not processed immediately after receiving in ChunkedStreamingMultiSubscriber.class

### Describe the bug

When using rest endpoint with `application/stream+json` content type, after receiving json object is not processed immediately. 
Client is expecting stream element separator to process element. Separator is sent as prefix with subsequent elements. This means that stream element is processed after receiving next one.

### Expected behavior

Stream element should be processed immediately after receiving.

Maybe ChunkedStreamingMultiSubscriber should add separator as postfix, not as prefix with next item.

### Actual behavior

Strem element is processed after receiving next one (or stream closed).

### How to Reproduce?

Reproducer: https://github.com/paloliska/json-stream

steps:
1. Run test in reproducer project.

### Output of `uname -a` or `ver`

Linux user 5.15.0-58-generic #64-Ubuntu SMP Thu Jan 5 11:43:13 UTC 2023 x86_64 x86_64 x86_64 GNU/Linux

### Output of `java -version`

openjdk version "17.0.5" 2022-10-18

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

2.16.0.Final, 3.0.1.Final

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

Apache Maven 3.6.3

### Additional information

_No response_

Changed files:
independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java
extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/StreamJsonTest.java
extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamTestCase.java
independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java
independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/StreamingUtil.java

diff --git a/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamTestCase.java b/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamTestCase.java
index 8c7704d24c4..2ae2a7f2895 100644
--- a/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamTestCase.java
+++ b/extensions/resteasy-reactive/quarkus-resteasy-reactive-jackson/deployment/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/streams/StreamTestCase.java
@@ -153,7 +153,6 @@ public void testStreamJsonMultiFromMulti() {
 
     private void testJsonMulti(String path) {
         Client client = ClientBuilder.newBuilder().register(new JacksonBasicMessageBodyReader(new ObjectMapper())).build();
-
         WebTarget target = client.target(uri.toString() + path);
         Multi<Message> multi = target.request().rx(MultiInvoker.class).get(Message.class);
         List<Message> list = multi.collect().asList().await().atMost(Duration.ofSeconds(30));
diff --git a/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/StreamJsonTest.java b/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/StreamJsonTest.java
index 07344da0fae..21ae3dd8b2e 100644
--- a/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/StreamJsonTest.java
+++ b/extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/StreamJsonTest.java
@@ -4,6 +4,7 @@
 import static org.junit.jupiter.api.Assertions.fail;
 
 import java.net.URI;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
@@ -32,10 +33,14 @@
 import io.quarkus.vertx.web.ReactiveRoutes;
 import io.quarkus.vertx.web.Route;
 import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.helpers.test.AssertSubscriber;
 import io.vertx.core.Vertx;
 import io.vertx.ext.web.RoutingContext;
 
 public class StreamJsonTest {
+
+    private static final long TICK_EVERY_MS = 200;
+
     @RegisterExtension
     static final QuarkusUnitTest TEST = new QuarkusUnitTest()
             .withApplicationRoot((jar) -> jar.addClasses(TestJacksonBasicMessageBodyReader.class));
@@ -109,6 +114,21 @@ void shouldReadNdjsonFromSingleMessage() throws InterruptedException {
         assertThat(collected).hasSize(4).containsAll(expected);
     }
 
+    /**
+     * Reproduce <a href="https://github.com/quarkusio/quarkus/issues/30690">#30690</a>.
+     */
+    @Test
+    public void shouldReadUpToThreeTicks() {
+        createClient(uri)
+                .ticks()
+                .onItem()
+                .invoke(Objects::nonNull)
+                .subscribe()
+                .withSubscriber(AssertSubscriber.create(3))
+                // wait for 3 ticks plus some half tick ms of extra time (this should not be necessary, but CI is slow)
+                .awaitItems(3, Duration.ofMillis((TICK_EVERY_MS * 3) + (TICK_EVERY_MS / 2)));
+    }
+
     private Client createClient(URI uri) {
         return RestClientBuilder.newBuilder().baseUri(uri).register(new TestJacksonBasicMessageBodyReader())
                 .build(Client.class);
@@ -133,6 +153,12 @@ public interface Client {
         @Produces(RestMediaType.APPLICATION_STREAM_JSON)
         @RestStreamElementType(MediaType.APPLICATION_JSON)
         Multi<Message> readPojoSingle();
+
+        @GET
+        @Path("/ticks")
+        @Produces(RestMediaType.APPLICATION_STREAM_JSON)
+        @RestStreamElementType(MediaType.APPLICATION_JSON)
+        Multi<String> ticks();
     }
 
     public static class ReactiveRoutesResource {
@@ -199,6 +225,19 @@ public String getPojosAsString() throws JsonProcessingException {
             }
             return result.toString();
         }
+
+        @GET
+        @Path("/ticks")
+        @Produces(RestMediaType.APPLICATION_STREAM_JSON)
+        @RestStreamElementType(MediaType.APPLICATION_JSON)
+        public Multi<String> getTicks() {
+            return Multi.createFrom()
+                    .ticks()
+                    .every(Duration.ofMillis(TICK_EVERY_MS))
+                    .log()
+                    .onItem()
+                    .transform((Long tick) -> "tick " + tick);
+        }
     }
 
     public static class Message {
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 e3a8820add5..e84f30da72e 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
@@ -199,12 +199,9 @@ public void handle(Buffer buffer) {
                         charset = charset == null ? "UTF-8" : charset;
                         byte[] separator = "\\n".getBytes(charset);
                         int start = 0;
-                        if (startsWith(bytes, separator)) {
-                            start += separator.length;
-                        }
                         while (start < bytes.length) {
                             int end = bytes.length;
-                            for (int i = start; i < bytes.length - separator.length; i++) {
+                            for (int i = start; i < end; i++) {
                                 if (bytes[i] == separator[0]) {
                                     int j;
                                     boolean matches = true;
@@ -222,7 +219,7 @@ public void handle(Buffer buffer) {
                             }
 
                             if (start < end) {
-                                ByteArrayInputStream in = new ByteArrayInputStream(bytes, start, end - start);
+                                ByteArrayInputStream in = new ByteArrayInputStream(bytes, start, end);
                                 R item = restClientRequestContext.readEntity(in, responseType, mediaType,
                                         response.getMetadata());
                                 multiRequest.emitter.emit(item);
@@ -241,18 +238,6 @@ public void handle(Buffer buffer) {
                     multiRequest.emitter.fail(t);
                 }
             }
-
-            private boolean startsWith(byte[] array, byte[] prefix) {
-                if (array.length < prefix.length) {
-                    return false;
-                }
-                for (int i = 0; i < prefix.length; i++) {
-                    if (array[i] != prefix[i]) {
-                        return false;
-                    }
-                }
-                return true;
-            }
         });
 
         // this captures the end of the response
diff --git a/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/StreamingUtil.java b/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/StreamingUtil.java
index b32281d6833..fbcec897700 100644
--- a/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/StreamingUtil.java
+++ b/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/StreamingUtil.java
@@ -24,7 +24,8 @@
 public class StreamingUtil {
 
     public static CompletionStage<?> send(ResteasyReactiveRequestContext context,
-            List<PublisherResponseHandler.StreamingResponseCustomizer> customizers, Object entity, String prefix) {
+            List<PublisherResponseHandler.StreamingResponseCustomizer> customizers, Object entity, String prefix,
+            String suffix) {
         ServerHttpResponse response = context.serverResponse();
         if (response.closed()) {
             // FIXME: check spec
@@ -46,6 +47,13 @@ public static CompletionStage<?> send(ResteasyReactiveRequestContext context,
             System.arraycopy(data, 0, prefixedData, prefixBytes.length, data.length);
             data = prefixedData;
         }
+        if (suffix != null) {
+            byte[] suffixBytes = suffix.getBytes(StandardCharsets.US_ASCII);
+            byte[] suffixedData = new byte[data.length + suffixBytes.length];
+            System.arraycopy(data, 0, suffixedData, 0, data.length);
+            System.arraycopy(suffixBytes, 0, suffixedData, data.length, suffixBytes.length);
+            data = suffixedData;
+        }
         return response.write(data);
     }
 
diff --git a/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java b/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java
index 1791a388da2..12c8f2722e7 100644
--- a/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java
+++ b/independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java
@@ -90,18 +90,18 @@ private static class ChunkedStreamingMultiSubscriber extends StreamingMultiSubsc
         @Override
         protected String messagePrefix() {
             // When message is chunked, we don't need to add prefixes at first
-            if (isFirstItem) {
-                isFirstItem = false;
-                return null;
-            }
+            return null;
+        }
 
-            // If it's not the first message, we need to append the messages with end of line delimiter.
+        @Override
+        protected String messageSuffix() {
             return LINE_SEPARATOR;
         }
 
         @Override
         protected String onCompleteText() {
-            return LINE_SEPARATOR;
+            // When message is chunked, we don't need to add text at the end of the messages
+            return null;
         }
     }
 
@@ -128,7 +128,7 @@ private static class StreamingMultiSubscriber extends AbstractMultiSubscriber {
         public void onNext(Object item) {
             List<StreamingResponseCustomizer> customizers = determineCustomizers(!hadItem);
             hadItem = true;
-            StreamingUtil.send(requestContext, customizers, item, messagePrefix())
+            StreamingUtil.send(requestContext, customizers, item, messagePrefix(), messageSuffix())
                     .handle(new BiFunction<Object, Throwable, Object>() {
                         @Override
                         public Object apply(Object v, Throwable t) {
@@ -182,11 +182,15 @@ public void onComplete() {
             }
             if (json) {
                 String postfix = onCompleteText();
-                byte[] postfixBytes = postfix.getBytes(StandardCharsets.US_ASCII);
-                requestContext.serverResponse().write(postfixBytes).handle((v, t) -> {
+                if (postfix != null) {
+                    byte[] postfixBytes = postfix.getBytes(StandardCharsets.US_ASCII);
+                    requestContext.serverResponse().write(postfixBytes).handle((v, t) -> {
+                        super.onComplete();
+                        return null;
+                    });
+                } else {
                     super.onComplete();
-                    return null;
-                });
+                }
             } else {
                 super.onComplete();
             }
@@ -209,6 +213,10 @@ protected String messagePrefix() {
             // if it's json, the message prefix starts with `[`.
             return json ? nextJsonPrefix : null;
         }
+
+        protected String messageSuffix() {
+            return null;
+        }
     }
 
     static abstract class AbstractMultiSubscriber implements Subscriber<Object> {