New implementation for SSE in webserver (#9224) (#9297)

Provides a new implementation for SSE in the webserver. This implementation does not use the normal output stream to serialize the events to avoid problems with buffering and chunked encoding. Instead, it writes directly to the underlying socket writer and flushes data as needed. As a result, chunked encoding is no longer used for SSE. Several tests have been updated.

Co-authored-by: Santiago Pericas-Geertsen <[email protected]>

common/testing/http-junit5/src/main/java/io/helidon/common/testing/http/junit5/SocketHttpClient.java

@@ -605,6 +605,15 @@ public SocketHttpClient sendChunk(String payload) throws IOException {
         return this;
     }
 
+    /**
+     * Provides access to underlying socket reader.
+     *
+     * @return the reader
+     */
+    public BufferedReader socketReader() {
+        return socketReader;
+    }
+
     /**
      * Override this to send a specific payload.
      *

webclient/sse/src/main/java/io/helidon/webclient/sse/SseSourceHandlerProvider.java

@@ -25,6 +25,7 @@
 import java.time.Duration;
 
 import io.helidon.common.GenericType;
+import io.helidon.common.buffers.DataReader;
 import io.helidon.common.media.type.MediaTypes;
 import io.helidon.http.media.MediaContext;
 import io.helidon.http.sse.SseEvent;
@@ -93,6 +94,9 @@ public <X extends Source<SseEvent>> void handle(X source, HttpClientResponse res
                 }
             }
 
+            source.onClose();
+        } catch (DataReader.InsufficientDataAvailableException e) {
+            // normal SSE termination when connection closed by server
             source.onClose();
         } catch (IOException e) {
             source.onError(e);

webserver/sse/src/main/java/io/helidon/webserver/sse/DataWriterSseSink.java

@@ -0,0 +1,182 @@
+/*
+ * Copyright (c) 2024 Oracle and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.helidon.webserver.sse;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+
+import io.helidon.common.GenericType;
+import io.helidon.common.buffers.BufferData;
+import io.helidon.common.media.type.MediaType;
+import io.helidon.common.media.type.MediaTypes;
+import io.helidon.http.DateTime;
+import io.helidon.http.Header;
+import io.helidon.http.HeaderNames;
+import io.helidon.http.HttpMediaType;
+import io.helidon.http.ServerResponseHeaders;
+import io.helidon.http.Status;
+import io.helidon.http.WritableHeaders;
+import io.helidon.http.media.EntityWriter;
+import io.helidon.http.media.MediaContext;
+import io.helidon.http.sse.SseEvent;
+import io.helidon.webserver.ConnectionContext;
+import io.helidon.webserver.ServerConnectionException;
+import io.helidon.webserver.http.ServerResponse;
+import io.helidon.webserver.http.spi.SinkProviderContext;
+
+import static io.helidon.http.HeaderValues.CONTENT_TYPE_EVENT_STREAM;
+import static io.helidon.http.HeaderValues.create;
+
+/**
+ * Implementation of an SSE sink. Emits {@link SseEvent}s.
+ */
+class DataWriterSseSink implements SseSink {
+
+    /**
+     * Type of SSE event sinks.
+     */
+    public static final GenericType<DataWriterSseSink> TYPE = GenericType.create(DataWriterSseSink.class);
+
+    private static final Header CACHE_NO_CACHE_ONLY = create(HeaderNames.CACHE_CONTROL, "no-cache");
+    private static final byte[] SSE_NL = "\n".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] SSE_ID = "id:".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] SSE_DATA = "data:".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] SSE_EVENT = "event:".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] SSE_COMMENT = ":".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] OK_200 = "HTTP/1.1 200 OK\r\n".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] DATE = "Date: ".getBytes(StandardCharsets.UTF_8);
+    private static final WritableHeaders<?> EMPTY_HEADERS = WritableHeaders.create();
+
+    private final ServerResponse response;
+    private final ConnectionContext ctx;
+    private final MediaContext mediaContext;
+    private final Runnable closeRunnable;
+
+    DataWriterSseSink(SinkProviderContext context) {
+        this.response = context.serverResponse();
+        this.ctx = context.connectionContext();
+        this.mediaContext = ctx.listenerContext().mediaContext();
+        this.closeRunnable = context.closeRunnable();
+        writeStatusAndHeaders();
+    }
+
+    @Override
+    public DataWriterSseSink emit(SseEvent sseEvent) {
+        BufferData bufferData = BufferData.growing(512);
+
+        Optional<String> comment = sseEvent.comment();
+        if (comment.isPresent()) {
+            bufferData.write(SSE_COMMENT);
+            bufferData.write(comment.get().getBytes(StandardCharsets.UTF_8));
+            bufferData.write(SSE_NL);
+        }
+        Optional<String> id = sseEvent.id();
+        if (id.isPresent()) {
+            bufferData.write(SSE_ID);
+            bufferData.write(id.get().getBytes(StandardCharsets.UTF_8));
+            bufferData.write(SSE_NL);
+        }
+        Optional<String> name = sseEvent.name();
+        if (name.isPresent()) {
+            bufferData.write(SSE_EVENT);
+            bufferData.write(name.get().getBytes(StandardCharsets.UTF_8));
+            bufferData.write(SSE_NL);
+        }
+        Object data = sseEvent.data();
+        if (data != null) {
+            bufferData.write(SSE_DATA);
+            byte[] bytes = serializeData(data, sseEvent.mediaType().orElse(MediaTypes.TEXT_PLAIN));
+            bufferData.write(bytes);
+            bufferData.write(SSE_NL);
+        }
+        bufferData.write(SSE_NL);
+
+        // write event to the network
+        ctx.dataWriter().writeNow(bufferData);
+        return this;
+    }
+
+    @Override
+    public void close() {
+        closeRunnable.run();
+        ctx.serverSocket().close();
+    }
+
+    void writeStatusAndHeaders() {
+        ServerResponseHeaders headers = response.headers();
+
+        // verify response has no status or content type
+        HttpMediaType ct = headers.contentType().orElse(null);
+        if (response.status().code() != Status.OK_200.code()
+                || ct != null && !CONTENT_TYPE_EVENT_STREAM.values().equals(ct.mediaType().text())) {
+            throw new IllegalStateException("ServerResponse instance cannot be used to create SseSink");
+        }
+
+        // start writing status line
+        BufferData buffer = BufferData.growing(256);
+        buffer.write(OK_200);
+
+        // serialize a date header if not included
+        if (!headers.contains(HeaderNames.DATE)) {
+            buffer.write(DATE);
+            byte[] dateBytes = DateTime.http1Bytes();
+            buffer.write(dateBytes);
+        }
+
+        // set up and write headers
+        if (ct == null) {
+            headers.add(CONTENT_TYPE_EVENT_STREAM);
+        }
+        headers.set(CACHE_NO_CACHE_ONLY);
+        for (Header header : headers) {
+            header.writeHttp1Header(buffer);
+        }
+
+        // complete heading
+        buffer.write('\r');        // "\r\n" - empty line after headers
+        buffer.write('\n');
+
+        // write response heading to the network
+        ctx.dataWriter().writeNow(buffer);
+    }
+
+    private byte[] serializeData(Object object, MediaType mediaType) {
+        if (object instanceof byte[] bytes) {
+            return bytes;
+        } else if (mediaContext != null) {
+            try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+                if (object instanceof String str && mediaType.equals(MediaTypes.TEXT_PLAIN)) {
+                    EntityWriter<String> writer = mediaContext.writer(GenericType.STRING, EMPTY_HEADERS, EMPTY_HEADERS);
+                    writer.write(GenericType.STRING, str, baos, EMPTY_HEADERS, EMPTY_HEADERS);
+                } else {
+                    GenericType<Object> type = GenericType.create(object);
+                    WritableHeaders<?> resHeaders = WritableHeaders.create();
+                    resHeaders.set(HeaderNames.CONTENT_TYPE, mediaType.text());
+                    EntityWriter<Object> writer = mediaContext.writer(type, EMPTY_HEADERS, resHeaders);
+                    writer.write(type, object, baos, EMPTY_HEADERS, resHeaders);
+                }
+                return baos.toByteArray();
+            } catch (IOException e) {
+                throw new ServerConnectionException("Failed to write SSE event", e);
+
+            }
+        }
+        throw new IllegalStateException("Unable to serialize SSE event without a media context");
+    }
+}

