Fix streaming of json strings in RESTEasy Reactive

Fixes: quarkusio#18043
nenros · Sep 8, 2022 · a82a205 · a82a205
1 parent 93cd6a8
commit a82a205
Show file tree

Hide file tree

Showing 8 changed files with 110 additions and 6 deletions.
diff --git a/...ent/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/StreamingTest.java b/...ent/src/test/java/io/quarkus/resteasy/reactive/jackson/deployment/test/StreamingTest.java
@@ -0,0 +1,45 @@
+package io.quarkus.resteasy.reactive.jackson.deployment.test;
+
+import static io.restassured.RestAssured.when;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+
+import org.hamcrest.CoreMatchers;
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.smallrye.mutiny.Multi;
+
+public class StreamingTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest config = new QuarkusUnitTest()
+            .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
+                    .addClasses(StreamingResource.class));
+
+    @Test
+    public void testSseMultiJsonString() {
+        when().get("/test/multi")
+                .then()
+                .statusCode(200)
+                .body(CoreMatchers.is("[\"Hello\",\"Hola\"]"));
+    }
+
+    @Path("/test")
+    public static class StreamingResource {
+
+        @GET
+        @Path("multi")
+        @Produces(MediaType.APPLICATION_JSON)
+        public Multi<String> multi() {
+            return Multi.createFrom().items("Hello", "Hola");
+        }
+    }
+
+}
diff --git a/...yment/src/test/java/io/quarkus/resteasy/reactive/jsonb/deployment/test/StreamingTest.java b/...yment/src/test/java/io/quarkus/resteasy/reactive/jsonb/deployment/test/StreamingTest.java
@@ -0,0 +1,45 @@
+package io.quarkus.resteasy.reactive.jsonb.deployment.test;
+
+import static io.restassured.RestAssured.when;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+
+import org.hamcrest.CoreMatchers;
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.smallrye.mutiny.Multi;
+
+public class StreamingTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest config = new QuarkusUnitTest()
+            .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
+                    .addClasses(StreamingResource.class));
+
+    @Test
+    public void testSseMultiJsonString() {
+        when().get("/test/multi")
+                .then()
+                .statusCode(200)
+                .body(CoreMatchers.is("[\"Hello\",\"Hola\"]"));
+    }
+
+    @Path("/test")
+    public static class StreamingResource {
+
+        @GET
+        @Path("multi")
+        @Produces(MediaType.APPLICATION_JSON)
+        public Multi<String> multi() {
+            return Multi.createFrom().items("Hello", "Hola");
+        }
+    }
+
+}
diff --git a/...nt-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/MultiNdjsonTest.java b/...nt-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/MultiNdjsonTest.java
@@ -55,7 +55,7 @@ void shouldReadNdjsonStringAsMulti() throws InterruptedException {
             fail("Streaming did not complete in time");
         }
         assertThat(collected).hasSize(4)
-                .contains("one", "two", "three", "four");
+                .contains("\"one\"", "\"two\"", "\"three\"", "\"four\"");
     }
 
     @Test

diff --git a/...ent-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/StreamJsonTest.java b/...ent-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/StreamJsonTest.java
@@ -55,7 +55,7 @@ void shouldReadStreamJsonStringAsMulti() throws InterruptedException {
             fail("Streaming did not complete in time");
         }
         assertThat(collected).hasSize(4)
-                .contains("one", "two", "3", "four");
+                .contains("\"one\"", "\"two\"", "\"3\"", "\"four\"");
     }
 
     @Test

diff --git a/...rc/main/java/org/jboss/resteasy/reactive/server/jackson/JacksonMessageBodyWriterUtil.java b/...rc/main/java/org/jboss/resteasy/reactive/server/jackson/JacksonMessageBodyWriterUtil.java
@@ -12,6 +12,7 @@
 import java.lang.annotation.Annotation;
 import java.nio.charset.StandardCharsets;
 import javax.ws.rs.core.MultivaluedMap;
