Updates SPI implementation to maintain backwards compatibility of old…

…er SSE implementation, which is also kept around. Changes to SseSink to support multiple implementations.

webclient/http1/src/main/java/io/helidon/webclient/http1/Http1CallChainBase.java

@@ -411,31 +411,23 @@ public int read() {
             if (finished) {
                 return -1;
             }
-            try {
-                ensureBuffer(512);
-                if (finished || currentBuffer == null) {
-                    return -1;
-                }
-                return currentBuffer.read();
-            } catch (DataReader.InsufficientDataAvailableException e) {
+            ensureBuffer(512);
+            if (finished || currentBuffer == null) {
                 return -1;
             }
+            return currentBuffer.read();
         }
 
         @Override
         public int read(byte[] b, int off, int len) {
             if (finished) {
                 return -1;
             }
-            try {
-                ensureBuffer(len);
-                if (finished || currentBuffer == null) {
-                    return -1;
-                }
-                return currentBuffer.read(b, off, len);
-            } catch (DataReader.InsufficientDataAvailableException e) {
+            ensureBuffer(len);
+            if (finished || currentBuffer == null) {
                 return -1;
             }
+            return currentBuffer.read(b, off, len);
         }
 
         private void ensureBuffer(int estimate) {

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.
+ */
+public 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 Replaced by {@link io.helidon.webserver.sse.DataWriterSseSink}.
+ */
+@Deprecated(since = "4.1.2", forRemoval = true)
+public 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();
+    }
+}