Merge multiple SendReceiptActions to same recipient to only send one receipt

[signal-cli] / 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 94a3f4bea4477617d897300c9f5e323e54c5c5e5..756c0a9474894ff8ca770c6b7b8b62f727f74fe9 100644 (file)
--- a/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java
+++ b/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java
@@ -108,6 +108,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,6 +140,7 @@ public class ManagerImpl implements Manager {
     private boolean ignoreAttachments = false;
 
     private Thread receiveThread;
+    private final Set<ReceiveMessageHandler> weakHandlers = new HashSet<>();
     private final Set<ReceiveMessageHandler> messageHandlers = new HashSet<>();
     private boolean isReceivingSynchronous;
 
@@ -401,6 +403,8 @@ public class ManagerImpl implements Manager {
 
     @Override
     public void submitRateLimitRecaptchaChallenge(String challenge, String captcha) throws IOException {
+        captcha = captcha == null ? null : captcha.replace("signalcaptcha://", "");
+
         dependencies.getAccountManager().submitRateLimitRecaptchaChallenge(challenge, captcha);
     }
 
@@ -904,14 +908,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();
+            }
         }
     }
 
@@ -925,13 +932,13 @@ public class ManagerImpl implements Manager {
                 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;
@@ -959,8 +966,9 @@ public class ManagerImpl implements Manager {
     public void removeReceiveHandler(final ReceiveMessageHandler handler) {
         final Thread thread;
         synchronized (messageHandlers) {
+            weakHandlers.remove(handler);
             messageHandlers.remove(handler);
-            if (!messageHandlers.isEmpty() || isReceivingSynchronous) {
+            if (!messageHandlers.isEmpty() || receiveThread == null || isReceivingSynchronous) {
                 return;
             }
             thread = receiveThread;
@@ -1020,7 +1028,8 @@ public class ManagerImpl implements Manager {
     ) throws IOException {
         retryFailedReceivedMessages(handler);
 
-        Set<HandleAction> queuedActions = new HashSet<>();
+        // Use a Map here because java Set doesn't have a get method ...
+        Map<HandleAction, HandleAction> queuedActions = new HashMap<>();
 
         final var signalWebSocket = dependencies.getSignalWebSocket();
         signalWebSocket.connect();
@@ -1049,7 +1058,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;
@@ -1090,11 +1099,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) {
@@ -1115,7 +1131,7 @@ public class ManagerImpl implements Manager {
                 }
             }
         }
-        handleQueuedActions(queuedActions);
+        handleQueuedActions(queuedActions.keySet());
         queuedActions.clear();
         dependencies.getSignalWebSocket().disconnect();
     }
@@ -1380,6 +1396,7 @@ public class ManagerImpl implements Manager {
     private void close(boolean closeAccount) throws IOException {
         Thread thread;
         synchronized (messageHandlers) {
+            weakHandlers.clear();
             messageHandlers.clear();
             thread = receiveThread;
             receiveThread = null;