From 56f8f3e05b5c34bd6c6db5f742495baf1f870e54 Mon Sep 17 00:00:00 2001 From: Romain Boulay <169323+RomainBoulay@users.noreply.github.com> Date: Mon, 28 Sep 2026 19:32:56 +0200 Subject: [PATCH] Add Last-Event-ID replay and configurable keep-alive to the SSE endpoint Each event of /api/v1/events now has an id. The last 1000 events are kept in memory, and a client that reconnects with a Last-Event-ID header first gets the events it missed. The keep-alive interval can be set with the new daemon option --sse-keepalive-interval. Closes #2132 --- man/signal-cli-jsonrpc.5.adoc | 5 + man/signal-cli.1.adoc | 4 + .../asamk/signal/commands/DaemonCommand.java | 30 ++- .../asamk/signal/http/HttpServerHandler.java | 63 +++--- .../org/asamk/signal/http/SseEventBuffer.java | 153 +++++++++++++ .../signal/http/SseInitialFlushTest.java | 3 +- .../org/asamk/signal/http/SseReplayTest.java | 201 ++++++++++++++++++ .../asamk/signal/testutil/ManagerMock.java | 4 +- 8 files changed, 428 insertions(+), 35 deletions(-) create mode 100644 src/main/java/org/asamk/signal/http/SseEventBuffer.java create mode 100644 src/test/java/org/asamk/signal/http/SseReplayTest.java diff --git a/man/signal-cli-jsonrpc.5.adoc b/man/signal-cli-jsonrpc.5.adoc index b9162925..8d86ba2f 100644 --- a/man/signal-cli-jsonrpc.5.adoc +++ b/man/signal-cli-jsonrpc.5.adoc @@ -38,6 +38,11 @@ With `--http` signal-cli exposes three endpoints; * GET /api/v1/events : Returns a Server-Sent Events (SSE) stream of incoming messages * GET /api/v1/check : Responds with 200 OK if daemon is running +Each event of `/api/v1/events` has an `id`. +signal-cli keeps the last 1000 events in memory. +A client that reconnects with a `Last-Event-ID` header first receives the events after that id. +If the id is unknown, e.g. because signal-cli was restarted, all kept events are sent. + == Basic usage In JSON-RPC mode, signal-cli will read requests from stdin. diff --git a/man/signal-cli.1.adoc b/man/signal-cli.1.adoc index 963d3c2c..c2a8d7c6 100644 --- a/man/signal-cli.1.adoc +++ b/man/signal-cli.1.adoc @@ -1105,6 +1105,10 @@ Expose a JSON-RPC interface as http endpoint (default localhost:8080). The JSON-RPC endpoint is `/api/v1/rpc`. + See **signal-cli-jsonrpc**(5) for info on the JSON-RPC interface. +*--sse-keepalive-interval* SECONDS:: +Seconds between keep-alive messages on the http events endpoint `/api/v1/events` (default 15). +A shorter interval lets signal-cli notice sooner that a client has disconnected. + *--dbus*:: Export DBus interface on user bus. + See **signal-cli-dbus**(5) for info on the dbus interface. diff --git a/src/main/java/org/asamk/signal/commands/DaemonCommand.java b/src/main/java/org/asamk/signal/commands/DaemonCommand.java index 3bb05dea..fd64a9f6 100644 --- a/src/main/java/org/asamk/signal/commands/DaemonCommand.java +++ b/src/main/java/org/asamk/signal/commands/DaemonCommand.java @@ -30,6 +30,7 @@ import java.net.InetSocketAddress; import java.net.UnixDomainSocketAddress; import java.nio.channels.Channel; import java.nio.channels.ServerSocketChannel; +import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.SequencedCollection; @@ -69,6 +70,10 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand { .nargs("?") .setConst("localhost:8080") .help("Expose a JSON-RPC interface as http endpoint (default localhost:8080)."); + subparser.addArgument("--sse-keepalive-interval") + .type(int.class) + .setDefault(15) + .help("Seconds between keep-alive messages on the http events endpoint (default 15)."); subparser.addArgument("--no-receive-stdout") .help("Don’t print received messages to stdout.") .action(Arguments.storeTrue()); @@ -188,7 +193,11 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand { final var httpAddress = ns.getString("http"); if (httpAddress != null) { final var address = IOUtils.parseInetSocketAddress(httpAddress); - daemonHandler.runHttp(address); + final var keepAliveInterval = ns.getInt("sse-keepalive-interval"); + if (keepAliveInterval < 1) { + throw new UserErrorException("--sse-keepalive-interval must be at least 1 second."); + } + daemonHandler.runHttp(address, Duration.ofSeconds(keepAliveInterval)); } final var isDbusSystem = Boolean.TRUE.equals(ns.getBoolean("dbus-system")); @@ -235,7 +244,10 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand { public abstract void runDbus(boolean isDbusSystem, final String busname) throws CommandException; - public abstract void runHttp(InetSocketAddress address) throws CommandException; + public abstract void runHttp( + InetSocketAddress address, + Duration keepAliveInterval + ) throws CommandException; protected final void runSocket(final SocketHandler socketHandler) { socketHandler.init(); @@ -291,8 +303,11 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand { } @Override - public void runHttp(InetSocketAddress address) throws CommandException { - runHttp(new HttpServerHandler(address, m)); + public void runHttp( + final InetSocketAddress address, + final Duration keepAliveInterval + ) throws CommandException { + runHttp(new HttpServerHandler(address, keepAliveInterval, m)); } } @@ -316,8 +331,11 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand { } @Override - public void runHttp(final InetSocketAddress address) throws CommandException { - runHttp(new HttpServerHandler(address, c)); + public void runHttp( + final InetSocketAddress address, + final Duration keepAliveInterval + ) throws CommandException { + runHttp(new HttpServerHandler(address, keepAliveInterval, c)); } } } diff --git a/src/main/java/org/asamk/signal/http/HttpServerHandler.java b/src/main/java/org/asamk/signal/http/HttpServerHandler.java index 11dd447f..fc5c9e0b 100644 --- a/src/main/java/org/asamk/signal/http/HttpServerHandler.java +++ b/src/main/java/org/asamk/signal/http/HttpServerHandler.java @@ -5,13 +5,13 @@ import com.sun.net.httpserver.HttpExchange; import com.sun.net.httpserver.HttpServer; import org.asamk.signal.commands.Commands; -import org.asamk.signal.json.JsonReceiveMessageHandler; import org.asamk.signal.jsonrpc.JsonRpcReader; import org.asamk.signal.jsonrpc.JsonRpcResponse; import org.asamk.signal.jsonrpc.JsonRpcSender; import org.asamk.signal.jsonrpc.SignalJsonRpcCommandHandler; import org.asamk.signal.manager.Manager; import org.asamk.signal.manager.MultiAccountManager; +import org.asamk.signal.manager.api.MessageEnvelope; import org.asamk.signal.manager.api.Pair; import org.asamk.signal.util.Util; import org.slf4j.Logger; @@ -19,6 +19,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.HashSet; import java.util.List; import java.util.Locale; @@ -34,6 +35,7 @@ public class HttpServerHandler implements AutoCloseable { private final ObjectMapper objectMapper = Util.createJsonObjectMapper(); private final InetSocketAddress address; + private final Duration keepAliveInterval; private final SignalJsonRpcCommandHandler commandHandler; private final MultiAccountManager c; @@ -41,17 +43,24 @@ public class HttpServerHandler implements AutoCloseable { private HttpServer server; private final AtomicBoolean shutdown = new AtomicBoolean(false); private final Set allowedHosts; + private final SseEventBuffer eventBuffer = new SseEventBuffer(objectMapper); - public HttpServerHandler(final InetSocketAddress address, final Manager m) { + public HttpServerHandler(final InetSocketAddress address, final Duration keepAliveInterval, final Manager m) { this.address = address; + this.keepAliveInterval = keepAliveInterval; commandHandler = new SignalJsonRpcCommandHandler(m, Commands::getCommand); this.c = null; this.m = m; this.allowedHosts = buildAllowedHosts(address); } - public HttpServerHandler(final InetSocketAddress address, final MultiAccountManager c) { + public HttpServerHandler( + final InetSocketAddress address, + final Duration keepAliveInterval, + final MultiAccountManager c + ) { this.address = address; + this.keepAliveInterval = keepAliveInterval; commandHandler = new SignalJsonRpcCommandHandler(c, Commands::getCommand); this.c = c; this.m = null; @@ -93,6 +102,7 @@ public class HttpServerHandler implements AutoCloseable { // Increase this delay when https://bugs.openjdk.org/browse/JDK-8304065 is fixed server.stop(2); server = null; + eventBuffer.close(); shutdown.set(false); } } @@ -199,22 +209,27 @@ public class HttpServerHandler implements AutoCloseable { // Flush HTTP response headers to the client immediately. // Without this, the JVM HttpServer buffers everything until a later write - // in the keep-alive loop (15 s), causing clients with shorter timeouts + // in the keep-alive loop, causing clients with shorter timeouts // (e.g. 10 s) to abort before receiving the initial response. httpExchange.getResponseBody().flush(); final var shouldStop = new AtomicBoolean(false); - final var handlers = subscribeReceiveHandlers(managers, sender, () -> { - shouldStop.set(true); - synchronized (this) { - this.notifyAll(); - } - }); + managers.forEach(eventBuffer::record); + final var client = eventBuffer.connect(managers, + sender, + httpExchange.getRequestHeaders().getFirst("Last-Event-ID"), + () -> { + shouldStop.set(true); + synchronized (this) { + this.notifyAll(); + } + }); + final var handlers = subscribeReceiveHandlers(managers); try { while (true) { synchronized (this) { - wait(15_000); + wait(keepAliveInterval.toMillis()); } if (shouldStop.get() || shutdown.get()) { break; @@ -227,6 +242,7 @@ public class HttpServerHandler implements AutoCloseable { } } } finally { + eventBuffer.disconnect(client); for (final var pair : handlers) { unsubscribeReceiveHandler(pair); } @@ -276,19 +292,17 @@ public class HttpServerHandler implements AutoCloseable { throw new AssertionError("Unreachable state"); } - private List> subscribeReceiveHandlers( - final List managers, - final ServerSentEventSender sender, - Callable unsubscribe - ) { + /** + * The events are sent by eventBuffer, these handlers only keep the managers receiving while the client is connected. + */ + private List> subscribeReceiveHandlers(final List managers) { return managers.stream().map(m1 -> { - final var receiveMessageHandler = new JsonReceiveMessageHandler(m1, s -> { - try { - sender.sendEvent(null, "receive", List.of(objectMapper.writeValueAsString(s))); - } catch (IOException e) { - unsubscribe.call(); + // An anonymous class instead of a lambda, as every connection needs its own handler instance + final var receiveMessageHandler = new Manager.ReceiveMessageHandler() { + @Override + public void handleMessage(final MessageEnvelope envelope, final Throwable e) { } - }); + }; m1.addReceiveHandler(receiveMessageHandler); return new Pair<>(m1, (Manager.ReceiveMessageHandler) receiveMessageHandler); }).toList(); @@ -300,11 +314,6 @@ public class HttpServerHandler implements AutoCloseable { m.removeReceiveHandler(handler); } - private interface Callable { - - void call(); - } - private Set buildAllowedHosts(final InetSocketAddress address) { final var s = new HashSet(); final var host = address == null ? null : address.getHostString(); diff --git a/src/main/java/org/asamk/signal/http/SseEventBuffer.java b/src/main/java/org/asamk/signal/http/SseEventBuffer.java new file mode 100644 index 00000000..00355202 --- /dev/null +++ b/src/main/java/org/asamk/signal/http/SseEventBuffer.java @@ -0,0 +1,153 @@ +package org.asamk.signal.http; + +import com.fasterxml.jackson.databind.ObjectMapper; + +import org.asamk.signal.json.JsonReceiveMessageHandler; +import org.asamk.signal.manager.Manager; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Keeps the most recent SSE events and sends new events to the connected clients. + * A client that reconnects with a Last-Event-ID header first gets the buffered events it missed. + *

