Implement keepalive messages for SSE connections

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.
  * <p>
- * 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<String[]> 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<SseEvent> 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();
     }
 

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