();
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/main/java/org/asamk/signal/json/JsonAdminDelete.java b/src/main/java/org/asamk/signal/json/JsonAdminDelete.java
index c8586c9e..42b24c10 100644
--- a/src/main/java/org/asamk/signal/json/JsonAdminDelete.java
+++ b/src/main/java/org/asamk/signal/json/JsonAdminDelete.java
@@ -8,7 +8,11 @@ import io.micronaut.jsonschema.JsonSchema;
@JsonSchema(title = "AdminDelete")
public record JsonAdminDelete(
- @Deprecated String targetAuthor, String targetAuthorNumber, String targetAuthorUuid, long targetSentTimestamp
+ @Deprecated String targetAuthor,
+ String targetAuthorNumber,
+ String targetAuthorUuid,
+ long targetTimestamp,
+ @Deprecated long targetSentTimestamp
) {
static JsonAdminDelete from(MessageEnvelope.Data.AdminDelete adminDelete) {
@@ -16,9 +20,13 @@ public record JsonAdminDelete(
final var targetAuthor = address.getLegacyIdentifier();
final var targetAuthorNumber = address.number().orElse(null);
final var targetAuthorUuid = address.uuid().map(UUID::toString).orElse(null);
- final var targetSentTimestamp = adminDelete.targetSentTimestamp();
+ final var targetTimestamp = adminDelete.targetSentTimestamp();
- return new JsonAdminDelete(targetAuthor, targetAuthorNumber, targetAuthorUuid, targetSentTimestamp);
+ return new JsonAdminDelete(targetAuthor,
+ targetAuthorNumber,
+ targetAuthorUuid,
+ targetTimestamp,
+ targetTimestamp);
}
}
diff --git a/src/main/java/org/asamk/signal/json/JsonDataMessage.java b/src/main/java/org/asamk/signal/json/JsonDataMessage.java
index a053945b..f2ea5b5b 100644
--- a/src/main/java/org/asamk/signal/json/JsonDataMessage.java
+++ b/src/main/java/org/asamk/signal/json/JsonDataMessage.java
@@ -67,7 +67,7 @@ record JsonDataMessage(
.map(JsonPreview::from)
.toList() : null;
final var remoteDelete = dataMessage.remoteDeleteId().isPresent()
- ? new JsonRemoteDelete(dataMessage.remoteDeleteId().get())
+ ? new JsonRemoteDelete(dataMessage.remoteDeleteId().get(), dataMessage.remoteDeleteId().get())
: null;
final var attachments = !dataMessage.attachments().isEmpty() ? dataMessage.attachments()
.stream()
diff --git a/src/main/java/org/asamk/signal/json/JsonRemoteDelete.java b/src/main/java/org/asamk/signal/json/JsonRemoteDelete.java
index b1def8e8..1a870fa9 100644
--- a/src/main/java/org/asamk/signal/json/JsonRemoteDelete.java
+++ b/src/main/java/org/asamk/signal/json/JsonRemoteDelete.java
@@ -3,4 +3,4 @@ package org.asamk.signal.json;
import io.micronaut.jsonschema.JsonSchema;
@JsonSchema(title = "RemoteDelete")
-record JsonRemoteDelete(long timestamp) {}
+record JsonRemoteDelete(long targetTimestamp, @Deprecated long timestamp) {}
diff --git a/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java b/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java
index cb40f327..468637ca 100644
--- a/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java
+++ b/src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java
@@ -26,8 +26,7 @@ import java.util.stream.StreamSupport;
public class JsonRpcReader {
private static final Logger logger = LoggerFactory.getLogger(JsonRpcReader.class);
- private static final int MAX_JSON_STRING_LENGTH =
- ((ServiceConfig.MAX_ATTACHMENT_SIZE + 2) / 3) * 4 + 1024;
+ private static final int MAX_JSON_STRING_LENGTH = ((ServiceConfig.MAX_ATTACHMENT_SIZE + 2) / 3) * 4 + 1024;
private final JsonRpcSender jsonRpcSender;
private final ObjectMapper objectMapper;
@@ -50,9 +49,10 @@ public class JsonRpcReader {
private static ObjectMapper createObjectMapper() {
final var objectMapper = Util.createJsonObjectMapper();
- objectMapper.getFactory().setStreamReadConstraints(StreamReadConstraints.builder()
- .maxStringLength(MAX_JSON_STRING_LENGTH)
- .build());
+ objectMapper.getFactory()
+ .setStreamReadConstraints(StreamReadConstraints.builder()
+ .maxStringLength(MAX_JSON_STRING_LENGTH)
+ .build());
return objectMapper;
}
diff --git a/src/main/java/org/asamk/signal/jsonrpc/SignalJsonRpcDispatcherHandler.java b/src/main/java/org/asamk/signal/jsonrpc/SignalJsonRpcDispatcherHandler.java
index 166c822a..b959452b 100644
--- a/src/main/java/org/asamk/signal/jsonrpc/SignalJsonRpcDispatcherHandler.java
+++ b/src/main/java/org/asamk/signal/jsonrpc/SignalJsonRpcDispatcherHandler.java
@@ -31,6 +31,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -61,17 +62,31 @@ public class SignalJsonRpcDispatcherHandler {
public void handleConnection(final MultiAccountManager c) {
this.commandHandler = new SignalJsonRpcCommandHandler(c, this::getCommand);
+ final Consumer onManagerAddedReceiveHandler = m -> subscribeReceive(m, true);
+ final Consumer onManagerRemovedReceiveHandler = this::unsubscribeReceive;
+ final Consumer onManagerAddedReceiveSubscriptionsHandler = m -> receiveHandlers.forEach((subscriptionId, handlers) -> handlers.add(
+ createReceiveHandler(m, subscriptionId, false)));
+ final Consumer onManagerAddedCallEventSubscriptionsHandler = m -> callEventHandlers.forEach((subscriptionId, handlers) -> handlers.add(
+ createCallEventHandler(m, subscriptionId)));
+
if (!noReceiveOnStart) {
this.subscribeReceive(c.getManagers(), true);
- c.addOnManagerAddedHandler(m -> subscribeReceive(m, true));
- c.addOnManagerRemovedHandler(this::unsubscribeReceive);
+ c.addOnManagerAddedHandler(onManagerAddedReceiveHandler);
+ c.addOnManagerRemovedHandler(onManagerRemovedReceiveHandler);
}
- c.addOnManagerAddedHandler(m -> receiveHandlers.forEach((subscriptionId, handlers) -> handlers.add(
- createReceiveHandler(m, subscriptionId, false))));
- c.addOnManagerAddedHandler(m -> callEventHandlers.forEach((subscriptionId, handlers) -> handlers.add(
- createCallEventHandler(m, subscriptionId))));
+ c.addOnManagerAddedHandler(onManagerAddedReceiveSubscriptionsHandler);
+ c.addOnManagerAddedHandler(onManagerAddedCallEventSubscriptionsHandler);
- handleConnection();
+ try {
+ handleConnection();
+ } finally {
+ if (!noReceiveOnStart) {
+ c.removeOnManagerAddedHandler(onManagerAddedReceiveHandler);
+ c.removeOnManagerRemovedHandler(onManagerRemovedReceiveHandler);
+ }
+ c.removeOnManagerAddedHandler(onManagerAddedReceiveSubscriptionsHandler);
+ c.removeOnManagerAddedHandler(onManagerAddedCallEventSubscriptionsHandler);
+ }
}
public void handleConnection(final Manager m) {
diff --git a/src/main/resources/META-INF/native-image/org.asamk/signal-cli/reachability-metadata.json b/src/main/resources/META-INF/native-image/org.asamk/signal-cli/reachability-metadata.json
index 60ff4c40..4af3819a 100644
--- a/src/main/resources/META-INF/native-image/org.asamk/signal-cli/reachability-metadata.json
+++ b/src/main/resources/META-INF/native-image/org.asamk/signal-cli/reachability-metadata.json
@@ -1415,6 +1415,9 @@
}
]
},
+ {
+ "type": "java.math.BigDecimal"
+ },
{
"type": "java.math.BigInteger"
},
@@ -1771,6 +1774,9 @@
{
"type": "kotlin.Any"
},
+ {
+ "type": "kotlin.Array"
+ },
{
"type": "kotlin.Boolean"
},
@@ -1946,42 +1952,7 @@
},
{
"type": "kotlin.reflect.jvm.internal.impl.resolve.scopes.DescriptorKindFilter",
- "allPublicFields": true,
- "fields": [
- {
- "name": "ALL"
- },
- {
- "name": "CALLABLES"
- },
- {
- "name": "CLASSIFIERS"
- },
- {
- "name": "Companion"
- },
- {
- "name": "FUNCTIONS"
- },
- {
- "name": "NON_SINGLETON_CLASSIFIERS"
- },
- {
- "name": "PACKAGES"
- },
- {
- "name": "SINGLETON_CLASSIFIERS"
- },
- {
- "name": "TYPE_ALIASES"
- },
- {
- "name": "VALUES"
- },
- {
- "name": "VARIABLES"
- }
- ]
+ "allPublicFields": true
},
{
"type": "kotlinx.coroutines.AwaitAll"
@@ -2513,6 +2484,10 @@
"type": "org.asamk.signal.commands.ListAccountsCommand$JsonAccount",
"allDeclaredFields": true,
"methods": [
+ {
+ "name": "aci",
+ "parameterTypes": []
+ },
{
"name": "number",
"parameterTypes": []
@@ -5118,6 +5093,15 @@
}
]
},
+ {
+ "type": "org.bouncycastle.jcajce.provider.asymmetric.XMSS$Mappings",
+ "methods": [
+ {
+ "name": "",
+ "parameterTypes": []
+ }
+ ]
+ },
{
"type": "org.bouncycastle.jcajce.provider.asymmetric.edec.SignatureSpi$Ed25519",
"methods": [
@@ -6566,6 +6550,18 @@
}
]
},
+ {
+ "type": "org.signal.libsignal.net.UsernameNotAvailableException",
+ "jniAccessible": true,
+ "methods": [
+ {
+ "name": "",
+ "parameterTypes": [
+ "java.lang.String"
+ ]
+ }
+ ]
+ },
{
"type": "org.signal.libsignal.net.internal.BridgeChatListener",
"jniAccessible": true
@@ -7184,6 +7180,9 @@
"type": "org.signal.libsignal.zkgroup.profiles.ExpiringProfileKeyCredential",
"allDeclaredFields": true
},
+ {
+ "type": "org.signal.libsignal.zkgroup.profiles.ExpiringProfileKeyCredentialResponse"
+ },
{
"type": "org.signal.libsignal.zkgroup.profiles.ProfileKey",
"allDeclaredFields": true
@@ -7250,6 +7249,15 @@
}
]
},
+ {
+ "type": "org.signal.network.util.JsonUtil$UuidDeserializer",
+ "methods": [
+ {
+ "name": "",
+ "parameterTypes": []
+ }
+ ]
+ },
{
"type": "org.signal.storageservice.storage.protos.groups.AccessControl",
"fields": [
@@ -8321,6 +8329,13 @@
{
"name": "",
"parameterTypes": []
+ },
+ {
+ "name": "",
+ "parameterTypes": [
+ "int",
+ "int"
+ ]
}
]
},
@@ -8416,6 +8431,17 @@
{
"name": "",
"parameterTypes": []
+ },
+ {
+ "name": "",
+ "parameterTypes": [
+ "java.lang.String",
+ "java.lang.String",
+ "java.lang.String",
+ "java.util.List",
+ "java.util.List",
+ "java.lang.Long"
+ ]
}
]
},
@@ -8463,6 +8489,12 @@
"allDeclaredMethods": true,
"allDeclaredConstructors": true
},
+ {
+ "type": "org.whispersystems.signalservice.api.profiles.SignalServiceProfile$Companion"
+ },
+ {
+ "type": "org.whispersystems.signalservice.api.profiles.SignalServiceProfile$RequestType"
+ },
{
"type": "org.whispersystems.signalservice.api.profiles.SignalServiceProfileWrite",
"allDeclaredFields": true,
@@ -8471,7 +8503,13 @@
},
{
"type": "org.whispersystems.signalservice.api.provisioning.ProvisioningMessage",
- "allDeclaredFields": true
+ "allDeclaredFields": true,
+ "methods": [
+ {
+ "name": "getBody",
+ "parameterTypes": []
+ }
+ ]
},
{
"type": "org.whispersystems.signalservice.api.push.ServiceId",
@@ -8919,6 +8957,9 @@
"allDeclaredMethods": true,
"allDeclaredConstructors": true
},
+ {
+ "type": "org.whispersystems.signalservice.internal.push.AuthCredentials$Companion"
+ },
{
"type": "org.whispersystems.signalservice.internal.push.ByteArrayDeserializerBase64",
"methods": [
@@ -8928,6 +8969,15 @@
}
]
},
+ {
+ "type": "org.whispersystems.signalservice.internal.push.ByteArraySerializerBase64NoPadding",
+ "methods": [
+ {
+ "name": "",
+ "parameterTypes": []
+ }
+ ]
+ },
{
"type": "org.whispersystems.signalservice.internal.push.CdsiAuthResponse",
"allDeclaredFields": true,
@@ -8935,6 +8985,13 @@
{
"name": "",
"parameterTypes": []
+ },
+ {
+ "name": "",
+ "parameterTypes": [
+ "java.lang.String",
+ "java.lang.String"
+ ]
}
]
},
@@ -8946,7 +9003,21 @@
},
{
"type": "org.whispersystems.signalservice.internal.push.ConfirmUsernameRequest",
- "allDeclaredFields": true
+ "allDeclaredFields": true,
+ "methods": [
+ {
+ "name": "getEncryptedUsername",
+ "parameterTypes": []
+ },
+ {
+ "name": "getUsernameHash",
+ "parameterTypes": []
+ },
+ {
+ "name": "getZkProof",
+ "parameterTypes": []
+ }
+ ]
},
{
"type": "org.whispersystems.signalservice.internal.push.ConfirmUsernameResponse",
@@ -9096,6 +9167,14 @@
"name": "",
"parameterTypes": []
},
+ {
+ "name": "",
+ "parameterTypes": [
+ "long",
+ "org.signal.libsignal.protocol.kem.KEMPublicKey",
+ "byte[]"
+ ]
+ },
{
"name": "getKeyId",
"parameterTypes": []
@@ -9182,6 +9261,9 @@
"allDeclaredMethods": true,
"allDeclaredConstructors": true
},
+ {
+ "type": "org.whispersystems.signalservice.internal.push.MismatchedDevices$Companion"
+ },
{
"type": "org.whispersystems.signalservice.internal.push.OneTimePreKeyCounts",
"allDeclaredFields": true,
@@ -9576,6 +9658,67 @@
{
"type": "org.whispersystems.signalservice.internal.push.StaleDevices",
"allDeclaredFields": true,
+ "methods": [
+ {
+ "name": "",
+ "parameterTypes": []
+ },
+ {
+ "name": "",
+ "parameterTypes": [
+ "java.util.List"
+ ]
+ }
+ ]
+ },
+ {
+ "type": "org.whispersystems.signalservice.internal.push.StickerUploadAttributes",
+ "fields": [
+ {
+ "name": "acl"
+ },
+ {
+ "name": "algorithm"
+ },
+ {
+ "name": "credential"
+ },
+ {
+ "name": "date"
+ },
+ {
+ "name": "id"
+ },
+ {
+ "name": "key"
+ },
+ {
+ "name": "policy"
+ },
+ {
+ "name": "signature"
+ }
+ ],
+ "methods": [
+ {
+ "name": "",
+ "parameterTypes": []
+ }
+ ]
+ },
+ {
+ "type": "org.whispersystems.signalservice.internal.push.StickerUploadAttributesResponse",
+ "fields": [
+ {
+ "name": "manifest"
+ },
+ {
+ "name": "packId"
+ },
+ {
+ "name": "stickers"
+ }
+ ],
"methods": [
{
"name": "",
diff --git a/src/test/java/org/asamk/signal/commands/SendCommandErrorTest.java b/src/test/java/org/asamk/signal/commands/SendCommandErrorTest.java
new file mode 100644
index 00000000..a2105437
--- /dev/null
+++ b/src/test/java/org/asamk/signal/commands/SendCommandErrorTest.java
@@ -0,0 +1,59 @@
+package org.asamk.signal.commands;
+
+import org.asamk.signal.commands.exceptions.UnexpectedErrorException;
+import org.asamk.signal.manager.Manager;
+import org.asamk.signal.manager.api.AttachmentInvalidException;
+import org.asamk.signal.output.JsonWriter;
+import org.asamk.signal.testutil.ManagerMock;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.IOException;
+import java.lang.reflect.Proxy;
+import java.net.SocketException;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class SendCommandErrorTest {
+
+ @ParameterizedTest
+ @MethodSource("failures")
+ void uploadAndTransmissionFailuresKeepTheirClassification(final Exception cause) {
+ final var error = send(cause);
+
+ assertEquals("Failed to send message: " + cause.getMessage() + " (" + cause.getClass().getSimpleName() + ")",
+ error.getMessage());
+ assertSame(cause, error.getCause());
+ }
+
+ private static Stream failures() {
+ return Stream.of(new AttachmentInvalidException("inline attachment #2: Connection reset"),
+ new SocketException("Connection reset"),
+ new IOException("Server error: 500"));
+ }
+
+ private static UnexpectedErrorException send(final Exception failure) {
+ final var delegate = ManagerMock.create("+15551234567");
+ final var manager = (Manager) Proxy.newProxyInstance(Manager.class.getClassLoader(),
+ new Class[]{Manager.class}, (proxy, method, args) -> {
+ if (method.getName().equals("sendMessage")) {
+ throw failure;
+ }
+ return method.invoke(delegate, args);
+ });
+ // Upload and transmission failures must map to the same exception type, so the exit code and the
+ // JSON-RPC error code stay the same.
+ return assertThrows(UnexpectedErrorException.class,
+ () -> new SendCommand().handleCommand(new JsonRpcNamespace(Map.of("noteToSelf",
+ true,
+ "message",
+ "hi",
+ "attachments",
+ List.of("photo.jpg"))), manager, (JsonWriter) ignored -> {}));
+ }
+}
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/jsonrpc/SubscribeCallEventsTest.java b/src/test/java/org/asamk/signal/jsonrpc/SubscribeCallEventsTest.java
index 9ec1c87d..cce6c888 100644
--- a/src/test/java/org/asamk/signal/jsonrpc/SubscribeCallEventsTest.java
+++ b/src/test/java/org/asamk/signal/jsonrpc/SubscribeCallEventsTest.java
@@ -67,6 +67,8 @@ class SubscribeCallEventsTest {
final List managers;
final List> addedHandlers = new ArrayList<>();
+ final List> removedHandlers = new ArrayList<>();
+ int addedHandlerRegistrationCount;
StubMultiAccountManager(List managers) {
this.managers = new ArrayList<>(managers);
@@ -79,11 +81,23 @@ class SubscribeCallEventsTest {
@Override
public void addOnManagerAddedHandler(Consumer handler) {
+ addedHandlerRegistrationCount++;
addedHandlers.add(handler);
}
+ @Override
+ public void removeOnManagerAddedHandler(Consumer handler) {
+ addedHandlers.remove(handler);
+ }
+
@Override
public void addOnManagerRemovedHandler(Consumer handler) {
+ removedHandlers.add(handler);
+ }
+
+ @Override
+ public void removeOnManagerRemovedHandler(Consumer handler) {
+ removedHandlers.remove(handler);
}
@Override
@@ -223,8 +237,9 @@ class SubscribeCallEventsTest {
assertEquals(1, manager1.state().addCallEventListenerCount.get(), "manager1 should have one listener");
assertEquals(1, manager2.state().addCallEventListenerCount.get(), "manager2 should have one listener");
- // Also registers an onManagerAdded handler for receive and one for call events
- assertEquals(2, multi.addedHandlers.size(), "should register onManagerAdded handlers");
+ assertEquals(2, multi.addedHandlerRegistrationCount, "should register onManagerAdded handlers");
+ assertEquals(0, multi.addedHandlers.size(), "should remove onManagerAdded handlers after connection closes");
+ assertEquals(0, multi.removedHandlers.size(), "should not register onManagerRemoved handlers in manual mode");
}
@Test
@@ -263,4 +278,17 @@ class SubscribeCallEventsTest {
manager1.state().addCallEventListenerCount.get(),
"call events should not be auto-subscribed in multi mode");
}
+
+ @Test
+ void multiAccountConnectionRemovesAllManagerHandlersWhenReceiveIsEnabled() {
+ var manager = createManager("+15551111111");
+ var multi = new StubMultiAccountManager(List.of(manager.manager()));
+ var handler = new SignalJsonRpcDispatcherHandler(new CapturingJsonWriter(), () -> null, false);
+
+ handler.handleConnection(multi);
+
+ assertEquals(3, multi.addedHandlerRegistrationCount);
+ assertEquals(0, multi.addedHandlers.size(), "should remove all onManagerAdded handlers after close");
+ assertEquals(0, multi.removedHandlers.size(), "should remove onManagerRemoved handler after close");
+ }
}
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);
}