Fixes support for application calls to flush() when using StreamingOutput

common/reactive/src/main/java/io/helidon/common/reactive/OutputStreamPublisher.java

@@ -29,6 +29,8 @@
 @SuppressWarnings("WeakerAccess")
 public class OutputStreamPublisher extends OutputStream implements Flow.Publisher<ByteBuffer> {
 
+    private static final byte[] FLUSH_BUFFER = new byte[0];
+
     private final SingleSubscriberHolder<ByteBuffer> subscriber = new SingleSubscriberHolder<>();
     private final Object invocationLock = new Object();
 
@@ -72,6 +74,16 @@ public void close() throws IOException {
         complete();
     }
 
+    /**
+     * Send empty buffer as an indication of a user-requested flush.
+     *
+     * @throws IOException If an I/O occurs.
+     */
+    @Override
+    public void flush() throws IOException {
+        publish(FLUSH_BUFFER, 0, 0);
+    }
+
     private void publish(byte[] buffer, int offset, int length) throws IOException {
         Objects.requireNonNull(buffer);
 

webserver/jersey/src/main/java/io/helidon/webserver/jersey/ResponseWriter.java

@@ -18,6 +18,7 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -135,12 +136,9 @@ public void write(int b) throws IOException {
             res.headers().put(entry.getKey(), entry.getValue());
         }
 
-        // in case of SSE every response chunk needs to be flushed
-        boolean doFlush = MediaType.SERVER_SENT_EVENTS_TYPE.isCompatible(context.getMediaType());
-
         res.send(ReactiveStreamsAdapter.publisherToFlow(
                     ReactiveStreamsAdapter.publisherFromFlow(publisher)
-                        .map(byteBuffer -> DataChunk.create(doFlush, byteBuffer))));
+                        .map(byteBuffer -> DataChunk.create(doFlush(context, byteBuffer), byteBuffer))));
 
         return publisher;
     }
@@ -183,4 +181,18 @@ public boolean enableResponseBuffering() {
         // Jersey should not try to do the buffering
         return false;
     }
+
+    /**
+     * Flush buffer if using SSE or if an empty buffer is received for writing. See
+     * {@link OutputStreamPublisher#flush()}. Manual flushing is required to support
+     * {@link javax.ws.rs.core.StreamingOutput} in MP.
+     *
+     * @param context The container response.
+     * @param byteBuffer The byte buffer to write.
+     * @return Outcome of test.
+     */
+    private static boolean doFlush(ContainerResponse context, ByteBuffer byteBuffer) {
+        return MediaType.SERVER_SENT_EVENTS_TYPE.isCompatible(context.getMediaType())
+                || byteBuffer.hasArray() && byteBuffer.array().length == 0;
+    }
 }

webserver/jersey/src/test/java/io/helidon/webserver/jersey/JerseyExampleResource.java

@@ -18,7 +18,9 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import javax.inject.Inject;
 import javax.inject.Named;
@@ -27,11 +29,13 @@
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
 import javax.ws.rs.QueryParam;
 import javax.ws.rs.WebApplicationException;
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.HttpHeaders;
 import javax.ws.rs.core.Response;
+import javax.ws.rs.core.StreamingOutput;
 import javax.ws.rs.core.UriInfo;
 
 import io.helidon.common.InputStreamHelper;
@@ -212,4 +216,20 @@ public Response deleteNotFound(@Context UriInfo uriInfo) {
     	throw new WebApplicationException(Response.status(404).entity("Not Found").build());
     }
 
+    @GET
+    @Path("/streamingOutput")
+    @Produces("application/stream+json")
+    public StreamingOutput getHelloOutputStream() {
+        return out -> {
+                try {
+                    out.write(("{ value: \"first\" }\n").getBytes(StandardCharsets.UTF_8));
+                    out.flush();
+                    Thread.sleep(500);     // wait before sending next chunk
+                    out.write(("{ value: \"second\" }\n").getBytes(StandardCharsets.UTF_8));
+                    out.flush();
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+        };
+    }
 }

webserver/jersey/src/test/java/io/helidon/webserver/jersey/JerseySupportTest.java

@@ -17,6 +17,7 @@
 package io.helidon.webserver.jersey;
 
 import java.io.BufferedReader;
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.net.URI;
@@ -298,6 +299,23 @@ public void pathEncoding2() {
         doAssert(response, "abc;");
     }
 
+    @Test
+    public void streamingOutput() throws IOException {
+        Response response = webTarget.path("jersey/first/streamingOutput")
+                .request()
+                .get();
+        assertEquals(Response.Status.Family.SUCCESSFUL, response.getStatusInfo().getFamily(),
+                "Unexpected error: " + response.getStatus());
+        try (InputStream is = response.readEntity(InputStream.class)) {
+            byte[] buffer = new byte[32];
+            int n = is.read(buffer);        // should read only first chunk
+            assertEquals(new String(buffer, 0, n), "{ value: \"first\" }\n");
+            while ((n = is.read(buffer)) > 0) {
+                // consume rest of stream
+            }
+        }
+    }
+
     static StringBuilder longData(int bytes) {
         StringBuilder data = new StringBuilder(bytes);
         int i = 0;