X-Git-Url: https://git.nmode.ca/signal-cli/blobdiff_plain/7e7e4150e19a65c646bb3f4ec33b5ec87224f8cf..71e0c3f80c4dddfa6c689f0aa897ecf9a011bce1:/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 269edb3b..cb9e4730 100644 --- a/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java +++ b/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java @@ -77,9 +77,11 @@ import org.whispersystems.signalservice.api.messages.SignalServiceReceiptMessage import org.whispersystems.signalservice.api.messages.SignalServiceTypingMessage; import org.whispersystems.signalservice.api.push.ACI; import org.whispersystems.signalservice.api.push.SignalServiceAddress; +import org.whispersystems.signalservice.api.push.exceptions.AuthorizationFailedException; import org.whispersystems.signalservice.api.util.DeviceNameUtil; import org.whispersystems.signalservice.api.util.InvalidNumberException; import org.whispersystems.signalservice.api.util.PhoneNumberFormatter; +import org.whispersystems.signalservice.api.websocket.WebSocketConnectionState; import org.whispersystems.signalservice.api.websocket.WebSocketUnavailableException; import org.whispersystems.signalservice.internal.contacts.crypto.Quote; import org.whispersystems.signalservice.internal.contacts.crypto.UnauthenticatedQuoteException; @@ -95,6 +97,7 @@ import java.net.URISyntaxException; import java.net.URLEncoder; import java.nio.charset.StandardCharsets; import java.security.SignatureException; +import java.time.Duration; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; @@ -111,6 +114,9 @@ import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; import java.util.stream.Stream; +import io.reactivex.rxjava3.core.Observable; +import io.reactivex.rxjava3.schedulers.Schedulers; + import static org.asamk.signal.manager.config.ServiceConfig.capabilities; public class ManagerImpl implements Manager { @@ -190,7 +196,7 @@ public class ManagerImpl implements Manager { unidentifiedAccessHelper::getAccessFor, this::resolveSignalServiceAddress); final GroupV2Helper groupV2Helper = new GroupV2Helper(profileHelper::getRecipientProfileKeyCredential, - this::getRecipientProfile, + profileHelper::getRecipientProfile, account::getSelfRecipientId, dependencies.getGroupsV2Operations(), dependencies.getGroupsV2Api(), @@ -202,6 +208,7 @@ public class ManagerImpl implements Manager { account.getRecipientStore(), this::handleIdentityFailure, this::getGroupInfo, + profileHelper::getRecipientProfile, this::refreshRegisteredUser); this.groupHelper = new GroupHelper(account, dependencies, @@ -240,7 +247,7 @@ public class ManagerImpl implements Manager { contactHelper, attachmentHelper, syncHelper, - this::getRecipientProfile, + profileHelper::getRecipientProfile, jobExecutor); this.identityHelper = new IdentityHelper(account, dependencies, @@ -267,11 +274,16 @@ public class ManagerImpl implements Manager { days); } } - preKeyHelper.refreshPreKeysIfNecessary(); - if (account.getAci() == null) { - account.setAci(dependencies.getAccountManager().getOwnAci()); + try { + preKeyHelper.refreshPreKeysIfNecessary(); + if (account.getAci() == null) { + account.setAci(ACI.parseOrNull(dependencies.getAccountManager().getWhoAmI().getAci())); + } + updateAccountAttributes(null); + } catch (AuthorizationFailedException e) { + account.setRegistered(false); + throw e; } - updateAccountAttributes(null); } /** @@ -435,7 +447,7 @@ public class ManagerImpl implements Manager { d.getCreated(), d.getLastSeen(), d.getId() == account.getDeviceId()); - }).collect(Collectors.toList()); + }).toList(); } @Override @@ -507,7 +519,7 @@ public class ManagerImpl implements Manager { @Override public List getGroups() { - return account.getGroupStore().getGroups().stream().map(this::toGroup).collect(Collectors.toList()); + return account.getGroupStore().getGroups().stream().map(this::toGroup).toList(); } private Group toGroup(final GroupInfo groupInfo) { @@ -618,7 +630,7 @@ public class ManagerImpl implements Manager { .map(sendMessageResult -> SendMessageResult.from(sendMessageResult, account.getRecipientStore(), account.getRecipientStore()::resolveRecipientAddress)) - .collect(Collectors.toList())); + .toList()); } } return new SendMessageResults(timestamp, results); @@ -647,7 +659,7 @@ public class ManagerImpl implements Manager { .map(r -> SendMessageResult.from(r, account.getRecipientStore(), account.getRecipientStore()::resolveRecipientAddress)) - .collect(Collectors.toList())); + .toList()); } } return new SendMessageResults(timestamp, results); @@ -994,7 +1006,7 @@ public class ManagerImpl implements Manager { logger.debug("Starting receiving messages"); while (!Thread.interrupted()) { try { - receiveMessagesInternal(1L, TimeUnit.HOURS, false, (envelope, e) -> { + receiveMessagesInternal(Duration.ofMinutes(1), false, (envelope, e) -> { synchronized (messageHandlers) { Stream.concat(messageHandlers.stream(), weakHandlers.stream()).forEach(h -> { try { @@ -1061,17 +1073,17 @@ public class ManagerImpl implements Manager { } @Override - public void receiveMessages(long timeout, TimeUnit unit, ReceiveMessageHandler handler) throws IOException { - receiveMessages(timeout, unit, true, handler); + public void receiveMessages(Duration timeout, ReceiveMessageHandler handler) throws IOException { + receiveMessages(timeout, true, handler); } @Override public void receiveMessages(ReceiveMessageHandler handler) throws IOException { - receiveMessages(1L, TimeUnit.HOURS, false, handler); + receiveMessages(Duration.ofMinutes(1), false, handler); } private void receiveMessages( - long timeout, TimeUnit unit, boolean returnOnTimeout, ReceiveMessageHandler handler + Duration timeout, boolean returnOnTimeout, ReceiveMessageHandler handler ) throws IOException { if (isReceiving()) { throw new IllegalStateException("Already receiving message."); @@ -1079,7 +1091,7 @@ public class ManagerImpl implements Manager { isReceivingSynchronous = true; receiveThread = Thread.currentThread(); try { - receiveMessagesInternal(timeout, unit, returnOnTimeout, handler); + receiveMessagesInternal(timeout, returnOnTimeout, handler); } finally { receiveThread = null; hasCaughtUpWithOldMessages = false; @@ -1088,7 +1100,7 @@ public class ManagerImpl implements Manager { } private void receiveMessagesInternal( - long timeout, TimeUnit unit, boolean returnOnTimeout, ReceiveMessageHandler handler + Duration timeout, boolean returnOnTimeout, ReceiveMessageHandler handler ) throws IOException { retryFailedReceivedMessages(handler); @@ -1096,6 +1108,12 @@ public class ManagerImpl implements Manager { Map queuedActions = new HashMap<>(); final var signalWebSocket = dependencies.getSignalWebSocket(); + final var webSocketStateDisposable = Observable.merge(signalWebSocket.getUnidentifiedWebSocketState(), + signalWebSocket.getWebSocketState()) + .subscribeOn(Schedulers.computation()) + .observeOn(Schedulers.computation()) + .distinctUntilChanged() + .subscribe(this::onWebSocketStateChange); signalWebSocket.connect(); hasCaughtUpWithOldMessages = false; @@ -1111,7 +1129,7 @@ public class ManagerImpl implements Manager { } logger.debug("Checking for new message from server"); try { - var result = signalWebSocket.readOrEmpty(unit.toMillis(timeout), envelope1 -> { + var result = signalWebSocket.readOrEmpty(timeout.toMillis(), envelope1 -> { final var recipientId = envelope1.hasSourceUuid() ? resolveRecipient(envelope1.getSourceAddress()) : null; @@ -1201,6 +1219,18 @@ public class ManagerImpl implements Manager { handleQueuedActions(queuedActions.keySet()); queuedActions.clear(); dependencies.getSignalWebSocket().disconnect(); + webSocketStateDisposable.dispose(); + } + + private void onWebSocketStateChange(final WebSocketConnectionState s) { + if (s.equals(WebSocketConnectionState.AUTHENTICATION_FAILED)) { + account.setRegistered(false); + try { + close(); + } catch (IOException e) { + e.printStackTrace(); + } + } } @Override @@ -1255,7 +1285,7 @@ public class ManagerImpl implements Manager { .getContacts() .stream() .map(p -> new Pair<>(account.getRecipientStore().resolveRecipientAddress(p.first()), p.second())) - .collect(Collectors.toList()); + .toList(); } @Override @@ -1291,11 +1321,7 @@ public class ManagerImpl implements Manager { @Override public List getIdentities() { - return account.getIdentityKeyStore() - .getIdentities() - .stream() - .map(this::toIdentity) - .collect(Collectors.toList()); + return account.getIdentityKeyStore().getIdentities().stream().map(this::toIdentity).toList(); } private Identity toIdentity(final IdentityInfo identityInfo) { @@ -1408,13 +1434,13 @@ public class ManagerImpl implements Manager { private SignalServiceAddress resolveSignalServiceAddress(RecipientId recipientId) { final var address = account.getRecipientStore().resolveRecipientAddress(recipientId); - if (address.getUuid().isPresent()) { + if (address.uuid().isPresent()) { return address.toSignalServiceAddress(); } // Address in recipient store doesn't have a uuid, this shouldn't happen // Try to retrieve the uuid from the server - final var number = address.getNumber().get(); + final var number = address.number().get(); final ACI aci; try { aci = getRegisteredUser(number);