X-Git-Url: https://git.nmode.ca/signal-cli/blobdiff_plain/e22cc457ae91e22deb54373078bef7e8cd22edcb..fa5c09d23b830f2999a52421c189a3e4661da99f:/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java diff --git a/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java b/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java index 5ee2e7d3..81c18683 100644 --- a/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java +++ b/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java @@ -95,6 +95,7 @@ import java.net.URISyntaxException; import java.net.URLEncoder; import java.nio.charset.StandardCharsets; import java.security.SignatureException; +import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; @@ -108,6 +109,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; +import java.util.stream.Stream; import static org.asamk.signal.manager.config.ServiceConfig.capabilities; @@ -139,7 +141,9 @@ public class ManagerImpl implements Manager { private boolean ignoreAttachments = false; private Thread receiveThread; + private final Set weakHandlers = new HashSet<>(); private final Set messageHandlers = new HashSet<>(); + private final List closedListeners = new ArrayList<>(); private boolean isReceivingSynchronous; ManagerImpl( @@ -152,7 +156,7 @@ public class ManagerImpl implements Manager { this.serviceEnvironmentConfig = serviceEnvironmentConfig; final var credentialsProvider = new DynamicCredentialsProvider(account.getAci(), - account.getUsername(), + account.getAccount(), account.getPassword(), account.getDeviceId()); final var sessionLock = new SignalSessionLock() { @@ -247,7 +251,7 @@ public class ManagerImpl implements Manager { @Override public String getSelfNumber() { - return account.getUsername(); + return account.getAccount(); } @Override @@ -281,7 +285,7 @@ public class ManagerImpl implements Manager { public Map> areUsersRegistered(Set numbers) throws IOException { Map canonicalizedNumbers = numbers.stream().collect(Collectors.toMap(n -> n, n -> { try { - return PhoneNumberFormatter.formatNumber(n, account.getUsername()); + return PhoneNumberFormatter.formatNumber(n, account.getAccount()); } catch (InvalidNumberException e) { return ""; } @@ -383,6 +387,7 @@ public class ManagerImpl implements Manager { dependencies.getAccountManager().setGcmId(Optional.absent()); account.setRegistered(false); + close(); } @Override @@ -397,10 +402,13 @@ public class ManagerImpl implements Manager { dependencies.getAccountManager().deleteAccount(); account.setRegistered(false); + close(); } @Override public void submitRateLimitRecaptchaChallenge(String challenge, String captcha) throws IOException { + captcha = captcha == null ? null : captcha.replace("signalcaptcha://", ""); + dependencies.getAccountManager().submitRateLimitRecaptchaChallenge(challenge, captcha); } @@ -675,6 +683,16 @@ public class ManagerImpl implements Manager { if (attachments != null) { messageBuilder.withAttachments(attachmentHelper.uploadAttachments(attachments)); } + if (message.mentions().size() > 0) { + final var mentions = new ArrayList(); + for (final var m : message.mentions()) { + final var recipientId = resolveRecipient(m.recipient()); + mentions.add(new SignalServiceDataMessage.Mention(resolveSignalServiceAddress(recipientId).getAci(), + m.start(), + m.length())); + } + messageBuilder.withMentions(mentions); + } } @Override @@ -904,14 +922,17 @@ public class ManagerImpl implements Manager { } @Override - public void addReceiveHandler(final ReceiveMessageHandler handler) { + public void addReceiveHandler(final ReceiveMessageHandler handler, final boolean isWeakListener) { if (isReceivingSynchronous) { throw new IllegalStateException("Already receiving message synchronously."); } synchronized (messageHandlers) { - messageHandlers.add(handler); - - startReceiveThreadIfRequired(); + if (isWeakListener) { + weakHandlers.add(handler); + } else { + messageHandlers.add(handler); + startReceiveThreadIfRequired(); + } } } @@ -920,17 +941,18 @@ public class ManagerImpl implements Manager { return; } receiveThread = new Thread(() -> { + logger.debug("Starting receiving messages"); while (!Thread.interrupted()) { try { receiveMessagesInternal(1L, TimeUnit.HOURS, false, (envelope, e) -> { synchronized (messageHandlers) { - for (ReceiveMessageHandler h : messageHandlers) { + Stream.concat(messageHandlers.stream(), weakHandlers.stream()).forEach(h -> { try { h.handleMessage(envelope, e); } catch (Exception ex) { logger.warn("Message handler failed, ignoring", ex); } - } + }); } }); break; @@ -938,12 +960,14 @@ public class ManagerImpl implements Manager { logger.warn("Receiving messages failed, retrying", e); } } + logger.debug("Finished receiving messages"); hasCaughtUpWithOldMessages = false; synchronized (messageHandlers) { receiveThread = null; // Check if in the meantime another handler has been registered if (!messageHandlers.isEmpty()) { + logger.debug("Another handler has been registered, starting receive thread again"); startReceiveThreadIfRequired(); } } @@ -956,12 +980,13 @@ public class ManagerImpl implements Manager { public void removeReceiveHandler(final ReceiveMessageHandler handler) { final Thread thread; synchronized (messageHandlers) { - thread = receiveThread; - receiveThread = null; + weakHandlers.remove(handler); messageHandlers.remove(handler); - if (!messageHandlers.isEmpty() || isReceivingSynchronous) { + if (!messageHandlers.isEmpty() || receiveThread == null || isReceivingSynchronous) { return; } + thread = receiveThread; + receiveThread = null; } stopReceiveThread(thread); @@ -1017,7 +1042,8 @@ public class ManagerImpl implements Manager { ) throws IOException { retryFailedReceivedMessages(handler); - Set queuedActions = new HashSet<>(); + // Use a Map here because java Set doesn't have a get method ... + Map queuedActions = new HashMap<>(); final var signalWebSocket = dependencies.getSignalWebSocket(); signalWebSocket.connect(); @@ -1029,7 +1055,10 @@ public class ManagerImpl implements Manager { while (!Thread.interrupted()) { SignalServiceEnvelope envelope; final CachedMessage[] cachedMessage = {null}; - account.setLastReceiveTimestamp(System.currentTimeMillis()); + final var nowMillis = System.currentTimeMillis(); + if (nowMillis - account.getLastReceiveTimestamp() > 60000) { + account.setLastReceiveTimestamp(nowMillis); + } logger.debug("Checking for new message from server"); try { var result = signalWebSocket.readOrEmpty(unit.toMillis(timeout), envelope1 -> { @@ -1046,7 +1075,7 @@ public class ManagerImpl implements Manager { logger.debug("New message received from server"); } else { logger.debug("Received indicator that server queue is empty"); - handleQueuedActions(queuedActions); + handleQueuedActions(queuedActions.keySet()); queuedActions.clear(); hasCaughtUpWithOldMessages = true; @@ -1087,11 +1116,18 @@ public class ManagerImpl implements Manager { } final var result = incomingMessageHandler.handleEnvelope(envelope, ignoreAttachments, handler); - queuedActions.addAll(result.first()); + for (final var h : result.first()) { + final var existingAction = queuedActions.get(h); + if (existingAction == null) { + queuedActions.put(h, h); + } else { + existingAction.mergeOther(h); + } + } final var exception = result.second(); if (hasCaughtUpWithOldMessages) { - handleQueuedActions(queuedActions); + handleQueuedActions(queuedActions.keySet()); queuedActions.clear(); } if (cachedMessage[0] != null) { @@ -1112,8 +1148,9 @@ public class ManagerImpl implements Manager { } } } - handleQueuedActions(queuedActions); + handleQueuedActions(queuedActions.keySet()); queuedActions.clear(); + dependencies.getSignalWebSocket().disconnect(); } @Override @@ -1241,7 +1278,7 @@ public class ManagerImpl implements Manager { /** * Trust this the identity with this fingerprint * - * @param recipient username of the identity + * @param recipient account of the identity * @param fingerprint Fingerprint */ @Override @@ -1258,7 +1295,7 @@ public class ManagerImpl implements Manager { /** * Trust this the identity with this safety number * - * @param recipient username of the identity + * @param recipient account of the identity * @param safetyNumber Safety number */ @Override @@ -1275,7 +1312,7 @@ public class ManagerImpl implements Manager { /** * Trust this the identity with this scannable safety number * - * @param recipient username of the identity + * @param recipient account of the identity * @param safetyNumber Scannable safety number */ @Override @@ -1292,7 +1329,7 @@ public class ManagerImpl implements Manager { /** * Trust all keys of this identity without verification * - * @param recipient username of the identity + * @param recipient account of the identity */ @Override public boolean trustIdentityAllKeys(RecipientIdentifier.Single recipient) { @@ -1305,6 +1342,13 @@ public class ManagerImpl implements Manager { return identityHelper.trustIdentityAllKeys(recipientId); } + @Override + public void addClosedListener(final Runnable listener) { + synchronized (closedListeners) { + closedListeners.add(listener); + } + } + private void handleIdentityFailure( final RecipientId recipientId, final org.whispersystems.signalservice.api.messages.SendMessageResult.IdentityFailure identityFailure @@ -1370,12 +1414,9 @@ public class ManagerImpl implements Manager { @Override public void close() throws IOException { - close(true); - } - - private void close(boolean closeAccount) throws IOException { Thread thread; synchronized (messageHandlers) { + weakHandlers.clear(); messageHandlers.clear(); thread = receiveThread; receiveThread = null; @@ -1387,7 +1428,12 @@ public class ManagerImpl implements Manager { dependencies.getSignalWebSocket().disconnect(); - if (closeAccount && account != null) { + synchronized (closedListeners) { + closedListeners.forEach(Runnable::run); + closedListeners.clear(); + } + + if (account != null) { account.close(); } account = null;