X-Git-Url: https://git.nmode.ca/signal-cli/blobdiff_plain/ff6b733cd0448c05f4be5aad32895cc8c748ee79..3f582e9c2e7db9adeb48dbc569f3c32509a63ad5:/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 742be520..941a6122 100644
--- a/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java
+++ b/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java
@@ -70,7 +70,6 @@ import org.whispersystems.signalservice.api.messages.SignalServiceTypingMessage;
 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.internal.util.DynamicCredentialsProvider;
 import org.whispersystems.signalservice.internal.util.Hex;
 import org.whispersystems.signalservice.internal.util.Util;
 
@@ -100,7 +99,6 @@ class ManagerImpl implements Manager {
     private final static Logger logger = LoggerFactory.getLogger(ManagerImpl.class);
 
     private SignalAccount account;
-    private final AccountFileUpdater accountFileUpdater;
     private final SignalDependencies dependencies;
     private final Context context;
 
@@ -111,6 +109,7 @@ class ManagerImpl implements Manager {
     private final Set<ReceiveMessageHandler> weakHandlers = new HashSet<>();
     private final Set<ReceiveMessageHandler> messageHandlers = new HashSet<>();
     private final List<Runnable> closedListeners = new ArrayList<>();
+    private final List<Runnable> addressChangedListeners = new ArrayList<>();
     private final CompositeDisposable disposable = new CompositeDisposable();
 
     ManagerImpl(
@@ -121,12 +120,7 @@ class ManagerImpl implements Manager {
             String userAgent
     ) {
         this.account = account;
-        this.accountFileUpdater = accountFileUpdater;
 
-        final var credentialsProvider = new DynamicCredentialsProvider(account.getAci(),
-                account.getNumber(),
-                account.getPassword(),
-                account.getDeviceId());
         final var sessionLock = new SignalSessionLock() {
             private final ReentrantLock LEGACY_LOCK = new ReentrantLock();
 
@@ -138,20 +132,20 @@ class ManagerImpl implements Manager {
         };
         this.dependencies = new SignalDependencies(serviceEnvironmentConfig,
                 userAgent,
-                credentialsProvider,
-                account.getSignalProtocolStore(),
+                account.getCredentialsProvider(),
+                account.getSignalServiceDataStore(),
                 executor,
                 sessionLock);
         final var avatarStore = new AvatarStore(pathConfig.avatarsPath());
         final var attachmentStore = new AttachmentStore(pathConfig.attachmentsPath());
         final var stickerPackStore = new StickerPackStore(pathConfig.stickerPacksPath());
 
-        this.context = new Context(account,
-                accountFileUpdater,
-                dependencies,
-                avatarStore,
-                attachmentStore,
-                stickerPackStore);
+        this.context = new Context(account, (number, aci) -> {
+            accountFileUpdater.updateAccountIdentifiers(number, aci);
+            synchronized (addressChangedListeners) {
+                addressChangedListeners.forEach(Runnable::run);
+            }
+        }, dependencies, avatarStore, attachmentStore, stickerPackStore);
         this.context.getAccountHelper().setUnregisteredListener(this::close);
         this.context.getReceiveHelper().setAuthenticationFailureListener(this::close);
         this.context.getReceiveHelper().setCaughtUpWithOldMessagesListener(() -> {
@@ -771,24 +765,17 @@ class ManagerImpl implements Manager {
         }
         receiveThread = new Thread(() -> {
             logger.debug("Starting receiving messages");
-            while (!Thread.interrupted()) {
-                try {
-                    context.getReceiveHelper().receiveMessages(Duration.ofMinutes(1), false, (envelope, e) -> {
-                        synchronized (messageHandlers) {
-                            Stream.concat(messageHandlers.stream(), weakHandlers.stream()).forEach(h -> {
-                                try {
-                                    h.handleMessage(envelope, e);
-                                } catch (Exception ex) {
-                                    logger.warn("Message handler failed, ignoring", ex);
-                                }
-                            });
+            context.getReceiveHelper().receiveMessagesContinuously((envelope, e) -> {
+                synchronized (messageHandlers) {
+                    Stream.concat(messageHandlers.stream(), weakHandlers.stream()).forEach(h -> {
+                        try {
+                            h.handleMessage(envelope, e);
+                        } catch (Throwable ex) {
+                            logger.warn("Message handler failed, ignoring", ex);
                         }
                     });
-                    break;
-                } catch (IOException e) {
-                    logger.warn("Receiving messages failed, retrying", e);
                 }
-            }
+            });
             logger.debug("Finished receiving messages");
             synchronized (messageHandlers) {
                 receiveThread = null;
@@ -822,7 +809,10 @@ class ManagerImpl implements Manager {
     }
 
     private void stopReceiveThread(final Thread thread) {
-        thread.interrupt();
+        if (context.getReceiveHelper().requestStopReceiveMessages()) {
+            logger.debug("Receive stop requested, interrupting read from server.");
+            thread.interrupt();
+        }
         try {
             thread.join();
         } catch (InterruptedException ignored) {
@@ -1005,6 +995,13 @@ class ManagerImpl implements Manager {
         return updated;
     }
 
+    @Override
+    public void addAddressChangedListener(final Runnable listener) {
+        synchronized (addressChangedListeners) {
+            addressChangedListeners.add(listener);
+        }
+    }
+
     @Override
     public void addClosedListener(final Runnable listener) {
         synchronized (closedListeners) {
@@ -1029,14 +1026,15 @@ class ManagerImpl implements Manager {
         dependencies.getSignalWebSocket().disconnect();
         disposable.dispose();
 
+        if (account != null) {
+            account.close();
+        }
+
         synchronized (closedListeners) {
             closedListeners.forEach(Runnable::run);
             closedListeners.clear();
         }
 
-        if (account != null) {
-            account.close();
-        }
         account = null;
     }
 }