From 91cb55a0c61ad4ccb2992a50bdebe1ffad7f05a9 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sun, 18 Oct 2026 13:35:44 -0400 Subject: [PATCH 1/2] feat: add Server-Sent Events API Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 4 + .../src/main/java/org/restlet/Request.java | 9 + .../main/java/org/restlet/data/MediaType.java | 3 + .../org/restlet/data/ServerSentEvent.java | 145 ++++++++ .../restlet/engine/netty/NettyServerCall.java | 80 +++-- .../engine/netty/NettyServerHandler.java | 8 +- .../engine/netty/NettyServerHelper.java | 41 +++ .../representation/EventStreamReader.java | 339 ++++++++++++++++++ .../EventStreamRepresentation.java | 219 +++++++++++ .../EventStreamConnectorTestCase.java | 291 +++++++++++++++ .../representation/EventStreamTestCase.java | 151 ++++++++ 11 files changed, 1263 insertions(+), 27 deletions(-) create mode 100644 org.restlet/src/main/java/org/restlet/data/ServerSentEvent.java create mode 100644 org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java create mode 100644 org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java create mode 100644 org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java create mode 100644 org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java diff --git a/changes.md b/changes.md index 12f7e31e66..4552c900c6 100644 --- a/changes.md +++ b/changes.md @@ -62,6 +62,10 @@ Changes log leftover of the Servlet mode, which is no longer supported. `Protocol.valueOf("war")` now returns a new, unknown protocol. Use CLAP or FILE references instead. - Enhancements + - Added the `EventStreamRepresentation`, `ServerSentEvent` and `EventStreamReader` APIs for + streaming and reading Server-Sent Events over the HTTP connectors, including optional + comment heartbeats and reconnects honoring `retry:` and `Last-Event-ID`. The + `server.maxEventStreams` parameter limits concurrent event streams. - Added `Response.getProtocol()` and `setProtocol()`: the protocol, with its version, used by the server to reply (for example HTTP/1.1, HTTP/2.0 or HTTP/3.0), set by the HTTP client connector. When unknown, the protocol of the request is returned. diff --git a/org.restlet/src/main/java/org/restlet/Request.java b/org.restlet/src/main/java/org/restlet/Request.java index 53a7ef36b6..cb5eeee2fe 100644 --- a/org.restlet/src/main/java/org/restlet/Request.java +++ b/org.restlet/src/main/java/org/restlet/Request.java @@ -469,6 +469,15 @@ public Series
getTrailers() { return result; } + /** + * Returns the last event ID sent by an SSE client, if present. + * + * @return The value of the {@code Last-Event-ID} header, or null. + */ + public String getLastEventId() { + return getHeaders().getFirstValue("Last-Event-ID", true); + } + /** * Returns the host reference. This may be different from the resourceRef's host, for example, * for URNs and other URIs that don't contain host information.
diff --git a/org.restlet/src/main/java/org/restlet/data/MediaType.java b/org.restlet/src/main/java/org/restlet/data/MediaType.java index 67e5db402e..d870a7d7c3 100644 --- a/org.restlet/src/main/java/org/restlet/data/MediaType.java +++ b/org.restlet/src/main/java/org/restlet/data/MediaType.java @@ -454,6 +454,9 @@ public final class MediaType extends Metadata { public static final MediaType TEXT_DAT = register("text/x-fixed-field", "Fixed-width Values"); + public static final MediaType TEXT_EVENT_STREAM = + register("text/event-stream", "Server-sent events"); + public static final MediaType TEXT_HTML = register("text/html", "HTML document"); public static final MediaType TEXT_J2ME_APP_DESCRIPTOR = diff --git a/org.restlet/src/main/java/org/restlet/data/ServerSentEvent.java b/org.restlet/src/main/java/org/restlet/data/ServerSentEvent.java new file mode 100644 index 0000000000..603db36682 --- /dev/null +++ b/org.restlet/src/main/java/org/restlet/data/ServerSentEvent.java @@ -0,0 +1,145 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.data; + +/** An immutable Server-Sent Event. */ +public final class ServerSentEvent { + + private final String id; + + private final String event; + + private final String data; + + private final Long retry; + + private final String comment; + + private ServerSentEvent(Builder builder) { + this.id = builder.id; + this.event = builder.event; + this.data = builder.data; + this.retry = builder.retry; + this.comment = builder.comment; + } + + /** Returns a builder for an event. */ + public static Builder builder() { + return new Builder(); + } + + /** Returns the event ID, or {@code null} when absent. */ + public String getId() { + return id; + } + + /** Returns the event type, or {@code null} when absent. */ + public String getEvent() { + return event; + } + + /** Returns the event data, or {@code null} when absent. */ + public String getData() { + return data; + } + + /** Returns the reconnection delay in milliseconds, or {@code null} when absent. */ + public Long getRetry() { + return retry; + } + + /** Returns the comment, or {@code null} when absent. */ + public String getComment() { + return comment; + } + + /** Builds an immutable {@link ServerSentEvent}. */ + public static final class Builder { + + private String id; + + private String event; + + private String data; + + private Long retry; + + private String comment; + + private Builder() {} + + /** + * Sets the event ID. + * + * @param id The ID; it must not contain a NUL, carriage return or line feed. + * @return This builder. + */ + public Builder id(String id) { + if (id != null + && (id.indexOf('\0') >= 0 || id.indexOf('\r') >= 0 || id.indexOf('\n') >= 0)) { + throw new IllegalArgumentException("An SSE id cannot contain NUL or a line break"); + } + this.id = id; + return this; + } + + /** + * Sets the event type. + * + * @param event The event type; it must not contain a line break. + * @return This builder. + */ + public Builder event(String event) { + rejectLineBreak(event, "event"); + this.event = event; + return this; + } + + /** Sets the event data. Line breaks are encoded as multiple {@code data:} fields. */ + public Builder data(String data) { + this.data = data; + return this; + } + + /** + * Sets the reconnection delay. + * + * @param retry The delay in milliseconds; must be non-negative. + * @return This builder. + */ + public Builder retry(Long retry) { + if (retry != null && retry < 0) { + throw new IllegalArgumentException("An SSE retry delay cannot be negative"); + } + this.retry = retry; + return this; + } + + /** Sets a comment, commonly used as a heartbeat. Line breaks create multiple comments. */ + public Builder comment(String comment) { + this.comment = comment; + return this; + } + + /** Builds the event. At least one field must be set. */ + public ServerSentEvent build() { + if (id == null && event == null && data == null && retry == null && comment == null) { + throw new IllegalStateException("An SSE must contain at least one field"); + } + return new ServerSentEvent(this); + } + + private static void rejectLineBreak(String value, String field) { + if (value != null && (value.indexOf('\r') >= 0 || value.indexOf('\n') >= 0)) { + throw new IllegalArgumentException( + "An SSE " + field + " field cannot contain a line break"); + } + } + } +} diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java index f67640d483..f4b0623a5e 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java @@ -50,6 +50,7 @@ import org.restlet.Context; import org.restlet.Response; import org.restlet.data.Header; +import org.restlet.data.MediaType; import org.restlet.data.Method; import org.restlet.data.Protocol; import org.restlet.data.Status; @@ -125,6 +126,9 @@ class NettyServerCall { /** The response body stream, if any. */ private volatile ChannelOutputStream responseBody; + /** Whether this call holds an open SSE stream permit. */ + private boolean eventStreamPermit; + /** The "Alt-Svc" header value advertising HTTP/3, or null. */ private volatile String altSvc; @@ -354,6 +358,15 @@ private InputStream peek(InputStream stream) { void commit(Response response) { try { responseTrailers = response.getTrailers(); + if (isEventStream(response.getEntity())) { + if (helper.tryAcquireEventStream()) { + eventStreamPermit = true; + channel.attr(NettyServerHandler.EVENT_STREAM_ACTIVE).set(true); + } else { + response.setStatus(Status.SERVER_ERROR_SERVICE_UNAVAILABLE); + response.setEntity(null); + } + } prepareEntity(response); addResponseHeaders(response); sendResponse(response); @@ -367,6 +380,12 @@ void commit(Response response) { } } + private static boolean isEventStream(Representation entity) { + return entity != null + && entity.getMediaType() != null + && MediaType.TEXT_EVENT_STREAM.includes(entity.getMediaType()); + } + /** Adds the entity headers and drops the entity when the status or method forbids it. */ private void prepareEntity(Response response) { Status status = response.getStatus(); @@ -604,37 +623,46 @@ void complete() { completed = true; try { - if (!responseHeadWritten) { - writeHead(true); + try { + if (!responseHeadWritten) { + writeHead(true); + } + if (responseBody != null) { + responseBody.close(); + } + } catch (IOException e) { + getLogger().log(Level.FINE, "Unable to complete the response: {0}", e.getMessage()); } - if (responseBody != null) { - responseBody.close(); + + if (!channel.isActive()) { + requestBody.close(); + return; } - } catch (IOException e) { - getLogger().log(Level.FINE, "Unable to complete the response: {0}", e.getMessage()); - } - if (!channel.isActive()) { + boolean keepAlive = HttpUtil.isKeepAlive(request) && requestBody.isFullyConsumed(); + LastHttpContent lastContent = LastHttpContent.EMPTY_LAST_CONTENT; + if (responseTrailersEnabled) { + DefaultLastHttpContent withTrailers = + new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER); + copyHeaders(responseTrailers, withTrailers.trailingHeaders(), kind.isStream()); + lastContent = withTrailers; + } + ChannelFuture future = channel.writeAndFlush(lastContent); requestBody.close(); - return; - } - boolean keepAlive = HttpUtil.isKeepAlive(request) && requestBody.isFullyConsumed(); - LastHttpContent lastContent = LastHttpContent.EMPTY_LAST_CONTENT; - if (responseTrailersEnabled) { - DefaultLastHttpContent withTrailers = new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER); - copyHeaders(responseTrailers, withTrailers.trailingHeaders(), kind.isStream()); - lastContent = withTrailers; - } - ChannelFuture future = channel.writeAndFlush(lastContent); - requestBody.close(); - - if (!kind.isStream()) { - if (keepAlive) { - // Ready for the next request on this connection. - future.addListener(f -> channel.read()); - } else { - future.addListener(ChannelFutureListener.CLOSE); + if (!kind.isStream()) { + if (keepAlive) { + // Ready for the next request on this connection. + future.addListener(f -> channel.read()); + } else { + future.addListener(ChannelFutureListener.CLOSE); + } + } + } finally { + if (eventStreamPermit) { + channel.attr(NettyServerHandler.EVENT_STREAM_ACTIVE).set(false); + helper.releaseEventStream(); + eventStreamPermit = false; } } } diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java index 847e09aedd..3cec34a87a 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java @@ -21,6 +21,7 @@ import io.netty.handler.codec.http.HttpVersion; import io.netty.handler.codec.http.LastHttpContent; import io.netty.handler.timeout.IdleStateEvent; +import io.netty.util.AttributeKey; import io.netty.util.ReferenceCountUtil; import java.io.IOException; import java.nio.channels.ClosedChannelException; @@ -38,6 +39,9 @@ */ class NettyServerHandler extends ChannelInboundHandlerAdapter { + static final AttributeKey EVENT_STREAM_ACTIVE = + AttributeKey.valueOf("restlet.eventStreamActive"); + private final NettyServerHelper helper; private final boolean confidential; @@ -192,7 +196,9 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { - ctx.close(); + if (!Boolean.TRUE.equals(ctx.channel().attr(EVENT_STREAM_ACTIVE).get())) { + ctx.close(); + } } else { super.userEventTriggered(ctx, evt); } diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java index 382b2b4264..f9f32d5a66 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java @@ -110,6 +110,13 @@ * 0, there is no limit. * * + * server.maxEventStreams + * int + * 0 + * Maximum number of open Server-Sent Event streams; extra streams receive a 503 response. + * When 0, there is no limit. + * + * * eventLoop.bossThreads * int * 1 @@ -201,6 +208,9 @@ public abstract class NettyServerHelper extends ServerHelper { /** Limits concurrent calls, or null when unlimited. */ private volatile Semaphore callPermits; + /** Limits concurrent SSE streams, or null when unlimited. */ + private volatile Semaphore eventStreamPermits; + /** The number of calls in progress. */ private final AtomicInteger activeCalls = new AtomicInteger(); @@ -498,6 +508,20 @@ void releaseCall() { } } + /** Tries to acquire a permit for an SSE response. */ + boolean tryAcquireEventStream() { + Semaphore permits = eventStreamPermits; + return permits == null || permits.tryAcquire(); + } + + /** Releases an acquired SSE response permit. */ + void releaseEventStream() { + Semaphore permits = eventStreamPermits; + if (permits != null) { + permits.release(); + } + } + /** * Returns the executor running Restlet calls. * @@ -520,6 +544,8 @@ public synchronized void start() throws Exception { int maxCalls = getServerMaxConcurrentCalls(); callPermits = (maxCalls > 0) ? new Semaphore(maxCalls) : null; + int maxEventStreams = getServerMaxEventStreams(); + eventStreamPermits = (maxEventStreams > 0) ? new Semaphore(maxEventStreams) : null; callExecutor = CallExecutors.create( "restlet-server", @@ -819,6 +845,21 @@ public int getServerMaxConcurrentCalls() { getHelpedParameters().getFirstValue("server.maxConcurrentCalls", "0")); } + /** + * Maximum number of open Server-Sent Event streams. Defaults to 0 (unlimited). + * + * @return Maximum number of open SSE streams. + */ + public int getServerMaxEventStreams() { + int result = + Integer.parseInt( + getHelpedParameters().getFirstValue("server.maxEventStreams", "0")); + if (result < 0) { + throw new IllegalArgumentException("'server.maxEventStreams' cannot be negative"); + } + return result; + } + /** * Number of event loop threads accepting connections. Defaults to 1. * diff --git a/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java b/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java new file mode 100644 index 0000000000..556c0ee94d --- /dev/null +++ b/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java @@ -0,0 +1,339 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.representation; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.Reader; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.util.Iterator; +import java.util.NoSuchElementException; +import java.util.Objects; +import org.restlet.data.ServerSentEvent; + +/** + * A blocking reader of Server-Sent Events. When created with an {@link EventStreamSource}, it + * reconnects after EOF, honoring the server's {@code retry:} field and passing the latest event ID + * to the source for use in a {@code Last-Event-ID} request header. + */ +public final class EventStreamReader implements Iterator, AutoCloseable { + + private static final long DEFAULT_RETRY_MS = 3000; + + private final EventStreamSource source; + + private final Object reconnectLock = new Object(); + + private BufferedReader reader; + + private ServerSentEvent next; + + private volatile String lastEventId = ""; + + private volatile long retryDelayMs = DEFAULT_RETRY_MS; + + private boolean beginning = true; + + private boolean currentEof; + + private boolean sourceStopped; + + private volatile boolean closed; + + /** + * Creates a one-shot reader. + * + * @param inputStream The UTF-8 encoded event stream. + */ + public EventStreamReader(InputStream inputStream) { + this( + new InputStreamReader( + Objects.requireNonNull(inputStream, "inputStream"), StandardCharsets.UTF_8), + null); + } + + /** + * Creates a one-shot reader. + * + * @param reader The event stream reader. + */ + public EventStreamReader(Reader reader) { + this(reader, null); + } + + /** + * Creates a reconnecting reader. The source is called initially and after each EOF. It receives + * the most recently parsed event ID; return null to stop reconnecting. + * + * @param source Opens a stream, setting {@code Last-Event-ID} from the supplied value when it + * is non-empty. + * @throws IOException If the initial connection fails. + */ + public EventStreamReader(EventStreamSource source) throws IOException { + this.source = Objects.requireNonNull(source, "source"); + InputStream inputStream = source.open(""); + if (inputStream == null) { + reader = new BufferedReader(new java.io.StringReader("")); + sourceStopped = true; + } else { + reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)); + } + } + + private EventStreamReader(InputStream inputStream, EventStreamSource source) { + this( + new InputStreamReader( + Objects.requireNonNull(inputStream, "inputStream"), StandardCharsets.UTF_8), + source); + } + + private EventStreamReader(Reader reader, EventStreamSource source) { + this.reader = new BufferedReader(Objects.requireNonNull(reader, "reader")); + this.source = source; + } + + /** Returns the last event ID seen, or the empty string before any ID was read. */ + public String getLastEventId() { + return lastEventId; + } + + /** Returns the most recently received reconnection delay in milliseconds. */ + public long getRetryDelayMs() { + return retryDelayMs; + } + + /** + * Reads the next event, blocking as needed. + * + * @return The next event, or null at EOF for a one-shot stream or when a source stops. + * @throws IOException If reading or reconnecting fails. + */ + public ServerSentEvent readEvent() throws IOException { + if (closed || sourceStopped) { + return null; + } + while (!closed) { + ServerSentEvent event = readCurrentEvent(); + if (event != null) { + return event; + } + if (!currentEof || source == null || sourceStopped || !reconnect()) { + return null; + } + } + return null; + } + + @Override + public boolean hasNext() { + if (next == null && !closed) { + try { + next = readEvent(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + return next != null; + } + + @Override + public ServerSentEvent next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + ServerSentEvent result = next; + next = null; + return result; + } + + @Override + public void close() throws IOException { + closed = true; + synchronized (reconnectLock) { + reconnectLock.notifyAll(); + } + reader.close(); + } + + private ServerSentEvent readCurrentEvent() throws IOException { + String id = null; + String event = null; + Long retry = null; + StringBuilder data = new StringBuilder(); + StringBuilder comment = new StringBuilder(); + boolean hasData = false; + boolean hasComment = false; + + while (!closed) { + String line = reader.readLine(); + if (line == null) { + currentEof = true; + return hasData || hasComment + ? buildEvent(id, event, retry, data, comment, hasData, hasComment) + : null; + } + if (beginning) { + beginning = false; + if (!line.isEmpty() && line.charAt(0) == '\uFEFF') { + line = line.substring(1); + } + } + if (line.isEmpty()) { + if (hasData || hasComment) { + return buildEvent(id, event, retry, data, comment, hasData, hasComment); + } + id = null; + event = null; + retry = null; + continue; + } + if (line.charAt(0) == ':') { + if (hasComment) { + comment.append('\n'); + } + String value = line.substring(1); + if (value.startsWith(" ")) { + value = value.substring(1); + } + comment.append(value); + hasComment = true; + continue; + } + + int separator = line.indexOf(':'); + String field = separator < 0 ? line : line.substring(0, separator); + String value = separator < 0 ? "" : line.substring(separator + 1); + if (value.startsWith(" ")) { + value = value.substring(1); + } + switch (field) { + case "data" -> { + if (hasData) { + data.append('\n'); + } + data.append(value); + hasData = true; + } + case "event" -> event = value; + case "id" -> { + if (value.indexOf('\0') < 0) { + id = value; + lastEventId = value; + } + } + case "retry" -> { + if (isDigits(value)) { + try { + retry = Long.parseLong(value); + retryDelayMs = retry; + } catch (NumberFormatException _) { + // An out-of-range retry value is ignored by the SSE parsing rules. + } + } + } + default -> { + // Unknown fields are ignored by the SSE protocol. + } + } + } + return null; + } + + private ServerSentEvent buildEvent( + String id, + String event, + Long retry, + StringBuilder data, + StringBuilder comment, + boolean hasData, + boolean hasComment) { + ServerSentEvent.Builder builder = ServerSentEvent.builder(); + if (id != null) { + builder.id(id); + } + if (event != null) { + builder.event(event); + } + if (retry != null) { + builder.retry(retry); + } + if (hasData) { + builder.data(data.toString()); + } + if (hasComment) { + builder.comment(comment.toString()); + } + return builder.build(); + } + + private boolean reconnect() throws IOException { + synchronized (reconnectLock) { + if (closed) { + return false; + } + if (retryDelayMs > 0) { + try { + reconnectLock.wait(retryDelayMs); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException( + "Interrupted while waiting to reconnect to the event stream", e); + } + } + if (closed) { + return false; + } + } + reader.close(); + InputStream inputStream; + try { + inputStream = source.open(lastEventId); + } catch (IOException e) { + sourceStopped = true; + throw e; + } + if (inputStream == null) { + sourceStopped = true; + return false; + } + reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)); + beginning = true; + currentEof = false; + return true; + } + + private static boolean isDigits(String value) { + if (value.isEmpty()) { + return false; + } + for (int i = 0; i < value.length(); i++) { + if (value.charAt(i) < '0' || value.charAt(i) > '9') { + return false; + } + } + return true; + } + + /** Opens a new stream using the latest event ID. */ + @FunctionalInterface + public interface EventStreamSource { + + /** + * Opens an event stream. + * + * @param lastEventId The last parsed event ID, or an empty string before any ID was read. + * @return The next UTF-8 stream, or null to stop reconnecting. + * @throws IOException If opening the stream fails. + */ + InputStream open(String lastEventId) throws IOException; + } +} diff --git a/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java b/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java new file mode 100644 index 0000000000..aee7b7a8d7 --- /dev/null +++ b/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java @@ -0,0 +1,219 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.representation; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import org.restlet.data.CharacterSet; +import org.restlet.data.MediaType; +import org.restlet.data.ServerSentEvent; + +/** + * A streaming {@code text/event-stream} representation. The writer runs on the connector's call + * thread; each event is flushed before {@link EventSink#send(ServerSentEvent)} returns. + */ +public final class EventStreamRepresentation extends OutputRepresentation { + + private final EventWriter writer; + + private final Duration heartbeatInterval; + + /** + * Creates an event stream. + * + * @param writer Writes events to the provided sink until the stream is complete. + */ + public EventStreamRepresentation(EventWriter writer) { + this(writer, null); + } + + /** + * Creates an event stream with optional automatic comment heartbeats. + * + * @param writer Writes events to the provided sink until the stream is complete. + * @param heartbeatInterval The heartbeat interval, or null to disable automatic heartbeats. + */ + public EventStreamRepresentation(EventWriter writer, Duration heartbeatInterval) { + super(MediaType.TEXT_EVENT_STREAM); + this.writer = Objects.requireNonNull(writer, "writer"); + if (heartbeatInterval != null + && (heartbeatInterval.isZero() || heartbeatInterval.isNegative())) { + throw new IllegalArgumentException("The heartbeat interval must be positive"); + } + this.heartbeatInterval = heartbeatInterval; + setCharacterSet(CharacterSet.UTF_8); + setTransient(true); + } + + @Override + public void write(OutputStream outputStream) throws IOException { + try (EventSink sink = new EventSink(outputStream)) { + if (heartbeatInterval == null) { + writer.write(sink); + return; + } + + AtomicBoolean stopped = new AtomicBoolean(); + AtomicReference heartbeatFailure = new AtomicReference<>(); + Thread writerThread = Thread.currentThread(); + Thread heartbeatThread = + Thread.ofVirtual() + .name("restlet-sse-heartbeat") + .start( + () -> + writeHeartbeats( + sink, stopped, heartbeatFailure, writerThread)); + Throwable writerFailure = null; + try { + writer.write(sink); + } catch (IOException | RuntimeException | Error e) { + writerFailure = e; + throw e; + } finally { + stopped.set(true); + heartbeatThread.interrupt(); + try { + heartbeatThread.join(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + IOException interrupted = + new IOException("Interrupted while stopping SSE heartbeats", e); + if (writerFailure != null) { + writerFailure.addSuppressed(interrupted); + } else { + throw interrupted; + } + } + } + IOException failure = heartbeatFailure.get(); + if (failure != null) { + throw failure; + } + } + } + + private void writeHeartbeats( + EventSink sink, + AtomicBoolean stopped, + AtomicReference failure, + Thread writerThread) { + while (!stopped.get()) { + try { + Thread.sleep(heartbeatInterval); + } catch (InterruptedException e) { + if (!stopped.get()) { + Thread.currentThread().interrupt(); + } + return; + } + if (stopped.get()) { + return; + } + try { + sink.send(ServerSentEvent.builder().comment("").build()); + } catch (IOException e) { + failure.set(e); + writerThread.interrupt(); + return; + } + } + } + + /** Writes events for the lifetime of the response. */ + @FunctionalInterface + public interface EventWriter { + + /** + * Writes events until the stream is complete. + * + * @param sink The event sink. + * @throws IOException If writing an event fails. + */ + void write(EventSink sink) throws IOException; + } + + /** A blocking sink for events in an {@link EventStreamRepresentation}. */ + public static final class EventSink implements AutoCloseable { + + private final OutputStream outputStream; + + private boolean closed; + + private EventSink(OutputStream outputStream) { + this.outputStream = Objects.requireNonNull(outputStream, "outputStream"); + } + + /** + * Encodes and flushes one event. + * + * @param event The event to send. + * @throws IOException If the response stream has closed or writing fails. + */ + public synchronized void send(ServerSentEvent event) throws IOException { + Objects.requireNonNull(event, "event"); + if (closed) { + throw new IOException("Event sink is closed"); + } + + writeField("id", event.getId()); + writeField("event", event.getEvent()); + writeMultilineField("data", event.getData()); + if (event.getRetry() != null) { + writeField("retry", Long.toString(event.getRetry())); + } + writeMultilineComment(event.getComment()); + write("\n"); + outputStream.flush(); + } + + /** + * Closes this sink without closing the connector-owned response stream. + * + * @throws IOException If flushing fails. + */ + @Override + public synchronized void close() throws IOException { + if (!closed) { + closed = true; + outputStream.flush(); + } + } + + private void writeField(String name, String value) throws IOException { + if (value != null) { + write(name + ":" + (value.isEmpty() ? "" : " " + value) + "\n"); + } + } + + private void writeMultilineField(String name, String value) throws IOException { + if (value != null) { + for (String line : value.split("\\r\\n|\\r|\\n", -1)) { + write(name + ":" + (line.isEmpty() ? "" : " " + line) + "\n"); + } + } + } + + private void writeMultilineComment(String value) throws IOException { + if (value != null) { + for (String line : value.split("\\r\\n|\\r|\\n", -1)) { + write(":" + (line.isEmpty() ? "" : " " + line) + "\n"); + } + } + } + + private void write(String value) throws IOException { + outputStream.write(value.getBytes(StandardCharsets.UTF_8)); + } + } +} diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java new file mode 100644 index 0000000000..ceba9302a6 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java @@ -0,0 +1,291 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.engine.connector; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.InputStream; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.restlet.Client; +import org.restlet.Context; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.Restlet; +import org.restlet.Server; +import org.restlet.data.MediaType; +import org.restlet.data.Method; +import org.restlet.data.Protocol; +import org.restlet.data.ServerSentEvent; +import org.restlet.data.Status; +import org.restlet.engine.Engine; +import org.restlet.representation.EventStreamReader; +import org.restlet.representation.EventStreamRepresentation; + +class EventStreamConnectorTestCase { + + private Server server; + + private Client client; + + @BeforeEach + void setUp() { + Engine.clearThreadLocalVariables(); + Engine.register(false); + Engine.getInstance().getRegisteredServers().add(new HttpServerHelper(null)); + Engine.getInstance().getRegisteredClients().add(new HttpClientHelper(null)); + } + + @AfterEach + void tearDown() throws Exception { + if (client != null) { + client.stop(); + } + if (server != null) { + server.stop(); + } + Engine.clearThreadLocalVariables(); + } + + @Test + void flushesEventsBeforeTheResourceCompletesOverHttp11() throws Exception { + assertStreamsBeforeCompletion(false); + } + + @Test + void flushesEventsBeforeTheResourceCompletesOverHttp2() throws Exception { + assertStreamsBeforeCompletion(true); + } + + @Test + void enforcesTheConfiguredOpenStreamLimit() throws Exception { + CountDownLatch firstStreamStarted = new CountDownLatch(1); + CountDownLatch finishFirstStream = new CountDownLatch(1); + Restlet restlet = + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity( + new EventStreamRepresentation( + sink -> { + sink.send( + ServerSentEvent.builder() + .data("started") + .build()); + firstStreamStarted.countDown(); + try { + if (!finishFirstStream.await(5, TimeUnit.SECONDS)) { + throw new java.io.IOException( + "Timed out waiting to finish the event stream"); + } + sink.send( + ServerSentEvent.builder() + .data("done") + .build()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new java.io.IOException( + "Interrupted while writing the event stream", + e); + } + })); + } + }; + Context serverContext = new Context(); + serverContext.getParameters().add("server.maxEventStreams", "1"); + server = new Server(serverContext, Protocol.HTTP, 0, restlet); + HttpServerHelper helper = + (HttpServerHelper) serverContext.getAttributes().get("org.restlet.engine.helper"); + assertEquals(1, helper.getServerMaxEventStreams()); + server.start(); + client = new Client(new Context(), Protocol.HTTP); + client.start(); + Client otherClient = new Client(new Context(), Protocol.HTTP); + otherClient.start(); + + try { + Response first = + client.handle( + new Request( + Method.GET, + "http://localhost:" + server.getActualPort() + "/first")); + assertTrue(firstStreamStarted.await(5, TimeUnit.SECONDS)); + Response second = + otherClient.handle( + new Request( + Method.GET, + "http://localhost:" + server.getActualPort() + "/second")); + assertEquals(Status.SERVER_ERROR_SERVICE_UNAVAILABLE, second.getStatus()); + + finishFirstStream.countDown(); + try (EventStreamReader reader = new EventStreamReader(first.getEntity().getStream())) { + assertEquals("started", reader.readEvent().getData()); + assertEquals("done", reader.readEvent().getData()); + assertNull(reader.readEvent()); + } + + Response third = + otherClient.handle( + new Request( + Method.GET, + "http://localhost:" + server.getActualPort() + "/third")); + assertEquals(Status.SUCCESS_OK, third.getStatus()); + try (EventStreamReader reader = new EventStreamReader(third.getEntity().getStream())) { + assertEquals("started", reader.readEvent().getData()); + assertEquals("done", reader.readEvent().getData()); + assertNull(reader.readEvent()); + } + } finally { + finishFirstStream.countDown(); + otherClient.stop(); + } + } + + @Test + void requestExposesLastEventIdHeader() throws Exception { + Restlet restlet = + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity(request.getLastEventId(), MediaType.TEXT_PLAIN); + } + }; + server = new Server(new Context(), Protocol.HTTP, 0, restlet); + server.start(); + client = new Client(new Context(), Protocol.HTTP); + client.start(); + Request request = + new Request(Method.GET, "http://localhost:" + server.getActualPort() + "/"); + request.getHeaders().add("Last-Event-ID", "resume-here"); + + Response response = client.handle(request); + + assertEquals("resume-here", response.getEntityAsText()); + } + + @Test + void openEventStreamIsNotClosedByTheConnectorIdleTimeout() throws Exception { + Restlet restlet = + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity( + new EventStreamRepresentation( + sink -> { + try { + Thread.sleep(350); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new java.io.IOException( + "Interrupted while awaiting an event", e); + } + sink.send( + ServerSentEvent.builder() + .data("after idle") + .build()); + })); + } + }; + Context serverContext = new Context(); + serverContext.getParameters().add("connector.idleTimeout", "100"); + server = new Server(serverContext, Protocol.HTTP, 0, restlet); + server.start(); + client = new Client(new Context(), Protocol.HTTP); + client.start(); + + Response response = + client.handle( + new Request( + Method.GET, "http://localhost:" + server.getActualPort() + "/")); + try (EventStreamReader reader = new EventStreamReader(response.getEntity().getStream())) { + assertEquals("after idle", reader.readEvent().getData()); + assertNull(reader.readEvent()); + } + } + + private void assertStreamsBeforeCompletion(boolean http2) throws Exception { + CountDownLatch firstEventSent = new CountDownLatch(1); + CountDownLatch continueWriting = new CountDownLatch(1); + CountDownLatch writerFinished = new CountDownLatch(1); + Restlet restlet = + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity( + new EventStreamRepresentation( + sink -> { + try { + sink.send( + ServerSentEvent.builder() + .id("one") + .data("first") + .build()); + firstEventSent.countDown(); + if (!continueWriting.await(5, TimeUnit.SECONDS)) { + throw new java.io.IOException( + "Timed out waiting to continue the event stream"); + } + sink.send( + ServerSentEvent.builder() + .data("second") + .build()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new java.io.IOException( + "Interrupted while writing the event stream", + e); + } finally { + writerFinished.countDown(); + } + })); + } + }; + + Context serverContext = new Context(); + Context clientContext = new Context(); + if (http2) { + serverContext.getParameters().add("http.transport.protocol", "HTTP2"); + clientContext.getParameters().add("httpClientTransportMode", "HTTP2"); + } + server = new Server(serverContext, Protocol.HTTP, 0, restlet); + server.start(); + client = new Client(clientContext, Protocol.HTTP); + client.start(); + + Response response = + client.handle( + new Request( + Method.GET, "http://localhost:" + server.getActualPort() + "/")); + assertEquals(http2 ? "2.0" : "1.1", response.getProtocol().getVersion()); + assertTrue(MediaType.TEXT_EVENT_STREAM.includes(response.getEntity().getMediaType())); + InputStream body = response.getEntity().getStream(); + try (EventStreamReader reader = new EventStreamReader(body)) { + assertTrue(firstEventSent.await(5, TimeUnit.SECONDS)); + ServerSentEvent first = reader.readEvent(); + assertEquals("one", first.getId()); + assertEquals("first", first.getData()); + assertFalse( + writerFinished.await(100, TimeUnit.MILLISECONDS), + "The first event must arrive while the response is still open"); + + continueWriting.countDown(); + assertEquals("second", reader.readEvent().getData()); + assertNull(reader.readEvent()); + } finally { + continueWriting.countDown(); + } + assertTrue(writerFinished.await(5, TimeUnit.SECONDS)); + } +} diff --git a/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java b/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java new file mode 100644 index 0000000000..44a1fac10d --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java @@ -0,0 +1,151 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.representation; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.StringReader; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import org.junit.jupiter.api.Test; +import org.restlet.data.MediaType; +import org.restlet.data.ServerSentEvent; + +class EventStreamTestCase { + + @Test + void serializesUtf8FieldsAndMultilineValues() throws IOException { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + EventStreamRepresentation representation = + new EventStreamRepresentation( + sink -> { + sink.send( + ServerSentEvent.builder() + .id("42") + .event("message") + .data(" héllo\nworld ") + .retry(250L) + .comment("heartbeat") + .build()); + }); + + representation.write(output); + + assertEquals(MediaType.TEXT_EVENT_STREAM, representation.getMediaType()); + assertEquals(StandardCharsets.UTF_8.name(), representation.getCharacterSet().getName()); + assertEquals( + "id: 42\n" + + "event: message\n" + + "data: héllo\n" + + "data: world \n" + + "retry: 250\n" + + ": heartbeat\n" + + "\n", + output.toString(StandardCharsets.UTF_8)); + } + + @Test + void emitsConfiguredCommentHeartbeatsWhileTheWriterIsIdle() throws IOException { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + EventStreamRepresentation representation = + new EventStreamRepresentation( + sink -> { + try { + Thread.sleep(Duration.ofMillis(60)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while awaiting an event", e); + } + sink.send(ServerSentEvent.builder().data("ready").build()); + }, + Duration.ofMillis(10)); + + representation.write(output); + + String content = output.toString(StandardCharsets.UTF_8); + assertTrue(content.contains(":\n\n"), content); + assertTrue(content.contains("data: ready\n\n"), content); + } + + @Test + void readsFieldsCommentsAndReconnectState() throws IOException { + EventStreamReader reader = + new EventStreamReader( + new StringReader( + "\uFEFFid:cursor-1\r\n" + + "event:update\r\n" + + "data:first\r\n" + + "data:second\r\n" + + "retry:5\r\n" + + "\r\n" + + ":keepalive\n\n" + + "data:\n\n")); + + ServerSentEvent event = reader.readEvent(); + assertEquals("cursor-1", event.getId()); + assertEquals("update", event.getEvent()); + assertEquals("first\nsecond", event.getData()); + assertEquals(5L, event.getRetry()); + assertEquals(5L, reader.getRetryDelayMs()); + assertEquals("cursor-1", reader.getLastEventId()); + + ServerSentEvent heartbeat = reader.readEvent(); + assertEquals("keepalive", heartbeat.getComment()); + assertNull(heartbeat.getData()); + + assertEquals("", reader.readEvent().getData()); + assertNull(reader.readEvent()); + reader.close(); + } + + @Test + void reconnectsAfterRetryAndPassesTheLastEventId() throws IOException { + List ids = new ArrayList<>(); + int[] attempts = {0}; + EventStreamReader reader = + new EventStreamReader( + lastEventId -> { + ids.add(lastEventId); + return switch (attempts[0]++) { + case 0 -> stream("id:next\nretry:0\ndata:first\n\n"); + case 1 -> stream("data:second\n\n"); + default -> null; + }; + }); + + assertEquals("first", reader.readEvent().getData()); + assertEquals("second", reader.readEvent().getData()); + assertNull(reader.readEvent()); + assertNull(reader.readEvent()); + assertEquals(List.of("", "next", "next"), ids); + reader.close(); + } + + @Test + void validatesEventFields() { + assertThrows(IllegalArgumentException.class, () -> ServerSentEvent.builder().id("bad\nid")); + assertThrows( + IllegalArgumentException.class, + () -> ServerSentEvent.builder().event("bad\nevent")); + assertThrows(IllegalArgumentException.class, () -> ServerSentEvent.builder().retry(-1L)); + } + + private static InputStream stream(String content) { + return new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8)); + } +} From 46bfaa87049f2528be623ea906d51cff8b61e271 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sun, 18 Oct 2026 22:45:01 -0400 Subject: [PATCH 2/2] fix: address SonarCloud issues in SSE API Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../representation/EventStreamReader.java | 257 ++++++++++-------- .../EventStreamRepresentation.java | 44 +-- .../EventStreamConnectorTestCase.java | 6 +- .../representation/EventStreamTestCase.java | 53 ++-- 4 files changed, 204 insertions(+), 156 deletions(-) diff --git a/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java b/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java index 556c0ee94d..177d9de104 100644 --- a/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java +++ b/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java @@ -89,13 +89,6 @@ public EventStreamReader(EventStreamSource source) throws IOException { } } - private EventStreamReader(InputStream inputStream, EventStreamSource source) { - this( - new InputStreamReader( - Objects.requireNonNull(inputStream, "inputStream"), StandardCharsets.UTF_8), - source); - } - private EventStreamReader(Reader reader, EventStreamSource source) { this.reader = new BufferedReader(Objects.requireNonNull(reader, "reader")); this.source = source; @@ -165,133 +158,84 @@ public void close() throws IOException { } private ServerSentEvent readCurrentEvent() throws IOException { - String id = null; - String event = null; - Long retry = null; - StringBuilder data = new StringBuilder(); - StringBuilder comment = new StringBuilder(); - boolean hasData = false; - boolean hasComment = false; - - while (!closed) { + PendingEvent pending = new PendingEvent(); + ServerSentEvent result = null; + while (result == null && !closed) { String line = reader.readLine(); if (line == null) { currentEof = true; - return hasData || hasComment - ? buildEvent(id, event, retry, data, comment, hasData, hasComment) - : null; - } - if (beginning) { - beginning = false; - if (!line.isEmpty() && line.charAt(0) == '\uFEFF') { - line = line.substring(1); - } - } - if (line.isEmpty()) { - if (hasData || hasComment) { - return buildEvent(id, event, retry, data, comment, hasData, hasComment); - } - id = null; - event = null; - retry = null; - continue; + return pending.build(); } - if (line.charAt(0) == ':') { - if (hasComment) { - comment.append('\n'); - } - String value = line.substring(1); - if (value.startsWith(" ")) { - value = value.substring(1); - } - comment.append(value); - hasComment = true; - continue; + result = processLine(stripBom(line), pending); + } + return result; + } + + private String stripBom(String line) { + if (beginning) { + beginning = false; + if (!line.isEmpty() && line.charAt(0) == '\uFEFF') { + return line.substring(1); } + } + return line; + } - int separator = line.indexOf(':'); - String field = separator < 0 ? line : line.substring(0, separator); - String value = separator < 0 ? "" : line.substring(separator + 1); - if (value.startsWith(" ")) { - value = value.substring(1); + private ServerSentEvent processLine(String line, PendingEvent pending) { + if (line.isEmpty()) { + ServerSentEvent event = pending.build(); + if (event == null) { + pending.resetFields(); } - switch (field) { - case "data" -> { - if (hasData) { - data.append('\n'); - } - data.append(value); - hasData = true; - } - case "event" -> event = value; - case "id" -> { - if (value.indexOf('\0') < 0) { - id = value; - lastEventId = value; - } - } - case "retry" -> { - if (isDigits(value)) { - try { - retry = Long.parseLong(value); - retryDelayMs = retry; - } catch (NumberFormatException _) { - // An out-of-range retry value is ignored by the SSE parsing rules. - } - } - } - default -> { - // Unknown fields are ignored by the SSE protocol. - } + return event; + } + if (line.charAt(0) == ':') { + pending.appendComment(stripLeadingSpace(line.substring(1))); + return null; + } + + int separator = line.indexOf(':'); + String field = separator < 0 ? line : line.substring(0, separator); + String value = separator < 0 ? "" : stripLeadingSpace(line.substring(separator + 1)); + switch (field) { + case "data" -> pending.appendData(value); + case "event" -> pending.event = value; + case "id" -> applyId(value, pending); + case "retry" -> applyRetry(value, pending); + default -> { + // Unknown fields are ignored by the SSE protocol. } } return null; } - private ServerSentEvent buildEvent( - String id, - String event, - Long retry, - StringBuilder data, - StringBuilder comment, - boolean hasData, - boolean hasComment) { - ServerSentEvent.Builder builder = ServerSentEvent.builder(); - if (id != null) { - builder.id(id); - } - if (event != null) { - builder.event(event); + private void applyId(String value, PendingEvent pending) { + if (value.indexOf('\0') < 0) { + pending.id = value; + lastEventId = value; } - if (retry != null) { - builder.retry(retry); - } - if (hasData) { - builder.data(data.toString()); + } + + private void applyRetry(String value, PendingEvent pending) { + if (!isDigits(value)) { + return; } - if (hasComment) { - builder.comment(comment.toString()); + try { + long retry = Long.parseLong(value); + pending.retry = retry; + retryDelayMs = retry; + } catch (NumberFormatException _) { + // An out-of-range retry value is ignored by the SSE parsing rules. } - return builder.build(); + } + + private static String stripLeadingSpace(String value) { + return value.startsWith(" ") ? value.substring(1) : value; } private boolean reconnect() throws IOException { - synchronized (reconnectLock) { - if (closed) { - return false; - } - if (retryDelayMs > 0) { - try { - reconnectLock.wait(retryDelayMs); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new IOException( - "Interrupted while waiting to reconnect to the event stream", e); - } - } - if (closed) { - return false; - } + if (!awaitReconnectDelay()) { + return false; } reader.close(); InputStream inputStream; @@ -311,6 +255,24 @@ private boolean reconnect() throws IOException { return true; } + private boolean awaitReconnectDelay() throws IOException { + synchronized (reconnectLock) { + long deadline = System.nanoTime() + retryDelayMs * 1_000_000L; + long remainingMs = retryDelayMs; + while (!closed && remainingMs > 0) { + try { + reconnectLock.wait(remainingMs); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException( + "Interrupted while waiting to reconnect to the event stream", e); + } + remainingMs = (deadline - System.nanoTime()) / 1_000_000L; + } + return !closed; + } + } + private static boolean isDigits(String value) { if (value.isEmpty()) { return false; @@ -323,6 +285,69 @@ private static boolean isDigits(String value) { return true; } + /** Accumulates the fields of the event being parsed. */ + private static final class PendingEvent { + + private final StringBuilder data = new StringBuilder(); + + private final StringBuilder comment = new StringBuilder(); + + private String id; + + private String event; + + private Long retry; + + private boolean hasData; + + private boolean hasComment; + + private void appendData(String value) { + if (hasData) { + data.append('\n'); + } + data.append(value); + hasData = true; + } + + private void appendComment(String value) { + if (hasComment) { + comment.append('\n'); + } + comment.append(value); + hasComment = true; + } + + private void resetFields() { + id = null; + event = null; + retry = null; + } + + private ServerSentEvent build() { + if (!hasData && !hasComment) { + return null; + } + ServerSentEvent.Builder builder = ServerSentEvent.builder(); + if (id != null) { + builder.id(id); + } + if (event != null) { + builder.event(event); + } + if (retry != null) { + builder.retry(retry); + } + if (hasData) { + builder.data(data.toString()); + } + if (hasComment) { + builder.comment(comment.toString()); + } + return builder.build(); + } + } + /** Opens a new stream using the latest event ID. */ @FunctionalInterface public interface EventStreamSource { diff --git a/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java b/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java index aee7b7a8d7..764fcc8c94 100644 --- a/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java +++ b/org.restlet/src/main/java/org/restlet/representation/EventStreamRepresentation.java @@ -45,12 +45,13 @@ public EventStreamRepresentation(EventWriter writer) { * @param heartbeatInterval The heartbeat interval, or null to disable automatic heartbeats. */ public EventStreamRepresentation(EventWriter writer, Duration heartbeatInterval) { - super(MediaType.TEXT_EVENT_STREAM); - this.writer = Objects.requireNonNull(writer, "writer"); + Objects.requireNonNull(writer, "writer"); if (heartbeatInterval != null && (heartbeatInterval.isZero() || heartbeatInterval.isNegative())) { throw new IllegalArgumentException("The heartbeat interval must be positive"); } + super(MediaType.TEXT_EVENT_STREAM); + this.writer = writer; this.heartbeatInterval = heartbeatInterval; setCharacterSet(CharacterSet.UTF_8); setTransient(true); @@ -74,28 +75,17 @@ public void write(OutputStream outputStream) throws IOException { () -> writeHeartbeats( sink, stopped, heartbeatFailure, writerThread)); - Throwable writerFailure = null; try { writer.write(sink); - } catch (IOException | RuntimeException | Error e) { - writerFailure = e; + } catch (IOException | RuntimeException e) { + stopHeartbeats(stopped, heartbeatThread, e); throw e; } finally { + // Ensures the heartbeat thread stops even when the writer fails with an Error. stopped.set(true); heartbeatThread.interrupt(); - try { - heartbeatThread.join(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - IOException interrupted = - new IOException("Interrupted while stopping SSE heartbeats", e); - if (writerFailure != null) { - writerFailure.addSuppressed(interrupted); - } else { - throw interrupted; - } - } } + stopHeartbeats(stopped, heartbeatThread, null); IOException failure = heartbeatFailure.get(); if (failure != null) { throw failure; @@ -103,6 +93,24 @@ public void write(OutputStream outputStream) throws IOException { } } + private static void stopHeartbeats( + AtomicBoolean stopped, Thread heartbeatThread, Exception writerFailure) + throws IOException { + stopped.set(true); + heartbeatThread.interrupt(); + try { + heartbeatThread.join(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + IOException interrupted = + new IOException("Interrupted while stopping SSE heartbeats", e); + if (writerFailure == null) { + throw interrupted; + } + writerFailure.addSuppressed(interrupted); + } + } + private void writeHeartbeats( EventSink sink, AtomicBoolean stopped, @@ -111,7 +119,7 @@ private void writeHeartbeats( while (!stopped.get()) { try { Thread.sleep(heartbeatInterval); - } catch (InterruptedException e) { + } catch (InterruptedException _) { if (!stopped.get()) { Thread.currentThread().interrupt(); } diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java index ceba9302a6..d264cba544 100644 --- a/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/connector/EventStreamConnectorTestCase.java @@ -185,7 +185,11 @@ public void handle(Request request, Response response) { new EventStreamRepresentation( sink -> { try { - Thread.sleep(350); + // Waits past the idle timeout before sending. + boolean released = + new CountDownLatch(1) + .await(350, TimeUnit.MILLISECONDS); + assertFalse(released); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new java.io.IOException( diff --git a/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java b/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java index 44a1fac10d..689ea81283 100644 --- a/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java +++ b/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java @@ -9,6 +9,7 @@ package org.restlet.representation; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -22,6 +23,8 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; import org.restlet.data.MediaType; import org.restlet.data.ServerSentEvent; @@ -49,13 +52,15 @@ void serializesUtf8FieldsAndMultilineValues() throws IOException { assertEquals(MediaType.TEXT_EVENT_STREAM, representation.getMediaType()); assertEquals(StandardCharsets.UTF_8.name(), representation.getCharacterSet().getName()); assertEquals( - "id: 42\n" - + "event: message\n" - + "data: héllo\n" - + "data: world \n" - + "retry: 250\n" - + ": heartbeat\n" - + "\n", + """ + id: 42 + event: message + data: héllo + data: world\s + retry: 250 + : heartbeat + + """, output.toString(StandardCharsets.UTF_8)); } @@ -66,7 +71,10 @@ void emitsConfiguredCommentHeartbeatsWhileTheWriterIsIdle() throws IOException { new EventStreamRepresentation( sink -> { try { - Thread.sleep(Duration.ofMillis(60)); + // Keeps the writer idle long enough for several heartbeats. + boolean released = + new CountDownLatch(1).await(60, TimeUnit.MILLISECONDS); + assertFalse(released); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException("Interrupted while awaiting an event", e); @@ -87,14 +95,18 @@ void readsFieldsCommentsAndReconnectState() throws IOException { EventStreamReader reader = new EventStreamReader( new StringReader( - "\uFEFFid:cursor-1\r\n" - + "event:update\r\n" - + "data:first\r\n" - + "data:second\r\n" - + "retry:5\r\n" - + "\r\n" - + ":keepalive\n\n" - + "data:\n\n")); + """ + \uFEFFid:cursor-1\r + event:update\r + data:first\r + data:second\r + retry:5\r + \r + :keepalive + + data: + + """)); ServerSentEvent event = reader.readEvent(); assertEquals("cursor-1", event.getId()); @@ -138,11 +150,10 @@ void reconnectsAfterRetryAndPassesTheLastEventId() throws IOException { @Test void validatesEventFields() { - assertThrows(IllegalArgumentException.class, () -> ServerSentEvent.builder().id("bad\nid")); - assertThrows( - IllegalArgumentException.class, - () -> ServerSentEvent.builder().event("bad\nevent")); - assertThrows(IllegalArgumentException.class, () -> ServerSentEvent.builder().retry(-1L)); + ServerSentEvent.Builder builder = ServerSentEvent.builder(); + assertThrows(IllegalArgumentException.class, () -> builder.id("bad\nid")); + assertThrows(IllegalArgumentException.class, () -> builder.event("bad\nevent")); + assertThrows(IllegalArgumentException.class, () -> builder.retry(-1L)); } private static InputStream stream(String content) {