webserver/sse/src/main/java/io/helidon/webserver/sse/OutputStreamSseSink.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright (c) 2023, 2024 Oracle and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.helidon.webserver.sse;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+import java.util.function.BiConsumer;
+
+import io.helidon.common.GenericType;
+import io.helidon.common.media.type.MediaType;
+import io.helidon.common.media.type.MediaTypes;
+import io.helidon.http.HttpMediaType;
+import io.helidon.http.Status;
+import io.helidon.http.sse.SseEvent;
+import io.helidon.webserver.http.ServerResponse;
+
+import static io.helidon.http.HeaderValues.CONTENT_TYPE_EVENT_STREAM;
+
+/**
+ * Deprecated implementation of an SSE sink. Emits {@link SseEvent}s.
+ *
+ * @deprecated Should use {@link io.helidon.webserver.sse.DataWriterSseSink}.
+ */
+@Deprecated(since = "4.1.2", forRemoval = true)
+class OutputStreamSseSink implements SseSink {
+
+    /**
+     * Type of SSE event sinks.
+     */
+    public static final GenericType<DataWriterSseSink> TYPE = GenericType.create(DataWriterSseSink.class);
+
+    private static final byte[] SSE_NL = "\n".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] SSE_ID = "id:".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] SSE_DATA = "data:".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] SSE_EVENT = "event:".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] SSE_COMMENT = ":".getBytes(StandardCharsets.UTF_8);
+
+    private final BiConsumer<Object, MediaType> eventConsumer;
+    private final Runnable closeRunnable;
+    private final OutputStream outputStream;
+
+    OutputStreamSseSink(ServerResponse serverResponse, BiConsumer<Object, MediaType> eventConsumer, Runnable closeRunnable) {
+        // Verify response has no status or content type
+        HttpMediaType ct = serverResponse.headers().contentType().orElse(null);
+        if (serverResponse.status().code() != Status.OK_200.code()
+                || ct != null && !CONTENT_TYPE_EVENT_STREAM.values().equals(ct.mediaType().text())) {
+            throw new IllegalStateException("ServerResponse instance cannot be used to create SseResponse");
+        }
+
+        // Ensure content type set for SSE
+        if (ct == null) {
+            serverResponse.headers().add(CONTENT_TYPE_EVENT_STREAM);
+        }
+
+        this.outputStream = serverResponse.outputStream();
+        this.eventConsumer = eventConsumer;
+        this.closeRunnable = closeRunnable;
+    }
+
+    @Override
+    public OutputStreamSseSink emit(SseEvent sseEvent) {
+        try {
+            Optional<String> comment = sseEvent.comment();
+            if (comment.isPresent()) {
+                outputStream.write(SSE_COMMENT);
+                outputStream.write(comment.get().getBytes(StandardCharsets.UTF_8));
+                outputStream.write(SSE_NL);
+            }
+            Optional<String> id = sseEvent.id();
+            if (id.isPresent()) {
+                outputStream.write(SSE_ID);
+                outputStream.write(id.get().getBytes(StandardCharsets.UTF_8));
+                outputStream.write(SSE_NL);
+            }
+            Optional<String> name = sseEvent.name();
+            if (name.isPresent()) {
+                outputStream.write(SSE_EVENT);
+                outputStream.write(name.get().getBytes(StandardCharsets.UTF_8));
+                outputStream.write(SSE_NL);
+            }
+            Object data = sseEvent.data();
+            if (data != SseEvent.NO_DATA) {
+                outputStream.write(SSE_DATA);
+                eventConsumer.accept(data, sseEvent.mediaType().orElse(MediaTypes.TEXT_PLAIN));
+                outputStream.write(SSE_NL);
+            }
+            outputStream.write(SSE_NL);
+            outputStream.flush();
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+        return this;
+    }
+
+    @Override
+    public void close() {
+        closeRunnable.run();
+    }
+}