From 7c89c5012e432bea26ec6a3d1e4895404fde09ab Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Fri, 7 Aug 2026 00:01:25 -0400 Subject: [PATCH 1/2] Implement SSE keepalive messages Implements sending a comment (`:`) every 30 seconds if no other event has been enqueued to prevent the connection from timing out and being closed. --- .../bluemap/common/web/SseConnection.java | 26 +++++++++++++++++-- 1 file changed, 24 insertions(+), 2 deletions(-) diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java index af805dc0a..ce8d64564 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java @@ -30,6 +30,7 @@ import java.nio.charset.StandardCharsets; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import lombok.SneakyThrows; @@ -46,6 +47,9 @@ public class SseConnection implements Closeable { // how many messages can be queued up for sending before being dropped private static final int QUEUE_CAPACITY = 64; + // how long to wait for an event before sending a keepalive + private static final long KEEPALIVE_INTERVAL_SECONDS = 30; + private final BlockingQueue queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); private volatile boolean closed = false; private volatile Runnable onClose; @@ -90,12 +94,20 @@ public void run(OutputStream out) throws IOException { try { while (!closed) { try { - event = queue.take(); + if (KEEPALIVE_INTERVAL_SECONDS > 0){ + event = queue.poll(KEEPALIVE_INTERVAL_SECONDS, TimeUnit.SECONDS); + } else { + event = queue.take(); + } } catch (InterruptedException _) { runningThread.interrupt(); break; } - send(out, event[0], event[1]); + if (event == null) { + sendKeepalive(out); + } else { + send(out, event[0], event[1]); + } } } finally { close(); @@ -119,6 +131,16 @@ private void send(OutputStream out, String eventType, String data) throws IOExce out.flush(); } + /** + * Write an SSE comment to the stream and flush it to keep the connection alive + * + * @throws IOException if the client has disconnected + */ + private void sendKeepalive(OutputStream out) throws IOException { + writeLine(out, ":"); + out.flush(); + } + @Override public synchronized void close() { if (closed) return; From 5c695016310288897d6cd583833b0418e826896d Mon Sep 17 00:00:00 2001 From: Carey Metcalfe Date: Fri, 7 Aug 2026 00:08:06 -0400 Subject: [PATCH 2/2] Refactor SSE events into a record This allows `SseConnectionManager.broadcast` to create a single object and give it to all the `SseConnection` instances instead of `SseConnection.enqueue` having to instantiate a new `String[]` each time. It also makes the logic that sends keepalives cleaner. --- .../bluemap/common/web/SseConnection.java | 41 ++++++++----------- .../common/web/SseConnectionManager.java | 3 +- 2 files changed, 19 insertions(+), 25 deletions(-) diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java index ce8d64564..bb43b783d 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnection.java @@ -37,20 +37,22 @@ /** * Represents a single Server-Sent Events (SSE) connection. *

- * Events can be queued via {@link #enqueue(String, String)} without blocking. + * Events can be queued via {@link #enqueue(SseEvent)} without blocking. * Call {@link #run(OutputStream)} on the thread that owns the connection's output-stream (e.g. * the HTTP connection's thread) to deliver queued events to it. This will block the calling thread * until the connection is closed. */ public class SseConnection implements Closeable { + public record SseEvent(String type, String data) {} + // how many messages can be queued up for sending before being dropped private static final int QUEUE_CAPACITY = 64; // how long to wait for an event before sending a keepalive private static final long KEEPALIVE_INTERVAL_SECONDS = 30; - private final BlockingQueue queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); + private final BlockingQueue queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); private volatile boolean closed = false; private volatile Runnable onClose; private volatile Thread runningThread; @@ -76,9 +78,9 @@ public synchronized boolean setOnClose(Runnable onClose) { * If this connection's queue is full (due to a slowly-reading client), the event is * silently dropped and the connection will be closed. */ - public void enqueue(String eventType, String data) { + public void enqueue(SseEvent event) { if (closed) return; - if (!queue.offer(new String[]{eventType, data})) { + if (!queue.offer(event)) { close(); } } @@ -90,7 +92,7 @@ public void enqueue(String eventType, String data) { */ public void run(OutputStream out) throws IOException { runningThread = Thread.currentThread(); - String[] event; + SseEvent event; try { while (!closed) { try { @@ -103,11 +105,7 @@ public void run(OutputStream out) throws IOException { runningThread.interrupt(); break; } - if (event == null) { - sendKeepalive(out); - } else { - send(out, event[0], event[1]); - } + send(out, event); } } finally { close(); @@ -121,23 +119,18 @@ private void writeLine(OutputStream out, String line) { /** * Write one SSE event with optional data to the stream and flush it. + * Will write a comment (:) as a keepalive if {@code event} is {@code null}. * * @throws IOException if the client has disconnected */ - private void send(OutputStream out, String eventType, String data) throws IOException { - writeLine(out, "event: " + eventType); - data.lines().forEach(l -> writeLine(out, "data: " + l)); - out.write('\n'); - out.flush(); - } - - /** - * Write an SSE comment to the stream and flush it to keep the connection alive - * - * @throws IOException if the client has disconnected - */ - private void sendKeepalive(OutputStream out) throws IOException { - writeLine(out, ":"); + private void send(OutputStream out, SseEvent event) throws IOException { + if (event == null) { + writeLine(out, ":"); + } else { + writeLine(out, "event: " + event.type()); + event.data().lines().forEach(l -> writeLine(out, "data: " + l)); + out.write('\n'); + } out.flush(); } diff --git a/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnectionManager.java b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnectionManager.java index 910e82bd4..cebf9020f 100644 --- a/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnectionManager.java +++ b/common/src/main/java/de/bluecolored/bluemap/common/web/SseConnectionManager.java @@ -97,8 +97,9 @@ public void remove(SseConnection connection) { */ public void broadcast(String eventType, String data) { if (closed) return; + SseConnection.SseEvent event = new SseConnection.SseEvent(eventType, data); for (SseConnection conn : connections) { - conn.enqueue(eventType, data); + conn.enqueue(event); } }