+ * Event ids have the form "{start time}-{sequence}", so an id from an earlier daemon run never matches. + */ +final class SseEventBuffer implements AutoCloseable { + + private static final Logger logger = LoggerFactory.getLogger(SseEventBuffer.class); + + static final int MAX_EVENTS = 1000; + + private final ObjectMapper objectMapper; + private final String idPrefix = System.currentTimeMillis() + "-"; + private final Map recorders = new HashMap<>(); + private final ArrayDeque events = new ArrayDeque<>(); + private final List clients = new ArrayList<>(); + private long nextSequence = 1; + + SseEventBuffer(final ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + } + + /** + * Start recording the messages of the given account. + * The handler is registered as a weak listener, so it never starts receiving by itself. + */ + void record(final Manager m) { + synchronized (recorders) { + if (recorders.containsKey(m)) { + return; + } + final var handler = new JsonReceiveMessageHandler(m, s -> add(m, s)); + m.addReceiveHandler(handler, true); + recorders.put(m, handler); + } + } + + /** + * Register a client for the given accounts. + * If lastEventId is set, the buffered events after it are sent first. + * An unknown id (too old, or from an earlier run) sends all buffered events. + */ + synchronized Client connect( + final Collection managers, + final ServerSentEventSender sender, + final String lastEventId, + final Runnable onClosed + ) { + final var client = new Client(managers, sender, onClosed); + if (lastEventId != null) { + final var lastSequence = getSequence(lastEventId); + for (final var event : events) { + if (event.sequence() > lastSequence && managers.contains(event.manager())) { + client.send(event); + } + } + } + clients.add(client); + return client; + } + + synchronized void disconnect(final Client client) { + clients.remove(client); + } + + @Override + public void close() { + synchronized (recorders) { + recorders.forEach(Manager::removeReceiveHandler); + recorders.clear(); + } + } + + private void add(final Manager m, final Object message) { + final String data; + try { + data = objectMapper.writeValueAsString(message); + } catch (IOException e) { + logger.warn("Failed to serialize received message, ignoring", e); + return; + } + + final Event event; + final List receivers; + synchronized (this) { + event = new Event(nextSequence++, m, data); + events.addLast(event); + if (events.size() > MAX_EVENTS) { + events.removeFirst(); + } + receivers = clients.stream().filter(c -> c.managers.contains(m)).toList(); + } + // Sent outside the lock, so a slow client doesn't block the other accounts + for (final var client : receivers) { + client.send(event); + } + } + + private long getSequence(final String eventId) { + if (eventId.startsWith(idPrefix)) { + try { + return Long.parseLong(eventId.substring(idPrefix.length())); + } catch (NumberFormatException ignored) { + } + } + return 0; + } + + private record Event(long sequence, Manager manager, String data) {} + + final class Client { + + private final Collection managers; + private final ServerSentEventSender sender; + private final Runnable onClosed; + + private Client( + final Collection managers, + final ServerSentEventSender sender, + final Runnable onClosed + ) { + this.managers = managers; + this.sender = sender; + this.onClosed = onClosed; + } + + private void send(final Event event) { + try { + sender.sendEvent(idPrefix + event.sequence(), "receive", List.of(event.data())); + } catch (IOException e) { + onClosed.run(); + } + } + } +} diff --git a/src/test/java/org/asamk/signal/http/SseInitialFlushTest.java b/src/test/java/org/asamk/signal/http/SseInitialFlushTest.java index 472a8fef..2153a399 100644 --- a/src/test/java/org/asamk/signal/http/SseInitialFlushTest.java +++ b/src/test/java/org/asamk/signal/http/SseInitialFlushTest.java @@ -10,6 +10,7 @@ import java.net.HttpURLConnection; import java.net.InetSocketAddress; import java.net.ServerSocket; import java.net.URI; +import java.time.Duration; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -44,7 +45,7 @@ class SseInitialFlushTest { void setUp() throws Exception { port = freePort(); final Manager manager = ManagerMock.create("+10000000000"); - handler = new HttpServerHandler(new InetSocketAddress("127.0.0.1", port), manager); + handler = new HttpServerHandler(new InetSocketAddress("127.0.0.1", port), Duration.ofSeconds(15), manager); handler.init(); } diff --git a/src/test/java/org/asamk/signal/http/SseReplayTest.java b/src/test/java/org/asamk/signal/http/SseReplayTest.java new file mode 100644 index 00000000..dd65f594 --- /dev/null +++ b/src/test/java/org/asamk/signal/http/SseReplayTest.java @@ -0,0 +1,201 @@ +package org.asamk.signal.http; + +import org.asamk.signal.testutil.ManagerMock; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.io.BufferedReader; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStreamReader; +import java.net.HttpURLConnection; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Duration; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests the event ids, the replay of missed events for a client that reconnects with a + * Last-Event-ID header, and the configurable keep-alive interval of GET /api/v1/events. + */ +class SseReplayTest { + + private final ManagerMock.State state = new ManagerMock.State(); + private HttpServerHandler handler; + private int port; + + private void start(final Duration keepAliveInterval) throws Exception { + try (var ss = new ServerSocket(0)) { + port = ss.getLocalPort(); + } + handler = new HttpServerHandler(new InetSocketAddress("127.0.0.1", port), + keepAliveInterval, + ManagerMock.create("+10000000000", state)); + handler.init(); + } + + @AfterEach + void tearDown() { + if (handler != null) { + handler.close(); + } + } + + @Test + void eventsHaveIds() throws Exception { + start(Duration.ofSeconds(15)); + try (var client = connect(null, 2)) { + receive("one"); + + final var event = client.next(); + assertNotNull(event.id()); + assertTrue(event.data().contains("\"one\"")); + } + } + + @Test + void reconnectWithLastEventIdReplaysMissedEvents() throws Exception { + start(Duration.ofSeconds(15)); + final String firstId; + try (var client = connect(null, 2)) { + receive("one"); + receive("two"); + firstId = client.next().id(); + client.next(); + } + receive("three"); + + try (var client = connect(firstId, 3)) { + assertTrue(client.next().data().contains("\"two\"")); + assertTrue(client.next().data().contains("\"three\"")); + + receive("four"); + assertTrue(client.next().data().contains("\"four\"")); + } + } + + @Test + void unknownLastEventIdReplaysAllBufferedEvents() throws Exception { + start(Duration.ofSeconds(15)); + try (var client = connect(null, 2)) { + receive("one"); + receive("two"); + client.next(); + client.next(); + } + + try (var client = connect("0-1", 3)) { + assertTrue(client.next().data().contains("\"one\"")); + assertTrue(client.next().data().contains("\"two\"")); + } + } + + @Test + void connectWithoutLastEventIdReplaysNothing() throws Exception { + start(Duration.ofSeconds(15)); + try (var client = connect(null, 2)) { + receive("one"); + client.next(); + } + + try (var client = connect(null, 3)) { + receive("two"); + assertTrue(client.next().data().contains("\"two\"")); + } + } + + @Test + void keepAliveIntervalIsConfigurable() throws Exception { + start(Duration.ofSeconds(1)); + try (var client = connect(null, 2)) { + // The read timeout is 5 s, the default interval of 15 s would fail here + assertEquals(":", client.nextLine()); + } + } + + /** + * Passes a message to the registered receive handlers, like the manager's receive thread does. + */ + private void receive(final String message) { + final var exception = new RuntimeException(message); + for (final var receiveHandler : state.receiveHandlers) { + receiveHandler.handleMessage(null, exception); + } + } + + /** + * Connects and waits until the server has registered the client, i.e. until the manager has + * seen the given total number of addReceiveHandler calls. + */ + private SseClient connect(final String lastEventId, final int addReceiveHandlerCount) throws Exception { + final var client = new SseClient(port, lastEventId); + final var deadline = System.nanoTime() + Duration.ofSeconds(5).toNanos(); + while (state.addReceiveHandlerCount.get() < addReceiveHandlerCount) { + if (System.nanoTime() > deadline) { + client.close(); + throw new AssertionError("Client was not registered in time"); + } + Thread.sleep(10); + } + return client; + } + + private record Event(String id, String data) {} + + private static final class SseClient implements AutoCloseable { + + private final HttpURLConnection connection; + private final BufferedReader reader; + + SseClient(final int port, final String lastEventId) throws IOException { + final var url = URI.create("http://127.0.0.1:" + port + "/api/v1/events").toURL(); + connection = (HttpURLConnection) url.openConnection(); + connection.setRequestProperty("Accept", "text/event-stream"); + if (lastEventId != null) { + connection.setRequestProperty("Last-Event-ID", lastEventId); + } + connection.setConnectTimeout(2_000); + connection.setReadTimeout(5_000); + assertEquals(200, connection.getResponseCode()); + reader = new BufferedReader(new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8)); + } + + String nextLine() throws IOException { + final var line = reader.readLine(); + if (line == null) { + throw new EOFException(); + } + return line; + } + + /** + * Reads the next event, skipping keep-alive comments. + */ + Event next() throws IOException { + String id = null; + String data = null; + while (true) { + final var line = nextLine(); + if (line.isEmpty()) { + if (data != null) { + return new Event(id, data); + } + } else if (line.startsWith("id:")) { + id = line.substring("id:".length()); + } else if (line.startsWith("data:")) { + data = line.substring("data:".length()); + } + } + } + + @Override + public void close() { + connection.disconnect(); + } + } +} diff --git a/src/test/java/org/asamk/signal/testutil/ManagerMock.java b/src/test/java/org/asamk/signal/testutil/ManagerMock.java index d10bdd60..7022e967 100644 --- a/src/test/java/org/asamk/signal/testutil/ManagerMock.java +++ b/src/test/java/org/asamk/signal/testutil/ManagerMock.java @@ -9,6 +9,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicInteger; public final class ManagerMock { @@ -24,7 +25,8 @@ public final class ManagerMock { public final AtomicInteger addCallEventListenerCount = new AtomicInteger(0); public final AtomicInteger removeCallEventListenerCount = new AtomicInteger(0); - public final List receiveHandlers = new ArrayList<>(); + // Written by server threads while tests read it + public final List receiveHandlers = new CopyOnWriteArrayList<>(); public final AtomicInteger addReceiveHandlerCount = new AtomicInteger(0); public final AtomicInteger removeReceiveHandlerCount = new AtomicInteger(0); }