nik/signal-cli
1
0
Fork 0
mirror of https://github.com/AsamK/signal-cli.git synced 2026-11-02 09:59:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Last-Event-ID replay and configurable keep-alive to the SSE endpoint

Browse Source 
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
This commit is contained in:
Romain Boulay 2026-09-28 19:32:56 +02:00
parent 76fb9768aa
commit 56f8f3e05b
8 changed files with 428 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
man/signal-cli-jsonrpc.5.adoc
View File

@ -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.

4
man/signal-cli.1.adoc
View File

@ -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.

30
src/main/java/org/asamk/signal/commands/DaemonCommand.java
View File

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

63
src/main/java/org/asamk/signal/http/HttpServerHandler.java
View File

@ -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<String> 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<Pair<Manager, Manager.ReceiveMessageHandler>> subscribeReceiveHandlers(
final List<Manager> 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<Pair<Manager, Manager.ReceiveMessageHandler>> subscribeReceiveHandlers(final List<Manager> 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<String> buildAllowedHosts(final InetSocketAddress address) {
final var s = new HashSet<String>();
final var host = address == null ? null : address.getHostString();

153
src/main/java/org/asamk/signal/http/SseEventBuffer.java Normal file
View File

@ -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.
* <p>
* 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<Manager, Manager.ReceiveMessageHandler> recorders = new HashMap<>();
private final ArrayDeque<Event> events = new ArrayDeque<>();
private final List<Client> 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<Manager> 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<Client> 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<Manager> managers;
private final ServerSentEventSender sender;
private final Runnable onClosed;
private Client(
final Collection<Manager> 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();
}
}
}
}

3
src/test/java/org/asamk/signal/http/SseInitialFlushTest.java
View File

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

201
src/test/java/org/asamk/signal/http/SseReplayTest.java Normal file
View File

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

4
src/test/java/org/asamk/signal/testutil/ManagerMock.java
View File

@ -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<Manager.ReceiveMessageHandler> receiveHandlers = new ArrayList<>();
// Written by server threads while tests read it
public final List<Manager.ReceiveMessageHandler> receiveHandlers = new CopyOnWriteArrayList<>();
public final AtomicInteger addReceiveHandlerCount = new AtomicInteger(0);
public final AtomicInteger removeReceiveHandlerCount = new AtomicInteger(0);
}