Only handle jsonRpc requests, after receive thread has caught up with old messages

diff --git a/lib/src/main/java/org/asamk/signal/manager/Manager.java b/lib/src/main/java/org/asamk/signal/manager/Manager.java
index c40fa7cdb2c2800f5f9d55323e5a30f9056d8521..c4c77b3453c4a2764a9ecc03e17da4afd1b674ec 100644 (file)
--- a/lib/src/main/java/org/asamk/signal/manager/Manager.java
+++ b/lib/src/main/java/org/asamk/signal/manager/Manager.java
@@ -141,6 +141,7 @@ public class Manager implements Closeable {
     private final IncomingMessageHandler incomingMessageHandler;
 
     private final Context context;
     private final IncomingMessageHandler incomingMessageHandler;
 
     private final Context context;

+    private boolean hasCaughtUpWithOldMessages = false;

 
     Manager(
             SignalAccount account,
 
     Manager(
             SignalAccount account,


@@ -865,7 +866,7 @@ public class Manager implements Closeable {
         final var signalWebSocket = dependencies.getSignalWebSocket();
         signalWebSocket.connect();
 
         final var signalWebSocket = dependencies.getSignalWebSocket();
         signalWebSocket.connect();
 

-        var hasCaughtUpWithOldMessages = false;
+        hasCaughtUpWithOldMessages = false;

 
         while (!Thread.interrupted()) {
             SignalServiceEnvelope envelope;
 
         while (!Thread.interrupted()) {
             SignalServiceEnvelope envelope;


@@ -885,11 +886,14 @@ public class Manager implements Closeable {
                     envelope = result.get();
                 } else {
                     // Received indicator that server queue is empty
                     envelope = result.get();
                 } else {
                     // Received indicator that server queue is empty

-                    hasCaughtUpWithOldMessages = true;
-

                     handleQueuedActions(queuedActions);
                     queuedActions.clear();
 
                     handleQueuedActions(queuedActions);
                     queuedActions.clear();
 

+                    hasCaughtUpWithOldMessages = true;
+                    synchronized (this) {
+                        this.notifyAll();
+                    }
+

                     // Continue to wait another timeout for new messages
                     continue;
                 }
                     // Continue to wait another timeout for new messages
                     continue;
                 }


@@ -936,17 +940,27 @@ public class Manager implements Closeable {
         handleQueuedActions(queuedActions);
     }
 
         handleQueuedActions(queuedActions);
     }
 

+    public boolean hasCaughtUpWithOldMessages() {
+        return hasCaughtUpWithOldMessages;
+    }
+

     private void handleQueuedActions(final Collection<HandleAction> queuedActions) {
     private void handleQueuedActions(final Collection<HandleAction> queuedActions) {

+        var interrupted = false;

         for (var action : queuedActions) {
             try {
                 action.execute(context);
             } catch (Throwable e) {
         for (var action : queuedActions) {
             try {
                 action.execute(context);
             } catch (Throwable e) {

-                if (e instanceof AssertionError && e.getCause() instanceof InterruptedException) {
-                    Thread.currentThread().interrupt();
+                if ((e instanceof AssertionError || e instanceof RuntimeException)
+                        && e.getCause() instanceof InterruptedException) {
+                    interrupted = true;
+                    continue;

                 }
                 logger.warn("Message action failed.", e);
             }
         }
                 }
                 logger.warn("Message action failed.", e);
             }
         }

+        if (interrupted) {
+            Thread.currentThread().interrupt();
+        }

     }
 
     public boolean isContactBlocked(final RecipientIdentifier.Single recipient) {
     }
 
     public boolean isContactBlocked(final RecipientIdentifier.Single recipient) {

diff --git a/src/main/java/org/asamk/signal/commands/JsonRpcDispatcherCommand.java b/src/main/java/org/asamk/signal/commands/JsonRpcDispatcherCommand.java
index 16d0cf717621e03aa50f9655423cfcd56579f8a5..d0e4dfec353325954d640e20457c968e8a8c470f 100644 (file)
--- a/src/main/java/org/asamk/signal/commands/JsonRpcDispatcherCommand.java
+++ b/src/main/java/org/asamk/signal/commands/JsonRpcDispatcherCommand.java
@@ -75,6 +75,16 @@ public class JsonRpcDispatcherCommand implements LocalCommand {
                 objectMapper.valueToTree(s),
                 null)), m, ignoreAttachments);
 
                 objectMapper.valueToTree(s),
                 null)), m, ignoreAttachments);
 

+        // Maybe this should be handled inside the Manager
+        while (!m.hasCaughtUpWithOldMessages()) {
+            try {
+                synchronized (m) {
+                    m.wait();
+                }
+            } catch (InterruptedException ignored) {
+            }
+        }
+

         final BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
 
         final var jsonRpcReader = new JsonRpcReader(jsonRpcSender, () -> {
         final BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
 
         final var jsonRpcReader = new JsonRpcReader(jsonRpcSender, () -> {