X-Git-Url: https://git.nmode.ca/signal-cli/blobdiff_plain/b7005884fdcfa8d95f54e557ddbf2fe4201962f8..9ad24614cb800c0e3f853dc985d8d4180bbdd04d:/src/main/java/org/asamk/signal/jsonrpc/SignalJsonRpcDispatcherHandler.java diff --git a/src/main/java/org/asamk/signal/jsonrpc/SignalJsonRpcDispatcherHandler.java b/src/main/java/org/asamk/signal/jsonrpc/SignalJsonRpcDispatcherHandler.java index c4e9775a..086681f7 100644 --- a/src/main/java/org/asamk/signal/jsonrpc/SignalJsonRpcDispatcherHandler.java +++ b/src/main/java/org/asamk/signal/jsonrpc/SignalJsonRpcDispatcherHandler.java @@ -5,11 +5,11 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonMappingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ContainerNode; +import com.fasterxml.jackson.databind.node.IntNode; import com.fasterxml.jackson.databind.node.ObjectNode; -import org.asamk.signal.JsonReceiveMessageHandler; -import org.asamk.signal.JsonWriter; import org.asamk.signal.commands.Command; import org.asamk.signal.commands.Commands; import org.asamk.signal.commands.JsonRpcMultiCommand; @@ -19,17 +19,22 @@ import org.asamk.signal.commands.exceptions.CommandException; import org.asamk.signal.commands.exceptions.IOErrorException; import org.asamk.signal.commands.exceptions.UntrustedKeyErrorException; import org.asamk.signal.commands.exceptions.UserErrorException; +import org.asamk.signal.json.JsonReceiveMessageHandler; import org.asamk.signal.manager.Manager; import org.asamk.signal.manager.MultiAccountManager; import org.asamk.signal.manager.RegistrationManager; +import org.asamk.signal.manager.api.Pair; +import org.asamk.signal.output.JsonWriter; import org.asamk.signal.util.Util; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; +import java.nio.channels.OverlappingFileLockException; import java.util.HashMap; +import java.util.List; import java.util.Map; -import java.util.Objects; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; public class SignalJsonRpcDispatcherHandler { @@ -46,7 +51,7 @@ public class SignalJsonRpcDispatcherHandler { private final boolean noReceiveOnStart; private MultiAccountManager c; - private final Map receiveHandlers = new HashMap<>(); + private final Map>> receiveHandlers = new HashMap<>(); private Manager m; @@ -63,7 +68,9 @@ public class SignalJsonRpcDispatcherHandler { this.c = c; if (!noReceiveOnStart) { - c.getAccountNumbers().stream().map(c::getManager).filter(Objects::nonNull).forEach(this::subscribeReceive); + this.subscribeReceive(c.getManagers()); + c.addOnManagerAddedHandler(this::subscribeReceive); + c.addOnManagerRemovedHandler(this::unsubscribeReceive); } handleConnection(); @@ -76,36 +83,52 @@ public class SignalJsonRpcDispatcherHandler { subscribeReceive(m); } + final var currentThread = Thread.currentThread(); + m.addClosedListener(currentThread::interrupt); + handleConnection(); } - private void subscribeReceive(final Manager m) { - if (receiveHandlers.containsKey(m)) { - return; - } + private static final AtomicInteger nextSubscriptionId = new AtomicInteger(0); - final var receiveMessageHandler = new JsonReceiveMessageHandler(m, - s -> jsonRpcSender.sendRequest(JsonRpcRequest.forNotification("receive", - objectMapper.valueToTree(s), - null))); - m.addReceiveHandler(receiveMessageHandler); - receiveHandlers.put(m, receiveMessageHandler); + private int subscribeReceive(final Manager manager) { + return subscribeReceive(List.of(manager)); + } - while (!m.hasCaughtUpWithOldMessages()) { - try { - synchronized (m) { - m.wait(); - } - } catch (InterruptedException ignored) { - } - } + private int subscribeReceive(final List managers) { + final var subscriptionId = nextSubscriptionId.getAndIncrement(); + final var handlers = managers.stream().map(m -> { + final var receiveMessageHandler = new JsonReceiveMessageHandler(m, s -> { + final ContainerNode params = objectMapper.valueToTree(s); + ((ObjectNode) params).set("subscription", IntNode.valueOf(subscriptionId)); + jsonRpcSender.sendRequest(JsonRpcRequest.forNotification("receive", params, null)); + }); + m.addReceiveHandler(receiveMessageHandler); + return new Pair<>(m, (Manager.ReceiveMessageHandler) receiveMessageHandler); + }).toList(); + receiveHandlers.put(subscriptionId, handlers); + + return subscriptionId; } - void unsubscribeReceive(final Manager m) { - final var receiveMessageHandler = receiveHandlers.remove(m); - if (receiveMessageHandler != null) { - m.removeReceiveHandler(receiveMessageHandler); + private boolean unsubscribeReceive(final int subscriptionId) { + final var handlers = receiveHandlers.remove(subscriptionId); + if (handlers == null) { + return false; } + for (final var pair : handlers) { + unsubscribeReceiveHandler(pair); + } + return true; + } + + private void unsubscribeReceive(final Manager m) { + final var subscriptionId = receiveHandlers.entrySet() + .stream() + .filter(e -> e.getValue().size() == 1 && e.getValue().get(0).first().equals(m)) + .map(Map.Entry::getKey) + .findFirst(); + subscriptionId.ifPresent(this::unsubscribeReceive); } private void handleConnection() { @@ -113,16 +136,28 @@ public class SignalJsonRpcDispatcherHandler { jsonRpcReader.readMessages((method, params) -> handleRequest(objectMapper, method, params), response -> logger.debug("Received unexpected response for id {}", response.getId())); } finally { - receiveHandlers.forEach(Manager::removeReceiveHandler); + receiveHandlers.forEach((_subscriptionId, handlers) -> handlers.forEach(this::unsubscribeReceiveHandler)); receiveHandlers.clear(); } } + private void unsubscribeReceiveHandler(final Pair pair) { + final var m = pair.first(); + final var handler = pair.second(); + m.removeReceiveHandler(handler); + } + private JsonNode handleRequest( final ObjectMapper objectMapper, final String method, ContainerNode params ) throws JsonRpcException { var command = getCommand(method); if (c != null) { + if (command instanceof JsonRpcSingleCommand jsonRpcCommand) { + final var manager = getManagerFromParams(params); + if (manager != null) { + return runCommand(objectMapper, params, new CommandRunnerImpl<>(manager, jsonRpcCommand)); + } + } if (command instanceof JsonRpcMultiCommand jsonRpcCommand) { return runCommand(objectMapper, params, new MultiCommandRunnerImpl<>(c, jsonRpcCommand)); } @@ -162,10 +197,15 @@ public class SignalJsonRpcDispatcherHandler { null)); } - private Manager getManagerFromParams(final ContainerNode params) { - if (params != null && params.has("account")) { + private Manager getManagerFromParams(final ContainerNode params) throws JsonRpcException { + if (params != null && params.hasNonNull("account")) { final var manager = c.getManager(params.get("account").asText()); ((ObjectNode) params).remove("account"); + if (manager == null) { + throw new JsonRpcException(new JsonRpcResponse.Error(JsonRpcResponse.Error.INVALID_PARAMS, + "Specified account does not exist", + null)); + } return manager; } return null; @@ -177,6 +217,9 @@ public class SignalJsonRpcDispatcherHandler { final var registrationManager = c.getNewRegistrationManager(params.get("account").asText()); ((ObjectNode) params).remove("account"); return registrationManager; + } catch (OverlappingFileLockException e) { + logger.warn("Account is already in use"); + return null; } catch (IOException | IllegalStateException e) { logger.warn("Failed to load registration manager", e); return null; @@ -264,22 +307,35 @@ public class SignalJsonRpcDispatcherHandler { e.getMessage(), null)); } catch (UserErrorException e) { - throw new JsonRpcException(new JsonRpcResponse.Error(USER_ERROR, e.getMessage(), null)); + throw new JsonRpcException(new JsonRpcResponse.Error(USER_ERROR, + e.getMessage(), + getErrorDataNode(objectMapper, result))); } catch (IOErrorException e) { - throw new JsonRpcException(new JsonRpcResponse.Error(IO_ERROR, e.getMessage(), null)); + throw new JsonRpcException(new JsonRpcResponse.Error(IO_ERROR, + e.getMessage(), + getErrorDataNode(objectMapper, result))); } catch (UntrustedKeyErrorException e) { - throw new JsonRpcException(new JsonRpcResponse.Error(UNTRUSTED_KEY_ERROR, e.getMessage(), null)); + throw new JsonRpcException(new JsonRpcResponse.Error(UNTRUSTED_KEY_ERROR, + e.getMessage(), + getErrorDataNode(objectMapper, result))); } catch (Throwable e) { logger.error("Command execution failed", e); throw new JsonRpcException(new JsonRpcResponse.Error(JsonRpcResponse.Error.INTERNAL_ERROR, e.getMessage(), - null)); + getErrorDataNode(objectMapper, result))); } Object output = result[0] == null ? Map.of() : result[0]; return objectMapper.valueToTree(output); } + private JsonNode getErrorDataNode(final ObjectMapper objectMapper, final Object[] result) { + if (result[0] == null) { + return null; + } + return objectMapper.valueToTree(Map.of("response", result[0])); + } + private void parseParamsAndRunCommand( final ObjectMapper objectMapper, final TreeNode params, @@ -300,7 +356,7 @@ public class SignalJsonRpcDispatcherHandler { command.handleCommand(requestParams, jsonWriter); } - private class SubscribeReceiveCommand implements JsonRpcSingleCommand { + private class SubscribeReceiveCommand implements JsonRpcSingleCommand, JsonRpcMultiCommand { @Override public String getName() { @@ -311,22 +367,67 @@ public class SignalJsonRpcDispatcherHandler { public void handleCommand( final Void request, final Manager m, final JsonWriter jsonWriter ) throws CommandException { - subscribeReceive(m); + final var subscriptionId = subscribeReceive(m); + jsonWriter.write(subscriptionId); + } + + @Override + public void handleCommand( + final Void request, final MultiAccountManager c, final JsonWriter jsonWriter + ) throws CommandException { + final var subscriptionId = subscribeReceive(c.getManagers()); + jsonWriter.write(subscriptionId); } } - private class UnsubscribeReceiveCommand implements JsonRpcSingleCommand { + private class UnsubscribeReceiveCommand implements JsonRpcSingleCommand, JsonRpcMultiCommand { @Override public String getName() { return "unsubscribeReceive"; } + @Override + public TypeReference getRequestType() { + return new TypeReference<>() {}; + } + @Override public void handleCommand( - final Void request, final Manager m, final JsonWriter jsonWriter + final JsonNode request, final Manager m, final JsonWriter jsonWriter + ) throws CommandException { + final var subscriptionId = getSubscriptionId(request); + if (subscriptionId == null) { + unsubscribeReceive(m); + } else { + if (!unsubscribeReceive(subscriptionId)) { + throw new UserErrorException("Unknown subscription id"); + } + } + } + + @Override + public void handleCommand( + final JsonNode request, final MultiAccountManager c, final JsonWriter jsonWriter ) throws CommandException { - unsubscribeReceive(m); + final var subscriptionId = getSubscriptionId(request); + if (subscriptionId == null) { + throw new UserErrorException("Missing subscription parameter with subscription id"); + } else { + if (!unsubscribeReceive(subscriptionId)) { + throw new UserErrorException("Unknown subscription id"); + } + } + } + + private Integer getSubscriptionId(final JsonNode request) { + if (request instanceof ArrayNode req) { + return req.get(0).asInt(); + } else if (request instanceof ObjectNode req) { + return req.get("subscription").asInt(); + } else { + return null; + } } } }