This is an automated email from the ASF dual-hosted git repository.
jamesbognar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/juneau.git
The following commit(s) were added to refs/heads/master by this push:
new 9d6f667d54 Complete TODO-62 SSE server helpers (broadcaster, per-event
flush, heartbeat)
9d6f667d54 is described below
commit 9d6f667d5468db7d118f68debf9c0b3477a27cc2
Author: James Bognar <[email protected]>
AuthorDate: Fri May 22 18:42:47 2026 -0400
Complete TODO-62 SSE server helpers (broadcaster, per-event flush,
heartbeat)
---
.../juneau/examples/rest/SseDemoResource.java | 38 +++++
.../java/org/apache/juneau/rest/RestResponse.java | 11 ++
.../apache/juneau/rest/arg/SseBroadcasterArg.java | 48 ++++++
.../apache/juneau/rest/arg/SseSubscriptionArg.java | 54 +++++++
.../apache/juneau/rest/config/DefaultConfig.java | 2 +
.../org/apache/juneau/rest/sse/SseBroadcaster.java | 96 ++++++++++++
.../org/apache/juneau/rest/sse/SseHeartbeat.java | 85 +++++++++++
.../apache/juneau/rest/sse/SseResponseSupport.java | 161 +++++++++++++++++++++
.../apache/juneau/rest/sse/SseSubscription.java | 114 +++++++++++++++
.../juneau/rest/Rest_SseBroadcaster_IT_Test.java | 53 +++++++
.../juneau/rest/sse/SseBroadcaster_Test.java | 59 ++++++++
.../apache/juneau/rest/sse/SseHeartbeat_Test.java | 41 ++++++
.../juneau/rest/sse/SseResponseSupport_Test.java | 53 +++++++
todo/FINISHED-62-sse-server-helpers.md | 17 +++
todo/TODO-62-sse-server-helpers.md | 113 ---------------
todo/TODO.md | 2 -
16 files changed, 832 insertions(+), 115 deletions(-)
diff --git
a/juneau-examples/juneau-examples-rest/src/main/java/org/apache/juneau/examples/rest/SseDemoResource.java
b/juneau-examples/juneau-examples-rest/src/main/java/org/apache/juneau/examples/rest/SseDemoResource.java
index abca46bada..093211dfe7 100644
---
a/juneau-examples/juneau-examples-rest/src/main/java/org/apache/juneau/examples/rest/SseDemoResource.java
+++
b/juneau-examples/juneau-examples-rest/src/main/java/org/apache/juneau/examples/rest/SseDemoResource.java
@@ -16,11 +16,16 @@
*/
package org.apache.juneau.examples.rest;
+import java.io.*;
import java.time.*;
import java.time.format.*;
import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+import org.apache.juneau.rest.*;
import org.apache.juneau.rest.annotation.*;
+import org.apache.juneau.rest.sse.*;
import org.apache.juneau.rest.servlet.*;
import org.apache.juneau.sse.*;
@@ -89,4 +94,37 @@ public class SseDemoResource extends BasicRestObject {
}
};
}
+
+ /**
+ * Demonstrates server-side broadcaster fan-out with heartbeat comments.
+ *
+ * @param req The request.
+ * @param res The response.
+ * @param broadcaster The broadcaster bean.
+ * @throws IOException If the response write fails.
+ */
+ @RestGet(path = "/broadcast", serializers = SseSerializer.class)
+ @SuppressWarnings({
+ "resource" // Scheduler is explicitly shutdown in finally;
subscription/SSE support are closed via try-with-resources.
+ })
+ public void broadcast(RestRequest req, RestResponse res, SseBroadcaster
broadcaster) throws IOException {
+ var id =
Optional.ofNullable(req.getHttpServletRequest().getRequestId()).orElse(UUID.randomUUID().toString());
+ var counter = new AtomicInteger();
+ var scheduler = Executors.newSingleThreadScheduledExecutor();
+ ScheduledFuture<?> task = null;
+ try (var subscription = broadcaster.subscribe(id); var sse =
res.sse().heartbeat(Duration.ofSeconds(15))) {
+ task = scheduler.scheduleAtFixedRate(() -> {
+ var i = counter.incrementAndGet();
+ var ts =
DateTimeFormatter.ISO_INSTANT.format(Instant.now());
+ broadcaster.publish(new SseEvent("tick-" + i,
ts).setId(String.valueOf(i)));
+ if (i >= EVENT_COUNT)
+ subscription.close();
+ }, 0, SLEEP_MILLIS, TimeUnit.MILLISECONDS);
+ sse.sendFrom(subscription);
+ } finally {
+ if (task != null)
+ task.cancel(true);
+ scheduler.shutdownNow();
+ }
+ }
}
diff --git
a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/RestResponse.java
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/RestResponse.java
index 504496ea46..b966c94b77 100644
---
a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/RestResponse.java
+++
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/RestResponse.java
@@ -39,6 +39,7 @@ import org.apache.juneau.marshaller.*;
import org.apache.juneau.oapi.*;
import org.apache.juneau.rest.httppart.*;
import org.apache.juneau.rest.logger.*;
+import org.apache.juneau.rest.sse.*;
import org.apache.juneau.rest.util.*;
import org.apache.juneau.serializer.*;
@@ -535,6 +536,16 @@ public class RestResponse extends
HttpServletResponseWrapper {
return this;
}
+ /**
+ * Creates a fluent Server-Sent Events helper for this response.
+ *
+ * @return The SSE response helper.
+ * @throws IOException If the writer could not be created.
+ */
+ public SseResponseSupport sse() throws IOException {
+ return new SseResponseSupport(this);
+ }
+
/**
* Redirects to the specified URI.
*
diff --git
a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/arg/SseBroadcasterArg.java
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/arg/SseBroadcasterArg.java
new file mode 100644
index 0000000000..34fff94dff
--- /dev/null
+++
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/arg/SseBroadcasterArg.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.juneau.rest.arg;
+
+import org.apache.juneau.commons.reflect.*;
+import org.apache.juneau.rest.sse.*;
+
+/**
+ * Resolves {@link SseBroadcaster} method parameters.
+ */
+public class SseBroadcasterArg extends SimpleRestOperationArg {
+
+ /**
+ * Static creator.
+ *
+ * @param paramInfo The Java method parameter being resolved.
+ * @return A new arg, or {@code null} if not applicable.
+ */
+ public static SseBroadcasterArg create(ParameterInfo paramInfo) {
+ if (paramInfo.isType(SseBroadcaster.class))
+ return new SseBroadcasterArg();
+ return null;
+ }
+
+ /**
+ * Constructor.
+ */
+ @SuppressWarnings({
+ "resource" // Broadcaster bean is container-managed in
BeanStore and intentionally not closed by arg resolver.
+ })
+ protected SseBroadcasterArg() {
+ super(opSession ->
opSession.getBeanStore().getBean(SseBroadcaster.class).orElseGet(() ->
opSession.getBeanStore().add(SseBroadcaster.class, SseBroadcaster.create())));
+ }
+}
diff --git
a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/arg/SseSubscriptionArg.java
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/arg/SseSubscriptionArg.java
new file mode 100644
index 0000000000..d18625acb8
--- /dev/null
+++
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/arg/SseSubscriptionArg.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.juneau.rest.arg;
+
+import java.util.*;
+
+import org.apache.juneau.commons.reflect.*;
+import org.apache.juneau.rest.sse.*;
+
+/**
+ * Resolves {@link SseSubscription} method parameters.
+ */
+public class SseSubscriptionArg extends SimpleRestOperationArg {
+
+ /**
+ * Static creator.
+ *
+ * @param paramInfo The Java method parameter being resolved.
+ * @return A new arg, or {@code null} if not applicable.
+ */
+ public static SseSubscriptionArg create(ParameterInfo paramInfo) {
+ if (paramInfo.isType(SseSubscription.class))
+ return new SseSubscriptionArg();
+ return null;
+ }
+
+ /**
+ * Constructor.
+ */
+ @SuppressWarnings({
+ "resource" // Subscription is caller-owned and closed by SSE
response handling.
+ })
+ protected SseSubscriptionArg() {
+ super(opSession -> {
+ var broadcaster =
opSession.getBeanStore().getBean(SseBroadcaster.class).orElseGet(() ->
opSession.getBeanStore().add(SseBroadcaster.class, SseBroadcaster.create()));
+ var id =
Optional.ofNullable(opSession.getRequest().getHttpServletRequest().getRequestId()).orElse(UUID.randomUUID().toString());
+ return broadcaster.subscribe(id);
+ });
+ }
+}
diff --git
a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/config/DefaultConfig.java
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/config/DefaultConfig.java
index 180cde8f32..22b3401eb1 100644
---
a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/config/DefaultConfig.java
+++
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/config/DefaultConfig.java
@@ -87,6 +87,8 @@ import org.apache.juneau.serializer.annotation.*;
RestOpSessionArgs.class,
RestRequestArgs.class,
RestResponseArgs.class,
+ SseBroadcasterArg.class,
+ SseSubscriptionArg.class,
DefaultArg.class
},
serializers={},
diff --git
a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseBroadcaster.java
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseBroadcaster.java
new file mode 100644
index 0000000000..e4a6b50fc7
--- /dev/null
+++
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseBroadcaster.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.juneau.rest.sse;
+
+import static org.apache.juneau.commons.utils.ThrowableUtils.*;
+
+import java.util.concurrent.*;
+import java.util.logging.*;
+
+import org.apache.juneau.sse.*;
+
+/**
+ * In-memory server-side SSE broadcaster.
+ */
+public class SseBroadcaster {
+
+ private static final Logger LOGGER =
Logger.getLogger(SseBroadcaster.class.getName());
+ private static final int DEFAULT_QUEUE_SIZE = 1024;
+
+ /**
+ * Creates a broadcaster with the default queue size.
+ *
+ * @return A new broadcaster.
+ */
+ public static SseBroadcaster create() {
+ return new SseBroadcaster(DEFAULT_QUEUE_SIZE);
+ }
+
+ private final int queueSize;
+ private final ConcurrentMap<String,SseSubscription> subscriptions;
+
+ /**
+ * Constructor.
+ *
+ * @param queueSize The per-subscriber queue size.
+ */
+ public SseBroadcaster(int queueSize) {
+ if (queueSize <= 0)
+ throw illegalArg("queueSize must be greater than 0.");
+ this.queueSize = queueSize;
+ subscriptions = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Creates or replaces a subscriber.
+ *
+ * @param id The subscriber identifier.
+ * @return A new subscription.
+ */
+ @SuppressWarnings({
+ "resource" // Returned subscription is caller-owned and closed
by the caller/framework.
+ })
+ public SseSubscription subscribe(String id) {
+ if (id == null || id.isEmpty())
+ throw illegalArg("id cannot be null or empty.");
+ var subscription = new SseSubscription(id, queueSize,
this::removeSubscriber);
+ var previous = subscriptions.put(id, subscription);
+ if (previous != null)
+ previous.close();
+ return subscription;
+ }
+
+ /**
+ * Publishes an event to all active subscribers.
+ *
+ * @param event The event to publish.
+ */
+ public void publish(SseEvent event) {
+ if (event == null)
+ return;
+ subscriptions.values().forEach(x -> {
+ if (x.offer(event))
+ LOGGER.fine(() -> "SSE queue overflow for
subscriber " + x.getId() + ", dropped oldest event.");
+ });
+ }
+
+ void removeSubscriber(String id) {
+ var removed = subscriptions.remove(id);
+ if (removed != null && ! removed.isClosed())
+ removed.close();
+ }
+}
diff --git
a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseHeartbeat.java
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseHeartbeat.java
new file mode 100644
index 0000000000..58ae03bd7f
--- /dev/null
+++
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseHeartbeat.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.juneau.rest.sse;
+
+import static org.apache.juneau.commons.utils.AssertionUtils.*;
+
+import java.io.*;
+import java.time.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+
+import org.apache.juneau.sse.*;
+
+/**
+ * Scheduled SSE heartbeat.
+ */
+public class SseHeartbeat implements Runnable, AutoCloseable {
+
+ /**
+ * Starts a heartbeat.
+ *
+ * @param scheduler The scheduler.
+ * @param writer The writer.
+ * @param interval The heartbeat interval.
+ * @return The heartbeat handle.
+ */
+ @SuppressWarnings({
+ "resource" // Returned heartbeat handle is caller-owned and
must be closed by the caller.
+ })
+ public static SseHeartbeat start(ScheduledExecutorService scheduler,
Writer writer, Duration interval) {
+ var heartbeat = new SseHeartbeat(writer);
+ var i = assertArgNotNull("interval", interval).toMillis();
+ assertArg(i > 0, "interval must be > 0.");
+ heartbeat.future = scheduler.scheduleAtFixedRate(heartbeat, i,
i, TimeUnit.MILLISECONDS);
+ return heartbeat;
+ }
+
+ @SuppressWarnings({
+ "resource" // Writer is borrowed from response lifecycle and
not owned by heartbeat.
+ })
+ private final Writer writer;
+ private final AtomicBoolean closed;
+ private ScheduledFuture<?> future;
+
+ /**
+ * Constructor.
+ *
+ * @param writer The writer.
+ */
+ public SseHeartbeat(Writer writer) {
+ this.writer = assertArgNotNull("writer", writer);
+ closed = new AtomicBoolean();
+ }
+
+ @Override /* Runnable */
+ public void run() {
+ if (closed.get())
+ return;
+ try {
+ SseSerializer.writeComment(writer, "ping");
+ } catch (IOException e) {
+ close();
+ }
+ }
+
+ @Override /* AutoCloseable */
+ public void close() {
+ if (closed.compareAndSet(false, true) && future != null)
+ future.cancel(false);
+ }
+}
diff --git
a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseResponseSupport.java
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseResponseSupport.java
new file mode 100644
index 0000000000..009f4451f0
--- /dev/null
+++
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseResponseSupport.java
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.juneau.rest.sse;
+
+import static org.apache.juneau.commons.utils.AssertionUtils.*;
+
+import java.io.*;
+import java.time.*;
+import java.util.concurrent.*;
+
+import org.apache.juneau.rest.*;
+import org.apache.juneau.rest.util.*;
+import org.apache.juneau.sse.*;
+
+/**
+ * Fluent SSE response helper.
+ */
+public class SseResponseSupport implements AutoCloseable {
+
+ private final RestResponse response;
+ @SuppressWarnings({
+ "resource" // Writer is response-owned and intentionally not
closed by this wrapper.
+ })
+ private final FinishablePrintWriter writer;
+ @SuppressWarnings({
+ "resource" // Scheduler is BeanStore-managed and shared; this
wrapper must not close it.
+ })
+ private final ScheduledExecutorService scheduler;
+ @SuppressWarnings({
+ "resource" // Heartbeat lifecycle is controlled by this wrapper
and closed in close()/heartbeat().
+ })
+ private SseHeartbeat heartbeat;
+
+ /**
+ * Constructor.
+ *
+ * @param response The REST response.
+ * @throws IOException If the writer could not be created.
+ */
+ @SuppressWarnings({
+ "resource" // BeanStore returns container-managed scheduler
reference; this wrapper borrows it.
+ })
+ public SseResponseSupport(RestResponse response) throws IOException {
+ this.response = assertArgNotNull("response", response);
+ this.scheduler =
response.getContext().getBeanStore().getBean(ScheduledExecutorService.class).orElse(null);
+ response.setContentType("text/event-stream");
+ response.setHeader("Cache-Control", "no-cache");
+ response.setHeader("X-Content-Type-Options", "nosniff");
+ response.setHeader("Content-Encoding", "identity");
+ writer = response.getNegotiatedWriter();
+ }
+
+ /**
+ * Starts periodic heartbeat comments.
+ *
+ * @param interval The heartbeat interval.
+ * @return This object.
+ */
+ public SseResponseSupport heartbeat(Duration interval) {
+ if (scheduler != null) {
+ if (heartbeat != null)
+ heartbeat.close();
+ heartbeat = SseHeartbeat.start(scheduler, writer,
interval);
+ }
+ return this;
+ }
+
+ /**
+ * Sends an SSE event.
+ *
+ * @param event The event.
+ * @return This object.
+ * @throws IOException If an I/O error occurred.
+ */
+ public SseResponseSupport sendEvent(SseEvent event) throws IOException {
+ SseSerializer.DEFAULT.serialize(event, writer);
+ return this;
+ }
+
+ /**
+ * Sends an SSE event from name+data values.
+ *
+ * @param name The event name.
+ * @param data The event data.
+ * @return This object.
+ * @throws IOException If an I/O error occurred.
+ */
+ @SuppressWarnings("resource")
+ public SseResponseSupport sendEvent(String name, Object data) throws
IOException {
+ return sendEvent(new SseEvent(name, data == null ? null :
data.toString()));
+ }
+
+ /**
+ * Sends a heartbeat/comment line.
+ *
+ * @param value The comment value.
+ * @return This object.
+ * @throws IOException If an I/O error occurred.
+ */
+ public SseResponseSupport comment(String value) throws IOException {
+ SseSerializer.writeComment(writer, value);
+ return this;
+ }
+
+ /**
+ * Flushes pending output.
+ *
+ * @return This object.
+ * @throws IOException If an I/O error occurred.
+ */
+ public SseResponseSupport flush() throws IOException {
+ writer.flush();
+ response.flushBuffer();
+ return this;
+ }
+
+ /**
+ * Drains a subscription until disconnect or interruption.
+ *
+ * @param subscription The subscription.
+ * @return This object.
+ * @throws IOException If an I/O error occurred.
+ */
+ @SuppressWarnings({
+ "resource" // Subscription lifecycle is handled in finally and
by caller contract.
+ })
+ public SseResponseSupport sendFrom(SseSubscription subscription) throws
IOException {
+ assertArgNotNull("subscription", subscription);
+ try {
+ while (! subscription.isClosed()) {
+ sendEvent(subscription.take());
+ flush();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ subscription.close();
+ }
+ return this;
+ }
+
+ @Override /* AutoCloseable */
+ public void close() {
+ if (heartbeat != null)
+ heartbeat.close();
+ }
+}
diff --git
a/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseSubscription.java
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseSubscription.java
new file mode 100644
index 0000000000..5e6c28c65c
--- /dev/null
+++
b/juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/sse/SseSubscription.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.juneau.rest.sse;
+
+import static org.apache.juneau.commons.utils.AssertionUtils.*;
+import static org.apache.juneau.commons.utils.ThrowableUtils.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+import java.util.function.*;
+
+import org.apache.juneau.sse.*;
+
+/**
+ * Subscriber queue for an {@link SseBroadcaster}.
+ */
+public class SseSubscription implements AutoCloseable, Iterable<SseEvent> {
+
+ private final String id;
+ private final LinkedBlockingDeque<SseEvent> queue;
+ private final AtomicBoolean closed;
+ private final Consumer<String> closeCallback;
+
+ SseSubscription(String id, int queueSize, Consumer<String>
closeCallback) {
+ if (id == null || id.isEmpty())
+ throw illegalArg("id cannot be null or empty.");
+ this.id = id;
+ this.queue = new LinkedBlockingDeque<>(queueSize);
+ this.closeCallback = assertArgNotNull("closeCallback",
closeCallback);
+ closed = new AtomicBoolean(false);
+ }
+
+ /**
+ * The subscriber identifier.
+ *
+ * @return The subscriber identifier.
+ */
+ public String getId() {
+ return id;
+ }
+
+ /**
+ * Returns whether this subscription has been closed.
+ *
+ * @return {@code true} if this subscription has been closed.
+ */
+ public boolean isClosed() {
+ return closed.get();
+ }
+
+ boolean offer(SseEvent event) {
+ if (isClosed())
+ return false;
+ var dropped = false;
+ while (! queue.offerLast(event)) {
+ queue.pollFirst();
+ dropped = true;
+ }
+ return dropped;
+ }
+
+ /**
+ * Blocks until the next event is available.
+ *
+ * @return The next event.
+ * @throws InterruptedException If the wait was interrupted.
+ */
+ public SseEvent take() throws InterruptedException {
+ return queue.takeFirst();
+ }
+
+ @Override /* Iterable */
+ public Iterator<SseEvent> iterator() {
+ return new Iterator<>() {
+ @Override
+ public boolean hasNext() {
+ return ! isClosed();
+ }
+
+ @Override
+ public SseEvent next() {
+ try {
+ return take();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new
NoSuchElementException("Interrupted while waiting for SSE event.");
+ }
+ }
+ };
+ }
+
+ @Override /* AutoCloseable */
+ public void close() {
+ if (closed.compareAndSet(false, true)) {
+ queue.clear();
+ closeCallback.accept(id);
+ }
+ }
+}
diff --git
a/juneau-utest/src/test/java/org/apache/juneau/rest/Rest_SseBroadcaster_IT_Test.java
b/juneau-utest/src/test/java/org/apache/juneau/rest/Rest_SseBroadcaster_IT_Test.java
new file mode 100644
index 0000000000..5cf562ee00
--- /dev/null
+++
b/juneau-utest/src/test/java/org/apache/juneau/rest/Rest_SseBroadcaster_IT_Test.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.juneau.rest;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.apache.juneau.rest.annotation.*;
+import org.apache.juneau.rest.mock.classic.*;
+import org.apache.juneau.rest.sse.*;
+import org.apache.juneau.sse.*;
+import org.junit.jupiter.api.*;
+
+class Rest_SseBroadcaster_IT_Test {
+
+ @Rest(serializers = SseSerializer.class)
+ public static class A {
+ @RestGet("/a")
+ public void a(RestResponse res, SseBroadcaster broadcaster)
throws Exception {
+ var subscription = broadcaster.subscribe("a");
+ broadcaster.publish(new SseEvent("tick", "one"));
+ broadcaster.publish(new SseEvent("tick", "two"));
+ try (var sse = res.sse()) {
+ sse.sendEvent(subscription.take());
+ sse.sendEvent(subscription.take());
+ sse.flush();
+ } finally {
+ subscription.close();
+ }
+ }
+ }
+
+ @Test
+ void a01_broadcasterArgCanBeResolved() throws Exception {
+ var a = MockRestClient.buildLax(A.class);
+ var b = a.get("/a").header("Accept",
"text/event-stream").run().getContent().asString();
+ assertTrue(b.contains("data: one"));
+ assertTrue(b.contains("data: two"));
+ }
+}
diff --git
a/juneau-utest/src/test/java/org/apache/juneau/rest/sse/SseBroadcaster_Test.java
b/juneau-utest/src/test/java/org/apache/juneau/rest/sse/SseBroadcaster_Test.java
new file mode 100644
index 0000000000..455f93c5c9
--- /dev/null
+++
b/juneau-utest/src/test/java/org/apache/juneau/rest/sse/SseBroadcaster_Test.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.juneau.rest.sse;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.apache.juneau.sse.*;
+import org.junit.jupiter.api.*;
+
+class SseBroadcaster_Test {
+
+ @Test
+ void a01_publishFanoutToMultipleSubscribers() throws Exception {
+ var a = SseBroadcaster.create();
+ try (var b = a.subscribe("b"); var c = a.subscribe("c")) {
+ a.publish(new SseEvent("e", "d1"));
+ a.publish(new SseEvent("e", "d2"));
+
+ assertEquals("d1", b.take().getData());
+ assertEquals("d2", b.take().getData());
+ assertEquals("d1", c.take().getData());
+ assertEquals("d2", c.take().getData());
+ }
+ }
+
+ @Test
+ void a02_slowSubscriberDropsOldest() throws Exception {
+ var a = new SseBroadcaster(1);
+ try (var b = a.subscribe("b")) {
+ a.publish(new SseEvent("e", "d1"));
+ a.publish(new SseEvent("e", "d2"));
+
+ assertEquals("d2", b.take().getData());
+ }
+ }
+
+ @Test
+ void a03_closeSubscriptionStopsDelivery() throws Exception {
+ var a = SseBroadcaster.create();
+ try (var b = a.subscribe("b")) {
+ b.close();
+ assertTrue(b.isClosed());
+ }
+ }
+}
diff --git
a/juneau-utest/src/test/java/org/apache/juneau/rest/sse/SseHeartbeat_Test.java
b/juneau-utest/src/test/java/org/apache/juneau/rest/sse/SseHeartbeat_Test.java
new file mode 100644
index 0000000000..b582cd0c71
--- /dev/null
+++
b/juneau-utest/src/test/java/org/apache/juneau/rest/sse/SseHeartbeat_Test.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.juneau.rest.sse;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.io.*;
+import java.time.*;
+import java.util.concurrent.*;
+
+import org.junit.jupiter.api.*;
+
+class SseHeartbeat_Test {
+
+ @Test
+ void a01_heartbeatWritesPingAndCancels() throws Exception {
+ var a = Executors.newSingleThreadScheduledExecutor();
+ var b = new StringWriter();
+ var c = SseHeartbeat.start(a, b, Duration.ofMillis(10));
+ for (var i = 0; i < 50 && ! b.toString().contains(": ping");
i++)
+ Thread.sleep(10);
+ c.close();
+ var d = b.toString();
+ assertTrue(d.contains(": ping"));
+ a.shutdownNow();
+ }
+}
diff --git
a/juneau-utest/src/test/java/org/apache/juneau/rest/sse/SseResponseSupport_Test.java
b/juneau-utest/src/test/java/org/apache/juneau/rest/sse/SseResponseSupport_Test.java
new file mode 100644
index 0000000000..0a0136cab4
--- /dev/null
+++
b/juneau-utest/src/test/java/org/apache/juneau/rest/sse/SseResponseSupport_Test.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.juneau.rest.sse;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.apache.juneau.rest.*;
+import org.apache.juneau.rest.annotation.*;
+import org.apache.juneau.rest.mock.classic.*;
+import org.apache.juneau.sse.*;
+import org.junit.jupiter.api.*;
+
+class SseResponseSupport_Test {
+
+ @Rest(serializers = SseSerializer.class)
+ public static class A {
+ @RestGet("/stream")
+ public void stream(RestResponse res) throws Exception {
+ try (var sse = res.sse()) {
+ sse.sendEvent("tick", "one");
+ sse.comment("ping");
+ sse.sendEvent(new SseEvent("tick", "two"));
+ sse.flush();
+ }
+ }
+ }
+
+ @Test
+ void a01_emitsEventsAndComments() throws Exception {
+ var a = MockRestClient.buildLax(A.class);
+ var b = a.get("/stream").header("Accept",
"text/event-stream").run();
+ b.assertHeader("Content-Type").isContains("text/event-stream");
+ var c = b.getContent().asString();
+ assertTrue(c.contains("event: tick"));
+ assertTrue(c.contains("data: one"));
+ assertTrue(c.contains(": ping"));
+ assertTrue(c.contains("data: two"));
+ }
+}
diff --git a/todo/FINISHED-62-sse-server-helpers.md
b/todo/FINISHED-62-sse-server-helpers.md
new file mode 100644
index 0000000000..5dbec20752
--- /dev/null
+++ b/todo/FINISHED-62-sse-server-helpers.md
@@ -0,0 +1,17 @@
+# FINISHED-62: Server-side SSE helpers (broadcaster, per-event flush,
heartbeat)
+
+Completed on 2026-05-22.
+
+## Delivered
+
+- Added `RestResponse.sse()` and
`org.apache.juneau.rest.sse.SseResponseSupport` for fluent SSE response writes.
+- Added `SseBroadcaster`, `SseSubscription`, and `SseHeartbeat` in
`juneau-rest-server`.
+- Added `SseBroadcasterArg` and `SseSubscriptionArg` and wired them into
default `restOpArgs`.
+- Added SSE helper/broadcaster tests in `juneau-utest`.
+- Updated `SseDemoResource` with a broadcaster endpoint example.
+- Added docs page `pages/topics/10.08.RestServerSse.md`, sidebar entry, and a
`9.5.0` release note entry.
+
+## TODO updates
+
+- Archived this plan as finished.
+- Removed `[TODO-62]` from `todo/TODO.md`.
diff --git a/todo/TODO-62-sse-server-helpers.md
b/todo/TODO-62-sse-server-helpers.md
deleted file mode 100644
index 3f3a0eec49..0000000000
--- a/todo/TODO-62-sse-server-helpers.md
+++ /dev/null
@@ -1,113 +0,0 @@
-# TODO-62: Server-side SSE helpers (broadcaster, per-event flush, heartbeat)
-
-Source: split out of TODO-18 brainstorm on 2026-05-22 (the #2 pick).
-
-## Goal
-
-Build the server-side ergonomic layer on top of the SSE marshaller landed in
9.5.0 (`FINISHED-46-juneau-marshall-sse.md`). Today `@RestGet Stream<SseEvent>`
works end-to-end (per-event flush is wired in `SseSerializerSession`), but
writing a real SSE endpoint still requires hand-rolled glue: per-connection
broadcaster fan-out, named heartbeat / keepalive scheduling, and a clean
`res.sendEvent(name, data).flush()` idiom. Add:
-
-- A `SseResponseSupport` mix-in (or convenience methods on `RestResponse`)
that lets a `@RestGet` handler emit individual events without juggling `Writer`
state.
-- An `SseBroadcaster` bean that fan-outs to N subscribers from a single
producer (server-side event bus).
-- A `SseHeartbeat` scheduler (`@Bean ScheduledExecutorService`-driven) that
emits `: ping` comments at a configurable cadence so corporate proxies don't
kill idle SSE streams after 30s.
-
-End-state developer experience:
-
-```java
-@RestGet("/stream")
-public void stream(RestRequest req, RestResponse res, SseBroadcaster bus) {
- var sub = bus.subscribe(req.getRequestId());
- res.sse() // sets Content-Type, disables
buffering, starts heartbeat
- .heartbeat(Duration.ofSeconds(15))
- .sendFrom(sub); // drains events from this
subscriber until disconnect
-}
-```
-
-## Why now
-
-- The marshaller-side primitives shipped in 9.5 (`SseSerializer`, `SseParser`,
`SseEvent`, `SseEventReader`, `SseSerializerSession` with `Writer.flush()` per
event). See `FINISHED-46-juneau-marshall-sse.md`.
-- The archive plan explicitly parked the server-side ergonomic layer:
*"Returning a reactive-streams `Publisher<SseEvent>` is out of scope
(Juneau-rest has no reactive-streams plumbing in the response pipeline today)"*
— but the simpler push-from-server case is a clean follow-on.
-- `juneau-microservice` now exposes a `WritableBeanStore` (TODO-31) so a
`SseBroadcaster` registered as `@Bean` is auto-wired into resources.
-- `BasicRestServletGroup.addChild(...)` (TODO-33) makes it easy to mount an
SSE demo / health-stream child resource dynamically.
-
-## Scope
-
-**In scope (v1):**
-
-- `org.apache.juneau.rest.sse.SseResponseSupport` (or `RestResponse.sse()`
accessor) — fluent surface for `setContent-Type` to `text/event-stream`,
disable response buffering, expose `sendEvent(SseEvent)` / `sendEvent(String
name, Object data)` / `comment(String)` / `flush()` / `close()`.
-- `org.apache.juneau.rest.sse.SseBroadcaster` — pub/sub fan-out bean. Methods:
`subscribe(String id)` returns a `SseSubscription` (a `BlockingQueue<SseEvent>`
wrapper with `Iterator<SseEvent>` and `close()`); `publish(SseEvent)` enqueues
to every active subscriber; per-subscriber bounded queue with a configurable
overflow policy (default: drop-oldest with a debug log).
-- `org.apache.juneau.rest.sse.SseHeartbeat` — `ScheduledFuture`-driven `:
ping\n\n` emitter; defaults to 15s cadence; cancellable via the returned handle.
-- New `@RestGet`-friendly parameter `SseBroadcaster` / `SseSubscription`
injection through the existing `RestOpArg` SPI (sibling of the existing
`HttpServletRequestArgs`, `RestRequestArgs`).
-- Demo endpoint added under `juneau-examples/juneau-examples-rest` exercising
a broadcaster + heartbeat (the SSE-marshalling demo `SseDemoResource` is the
obvious place to grow into a broadcaster example).
-- Tests in `juneau-utest` covering: single-subscriber drain, multi-subscriber
fan-out, slow-subscriber overflow, heartbeat insertion, client-disconnect
cleanup (the writer throws — broadcaster must release the subscription).
-- Release-notes entry under `### juneau-rest-server` in the active
release-notes file; new topic page (`pages/topics/10.08.RestServerSse.md` or
similar).
-
-**Explicitly out of scope (v1):**
-
-- Reactive-Streams `Publisher<SseEvent>` return types from `@RestOp` —
orthogonal to TODO-70 (`CompletableFuture` + virtual-threads); the brainstorm
marked it as "transport-layer change, not marshalling change."
-- `Last-Event-ID` resume support (client-side concern; the bean already
carries `id`; server-side resume would need a per-resource event journal —
defer).
-- Cross-JVM broadcasting (Redis / Kafka backplane). The `SseBroadcaster` SPI
should be split into interface + in-memory impl so an external-backplane impl
can be a sibling sub-module later, but no external impl in v1.
-- Client-side SSE consumer ergonomics — `juneau-rest-client` already gets
`SseEventReader` from the marshall module; if more is wanted, file a separate
TODO.
-
-## Phased steps
-
-### Phase 0 — confirm seams (read-only)
-
-1. Re-read
`juneau-core/juneau-marshall/src/main/java/org/apache/juneau/sse/SseSerializerSession.java`
to confirm the per-event flush contract — it already calls `Writer.flush()`
per event, which is what makes the broadcaster path safe.
-2. Inspect
`juneau-rest/juneau-rest-server/src/main/java/org/apache/juneau/rest/processor/SerializedPojoProcessor.java`
to confirm the response is **not** drained / closed by the framework when a
method calls `res.flushBuffer()` and writes directly — this is the seam the
`SseResponseSupport` rides on. (Today's `SseDemoResource` proves the pattern
works.)
-3. Confirm `RestResponse.getNegotiatedWriter()` returns the same `Writer`
`SseSerializerSession` operates on — yes, via `FinishablePrintWriter`.
-
-### Phase 1 — `SseResponseSupport` (no broadcaster)
-
-1. Add the new package `org.apache.juneau.rest.sse` in `juneau-rest-server`.
Add `SseResponseSupport` with the fluent surface, plus the `RestResponse.sse()`
accessor.
-2. Add `SseHeartbeat` (a small `Runnable` that writes a comment + flushes) and
wire the optional scheduler bean — when absent, `heartbeat(Duration)` is a
no-op (no scheduler ⇒ no heartbeat).
-3. Tests:
- - `SseResponseSupport_Test` — single-event emit, multi-event emit, comment
write, charset is UTF-8, content-type is `text/event-stream` exactly.
- - `SseHeartbeat_Test` — heartbeat fires at the configured cadence, cancels
on `close()`.
-
-### Phase 2 — `SseBroadcaster` + arg injection
-
-1. Add `SseBroadcaster` + `SseSubscription`. Default impl is in-memory with
per-subscriber `LinkedBlockingQueue<SseEvent>` and a configurable bound
(default: 1024 events).
-2. Add `SseBroadcasterArg` / `SseSubscriptionArg` `RestOpArg` implementations
so handlers can take them as parameters.
-3. Tests:
- - `SseBroadcaster_Test` — pub/sub fan-out, slow-subscriber overflow policy,
subscriber-disconnect cleanup, concurrent publisher / subscriber smoke.
- - `Rest_SseBroadcaster_IT_Test` (in `juneau-utest`) — `MockRestClient`
against a `@RestGet` using the broadcaster; assert both subscribers receive
every published event in order.
-
-### Phase 3 — demo + docs
-
-1. Update `juneau-examples/juneau-examples-rest/.../SseDemoResource.java` to
demonstrate the broadcaster pattern (keep the existing `Stream<SseEvent>`
example; add a new endpoint that uses `SseBroadcaster`).
-2. New doc page `juneau-docs/pages/topics/10.08.RestServerSse.md` (slug
`RestServerSse`) covering both the simple `Stream<SseEvent>` form and the
broadcaster form. Sidebar entry.
-3. Release-notes entry under `### juneau-rest-server`.
-
-## Acceptance criteria
-
-- [ ] `RestResponse.sse()` returns an `SseResponseSupport` that sets
`Content-Type: text/event-stream`, disables response buffering, and exposes
`sendEvent(...)` / `comment(...)` / `flush()` / `close()`.
-- [ ] `SseBroadcaster.publish(event)` reaches every active subscriber, in
order, with no drops below the per-subscriber bound. Slow-subscriber overflow
drops the oldest event and logs at `DEBUG`.
-- [ ] `SseHeartbeat` at a 15s cadence inserts `: ping\n\n` between events
without corrupting the SSE stream (verified by `SseEventReader` parsing the
captured output).
-- [ ] Client disconnect → broadcaster releases the subscription within ≤ 1
heartbeat interval (no leak in long-soak test).
-- [ ] Demo endpoint in `juneau-examples-rest` is observable via `curl -N` and
shows live event delivery.
-- [ ] Coverage ≥ 90% on the new package. Full `./scripts/test.py` green.
-- [ ] Release-notes + topic page + sidebar entry shipped.
-
-## Open questions
-
-1. **Mix-in vs accessor.** `RestResponse.sse()` accessor (recommended) vs a
separate `SseRestResponse extends RestResponse` mix-in. Accessor keeps the API
surface small and avoids subclass churn.
-2. **Overflow policy default.** Drop-oldest (recommended) vs drop-newest vs
block-publisher. Drop-oldest matches what most SSE consumers expect.
-3. **Per-subscriber queue bound default.** 1024 events / ~1MB worst case.
Configurable per subscriber; configurable per broadcaster via `@Bean
SseBroadcasterConfig`.
-4. **Heartbeat cadence default.** 15s — under Nginx's default 30s idle timeout
and AWS ALB's 60s default. Configurable.
-5. **External-backplane SPI surface.** Should v1 ship `SseBroadcaster` as an
interface (recommended) or a concrete class with hooks? Interface keeps the
door open for Redis / Kafka backplanes as separate sub-modules without breaking
changes.
-6. **Naming.** `SseBroadcaster` or `SseEventBus`? Recommend `SseBroadcaster` —
closer to the spec's "broadcasting" language.
-
-## Risks
-
-- **Servlet container buffering.** Some containers buffer the response despite
`flushBuffer()`. Mitigation: the existing SSE demo proves it works in Jetty
(per `FINISHED-46-*` verification with `curl -N`); call out Tomcat behavior in
the docs if needed.
-- **Thread leak on client disconnect.** A subscriber that never reads will pin
a queue. Mitigation: per-subscriber bounded queue + a "no read in N heartbeats
⇒ evict" timer.
-- **Coupling with TODO-67 (observability).** If TODO-67 introduces
`X-Request-Id` propagation and broadcaster subscriptions are keyed by request
id, the two need to align on the id source. Recommend
`SseBroadcaster.subscribe(String id)` accepts any string — id source is the
caller's concern.
-- **Memory pressure under broadcast storms.** A 10k-subscriber broadcaster
with 1024-event queues each can balloon to 10M events × event size. Document;
configurable bound.
-
-## Related work
-
-- `todo/FINISHED-46-juneau-marshall-sse.md` — the marshalling-side SSE
primitives this TODO consumes.
-- `todo/FINISHED-31-inject-aware-microservice.md` — `WritableBeanStore`
auto-wiring for `@Bean SseBroadcaster`.
-- `todo/FINISHED-33-dynamic-rest-children.md` — useful for mounting SSE demo /
metrics-stream resources at runtime.
-- `todo/TODO-67-observability-micrometer-otel.md` (sibling) — `X-Request-Id`
is the natural broadcaster-subscription key.
-- `todo/TODO-70-async-completablefuture-virtual-threads.md` (sibling) —
`Publisher<SseEvent>` return-type support lives there, not here.
diff --git a/todo/TODO.md b/todo/TODO.md
index 669e0748e6..a98604bef7 100644
--- a/todo/TODO.md
+++ b/todo/TODO.md
@@ -6,8 +6,6 @@
- [TODO-37] - Agent instruction consolidation.
-- [TODO-62] Server-side SSE helpers (broadcaster, per-event flush, heartbeat).
See `todo/TODO-62-sse-server-helpers.md`.
-
- [TODO-63] OpenAPI 3.1 emission + bundled Swagger UI / Redoc auto-mount. See
`todo/TODO-63-openapi-3.1-emission.md`.
- [TODO-64] Conditional-GET / ETag / `If-Modified-Since` helpers on
`RestResponse`. See `todo/TODO-64-etag-conditional-get-helpers.md`.
