Improve behavior with synchronous and asynchronous receivers

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 ec168a4c345af2c3cc8b8ba34d6dd2195ff0da43..01d88e46c43b669c13395ad77f30cdb01a58c0ca 100644 (file)
--- a/lib/src/main/java/org/asamk/signal/manager/Manager.java
+++ b/lib/src/main/java/org/asamk/signal/manager/Manager.java
@@ -1,5 +1,6 @@
 package org.asamk.signal.manager;
 
 package org.asamk.signal.manager;
 

+import org.asamk.signal.manager.api.AlreadyReceivingException;

 import org.asamk.signal.manager.api.AttachmentInvalidException;
 import org.asamk.signal.manager.api.Configuration;
 import org.asamk.signal.manager.api.Device;
 import org.asamk.signal.manager.api.AttachmentInvalidException;
 import org.asamk.signal.manager.api.Configuration;
 import org.asamk.signal.manager.api.Device;


@@ -204,7 +205,7 @@ public interface Manager extends Closeable {
      */
     public void receiveMessages(
             Optional<Duration> timeout, Optional<Integer> maxMessages, ReceiveMessageHandler handler
      */
     public void receiveMessages(
             Optional<Duration> timeout, Optional<Integer> maxMessages, ReceiveMessageHandler handler

-    ) throws IOException;
+    ) throws IOException, AlreadyReceivingException;

 
     void setReceiveConfig(ReceiveConfig receiveConfig);
 
 
     void setReceiveConfig(ReceiveConfig receiveConfig);
 

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 95f5bde4e710d62bf9f9b538c2624f027bf5ec2f..3123163022a531a327790aa68052fc4a391f0a7d 100644 (file)
--- a/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java
+++ b/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java
@@ -16,6 +16,7 @@
  */
 package org.asamk.signal.manager;
 
  */
 package org.asamk.signal.manager;
 

+import org.asamk.signal.manager.api.AlreadyReceivingException;

 import org.asamk.signal.manager.api.AttachmentInvalidException;
 import org.asamk.signal.manager.api.Configuration;
 import org.asamk.signal.manager.api.Device;
 import org.asamk.signal.manager.api.AttachmentInvalidException;
 import org.asamk.signal.manager.api.Configuration;
 import org.asamk.signal.manager.api.Device;


@@ -25,6 +26,7 @@ import org.asamk.signal.manager.api.InactiveGroupLinkException;
 import org.asamk.signal.manager.api.InvalidDeviceLinkException;
 import org.asamk.signal.manager.api.InvalidStickerException;
 import org.asamk.signal.manager.api.Message;
 import org.asamk.signal.manager.api.InvalidDeviceLinkException;
 import org.asamk.signal.manager.api.InvalidStickerException;
 import org.asamk.signal.manager.api.Message;

+import org.asamk.signal.manager.api.MessageEnvelope;

 import org.asamk.signal.manager.api.NotPrimaryDeviceException;
 import org.asamk.signal.manager.api.Pair;
 import org.asamk.signal.manager.api.PendingAdminApprovalException;
 import org.asamk.signal.manager.api.NotPrimaryDeviceException;
 import org.asamk.signal.manager.api.Pair;
 import org.asamk.signal.manager.api.PendingAdminApprovalException;


@@ -874,9 +876,6 @@ class ManagerImpl implements Manager {
 
     @Override
     public void addReceiveHandler(final ReceiveMessageHandler handler, final boolean isWeakListener) {
 
     @Override
     public void addReceiveHandler(final ReceiveMessageHandler handler, final boolean isWeakListener) {

-        if (isReceivingSynchronous) {
-            throw new IllegalStateException("Already receiving message synchronously.");
-        }

         synchronized (messageHandlers) {
             if (isWeakListener) {
                 weakHandlers.add(handler);
         synchronized (messageHandlers) {
             if (isWeakListener) {
                 weakHandlers.add(handler);


@@ -890,23 +889,12 @@ class ManagerImpl implements Manager {
     private static final AtomicInteger threadNumber = new AtomicInteger(0);
 
     private void startReceiveThreadIfRequired() {
     private static final AtomicInteger threadNumber = new AtomicInteger(0);
 
     private void startReceiveThreadIfRequired() {

-        if (receiveThread != null) {
+        if (receiveThread != null || isReceivingSynchronous) {

             return;
         }
         receiveThread = new Thread(() -> {
             logger.debug("Starting receiving messages");
             return;
         }
         receiveThread = new Thread(() -> {
             logger.debug("Starting receiving messages");

-            context.getReceiveHelper().receiveMessagesContinuously((envelope, e) -> {
-                synchronized (messageHandlers) {
-                    final var handlers = Stream.concat(messageHandlers.stream(), weakHandlers.stream()).toList();
-                    handlers.forEach(h -> {
-                        try {
-                            h.handleMessage(envelope, e);
-                        } catch (Throwable ex) {
-                            logger.warn("Message handler failed, ignoring", ex);
-                        }
-                    });
-                }
-            });
+            context.getReceiveHelper().receiveMessagesContinuously(this::passReceivedMessageToHandlers);

             logger.debug("Finished receiving messages");
             synchronized (messageHandlers) {
                 receiveThread = null;
             logger.debug("Finished receiving messages");
             synchronized (messageHandlers) {
                 receiveThread = null;


@@ -923,6 +911,18 @@ class ManagerImpl implements Manager {
         receiveThread.start();
     }
 
         receiveThread.start();
     }
 

+    private void passReceivedMessageToHandlers(MessageEnvelope envelope, Throwable 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);
+                }
+            });
+        }
+    }
+

     @Override
     public void removeReceiveHandler(final ReceiveMessageHandler handler) {
         final Thread thread;
     @Override
     public void removeReceiveHandler(final ReceiveMessageHandler handler) {
         final Thread thread;


@@ -962,26 +962,34 @@ class ManagerImpl implements Manager {
 
     @Override
     public void receiveMessages(
 
     @Override
     public void receiveMessages(

-            Optional<Duration> timeout,
-            Optional<Integer> maxMessages,
-            ReceiveMessageHandler handler
-    ) throws IOException {
+            Optional<Duration> timeout, Optional<Integer> maxMessages, ReceiveMessageHandler handler
+    ) throws IOException, AlreadyReceivingException {

         receiveMessages(timeout.orElse(Duration.ofMinutes(1)), timeout.isPresent(), maxMessages.orElse(null), handler);
     }
 
     private void receiveMessages(
             Duration timeout, boolean returnOnTimeout, Integer maxMessages, ReceiveMessageHandler handler
         receiveMessages(timeout.orElse(Duration.ofMinutes(1)), timeout.isPresent(), maxMessages.orElse(null), handler);
     }
 
     private void receiveMessages(
             Duration timeout, boolean returnOnTimeout, Integer maxMessages, ReceiveMessageHandler handler

-    ) throws IOException {
-        if (isReceiving()) {
-            throw new IllegalStateException("Already receiving message.");
+    ) throws IOException, AlreadyReceivingException {
+        synchronized (messageHandlers) {
+            if (isReceiving()) {
+                throw new AlreadyReceivingException("Already receiving message.");
+            }
+            isReceivingSynchronous = true;
+            receiveThread = Thread.currentThread();

         }
         }

-        isReceivingSynchronous = true;
-        receiveThread = Thread.currentThread();

         try {
         try {

-            context.getReceiveHelper().receiveMessages(timeout, returnOnTimeout, maxMessages, handler);
+            context.getReceiveHelper().receiveMessages(timeout, returnOnTimeout, maxMessages, (envelope, e) -> {
+                passReceivedMessageToHandlers(envelope, e);
+                handler.handleMessage(envelope, e);
+            });

         } finally {
         } finally {

-            receiveThread = null;
-            isReceivingSynchronous = false;
+            synchronized (messageHandlers) {
+                receiveThread = null;
+                isReceivingSynchronous = false;
+                if (messageHandlers.size() > 0) {
+                    startReceiveThreadIfRequired();
+                }
+            }

         }
     }
 
         }
     }
 

diff --git a/lib/src/main/java/org/asamk/signal/manager/api/AlreadyReceivingException.java b/lib/src/main/java/org/asamk/signal/manager/api/AlreadyReceivingException.java
new file mode 100644 (file)
index 0000000..298e3f2
--- /dev/null
+++ b/lib/src/main/java/org/asamk/signal/manager/api/AlreadyReceivingException.java
@@ -0,0 +1,12 @@
+package org.asamk.signal.manager.api;
+
+public class AlreadyReceivingException extends Exception {
+
+    public AlreadyReceivingException(String message) {
+        super(message);
+    }
+
+    public AlreadyReceivingException(String message, Exception e) {
+        super(message, e);
+    }
+}

diff --git a/src/main/java/org/asamk/signal/commands/ReceiveCommand.java b/src/main/java/org/asamk/signal/commands/ReceiveCommand.java
index 79d0ee1dad5dad95058c98fa1820ce66d434bec4..ab65f8377eeccf8fb1181f62ecffc9ceb38089f3 100644 (file)
--- a/src/main/java/org/asamk/signal/commands/ReceiveCommand.java
+++ b/src/main/java/org/asamk/signal/commands/ReceiveCommand.java
@@ -10,8 +10,10 @@ import org.asamk.signal.OutputType;
 import org.asamk.signal.ReceiveMessageHandler;
 import org.asamk.signal.commands.exceptions.CommandException;
 import org.asamk.signal.commands.exceptions.IOErrorException;
 import org.asamk.signal.ReceiveMessageHandler;
 import org.asamk.signal.commands.exceptions.CommandException;
 import org.asamk.signal.commands.exceptions.IOErrorException;

+import org.asamk.signal.commands.exceptions.UserErrorException;

 import org.asamk.signal.json.JsonReceiveMessageHandler;
 import org.asamk.signal.manager.Manager;
 import org.asamk.signal.json.JsonReceiveMessageHandler;
 import org.asamk.signal.manager.Manager;

+import org.asamk.signal.manager.api.AlreadyReceivingException;

 import org.asamk.signal.manager.api.ReceiveConfig;
 import org.asamk.signal.output.JsonWriter;
 import org.asamk.signal.output.OutputWriter;
 import org.asamk.signal.manager.api.ReceiveConfig;
 import org.asamk.signal.output.JsonWriter;
 import org.asamk.signal.output.OutputWriter;


@@ -79,6 +81,8 @@ public class ReceiveCommand implements LocalCommand, JsonRpcSingleCommand<Receiv
             m.receiveMessages(Optional.ofNullable(duration), Optional.ofNullable(maxMessages), handler);
         } catch (IOException e) {
             throw new IOErrorException("Error while receiving messages: " + e.getMessage(), e);
             m.receiveMessages(Optional.ofNullable(duration), Optional.ofNullable(maxMessages), handler);
         } catch (IOException e) {
             throw new IOErrorException("Error while receiving messages: " + e.getMessage(), e);

+        } catch (AlreadyReceivingException e) {
+            throw new UserErrorException("Receive command cannot be used if messages are already being received.", e);

         }
     }
 
         }
     }
 


@@ -103,6 +107,8 @@ public class ReceiveCommand implements LocalCommand, JsonRpcSingleCommand<Receiv
             jsonWriter.write(messages);
         } catch (IOException e) {
             throw new IOErrorException("Error while receiving messages: " + e.getMessage(), e);
             jsonWriter.write(messages);
         } catch (IOException e) {
             throw new IOErrorException("Error while receiving messages: " + e.getMessage(), e);

+        } catch (AlreadyReceivingException e) {
+            throw new UserErrorException("Receive command cannot be used if messages are already being received.", e);

         }
     }
 
         }
     }