feat: add Server-Sent Events API

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.

org.restlet/src/main/java/org/restlet/Request.java

@@ -469,6 +469,15 @@ public Series<Header> 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.<br>

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 =

org.restlet/src/main/java/org/restlet/data/ServerSentEvent.java

@@ -0,0 +1,145 @@
+/**
+ * Copyright 2005-2026 Qlik
+ *<p>
+ * 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
+ *<p>
+ * 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");
+            }
+        }
+    }
+}

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;
             }
         }
     }

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<Boolean> 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);
         }