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..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 @@ -30,23 +30,29 @@ import java.nio.charset.StandardCharsets; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import lombok.SneakyThrows; /** * 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; - private final BlockingQueue queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); + // 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; private volatile Thread runningThread; @@ -72,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(); } } @@ -86,16 +92,20 @@ 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 { - 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]); + send(out, event); } } finally { close(); @@ -109,13 +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'); + 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); } }