+import org.jboss.resteasy.reactive.server.StreamingOutputStream;
 
 public final class JacksonMessageBodyWriterUtil {
 
@@ -43,7 +44,8 @@ public static void setNecessaryJsonFactoryConfig(JsonFactory jsonFactory) {
     public static void doLegacyWrite(Object o, Annotation[] annotations, MultivaluedMap<String, Object> httpHeaders,
             OutputStream entityStream, ObjectWriter defaultWriter) throws IOException {
         setContentTypeIfNecessary(httpHeaders);
-        if (o instanceof String) { // YUK: done in order to avoid adding extra quotes...
+        if ((o instanceof String) && (!(entityStream instanceof StreamingOutputStream))) {
+            // YUK: done in order to avoid adding extra quotes... when we are not streaming a result
             entityStream.write(((String) o).getBytes(StandardCharsets.UTF_8));
         } else {
             if (annotations != null) {

diff --git a/.../jsonb/src/main/java/org/jboss/resteasy/reactive/server/jsonb/JsonbMessageBodyWriter.java b/.../jsonb/src/main/java/org/jboss/resteasy/reactive/server/jsonb/JsonbMessageBodyWriter.java
@@ -11,6 +11,7 @@
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.MultivaluedMap;
 import org.jboss.resteasy.reactive.common.providers.serialisers.JsonMessageBodyWriterUtil;
+import org.jboss.resteasy.reactive.server.StreamingOutputStream;
 import org.jboss.resteasy.reactive.server.spi.ServerMessageBodyWriter;
 import org.jboss.resteasy.reactive.server.spi.ServerRequestContext;
 
@@ -27,7 +28,8 @@ public JsonbMessageBodyWriter(Jsonb json) {
     public void writeTo(Object o, Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType,
             MultivaluedMap<String, Object> httpHeaders, OutputStream entityStream) throws IOException, WebApplicationException {
         JsonMessageBodyWriterUtil.setContentTypeIfNecessary(httpHeaders);
-        if (o instanceof String) { // YUK: done in order to avoid adding extra quotes...
+        if ((o instanceof String) && (!(entityStream instanceof StreamingOutputStream))) {
+            // YUK: done in order to avoid adding extra quotes... when we are not streaming a result
             entityStream.write(((String) o).getBytes(StandardCharsets.UTF_8));
         } else {
             json.toJson(o, type, entityStream);

diff --git a/...erver/runtime/src/main/java/org/jboss/resteasy/reactive/server/StreamingOutputStream.java b/...erver/runtime/src/main/java/org/jboss/resteasy/reactive/server/StreamingOutputStream.java
@@ -0,0 +1,10 @@
+package org.jboss.resteasy.reactive.server;
+
+import java.io.ByteArrayOutputStream;
+
+/**
+ * The only reason we use this is to give MessageBodyWriter classes the ability to tell
+ * if they are being called in a streaming context
+ */
+public class StreamingOutputStream extends ByteArrayOutputStream {
+}
diff --git a/...e/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/StreamingUtil.java b/...e/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/core/StreamingUtil.java
@@ -1,6 +1,5 @@
 package org.jboss.resteasy.reactive.server.core;
 
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.lang.reflect.Type;
 import java.nio.charset.StandardCharsets;
@@ -13,6 +12,7 @@
 import javax.ws.rs.core.Response;
 import javax.ws.rs.ext.MessageBodyWriter;
 import org.jboss.resteasy.reactive.common.core.Serialisers;
+import org.jboss.resteasy.reactive.server.StreamingOutputStream;
 import org.jboss.resteasy.reactive.server.handlers.PublisherResponseHandler;
 import org.jboss.resteasy.reactive.server.spi.ServerHttpResponse;
 
@@ -56,7 +56,7 @@ private static byte[] serialiseEntity(ResteasyReactiveRequestContext context, Ob
         MessageBodyWriter<Object>[] writers = (MessageBodyWriter<Object>[]) serialisers
                 .findWriters(null, entityClass, mediaType, RuntimeType.SERVER)
                 .toArray(ServerSerialisers.NO_WRITER);
-        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        StreamingOutputStream baos = new StreamingOutputStream();
         boolean wrote = false;
         for (MessageBodyWriter<Object> writer : writers) {
             // Spec(API) says we should use class/type/mediaType but doesn't talk about annotations