X-Git-Url: https://git.nmode.ca/signal-cli/blobdiff_plain/fc2e9bbfaec8dbe6e317b951c0d28fa2e9421347..5d33f71d4d337e45c6051274b404424562503938:/src/main/java/org/asamk/signal/dbus/DbusManagerImpl.java

diff --git a/src/main/java/org/asamk/signal/dbus/DbusManagerImpl.java b/src/main/java/org/asamk/signal/dbus/DbusManagerImpl.java
index 49aa0e27..cd6c6715 100644
--- a/src/main/java/org/asamk/signal/dbus/DbusManagerImpl.java
+++ b/src/main/java/org/asamk/signal/dbus/DbusManagerImpl.java
@@ -3,6 +3,7 @@ package org.asamk.signal.dbus;
 import org.asamk.Signal;
 import org.asamk.signal.DbusConfig;
 import org.asamk.signal.manager.Manager;
+import org.asamk.signal.manager.api.AlreadyReceivingException;
 import org.asamk.signal.manager.api.AttachmentInvalidException;
 import org.asamk.signal.manager.api.CaptchaRequiredException;
 import org.asamk.signal.manager.api.Configuration;
@@ -548,10 +549,17 @@ public class DbusManagerImpl implements Manager {
         }
     }
 
+    private Thread receiveThread;
+
     @Override
     public void receiveMessages(
             Optional<Duration> timeout, Optional<Integer> maxMessages, ReceiveMessageHandler handler
-    ) throws IOException {
+    ) throws IOException, AlreadyReceivingException {
+        if (receiveThread != null) {
+            throw new AlreadyReceivingException("Already receiving message.");
+        }
+        receiveThread = Thread.currentThread();
+
         final var remainingMessages = new AtomicInteger(maxMessages.orElse(-1));
         final var lastMessage = new AtomicLong(System.currentTimeMillis());
         final var thread = Thread.currentThread();
@@ -577,6 +585,7 @@ public class DbusManagerImpl implements Manager {
                     }
                     Thread.sleep(sleepTimeRemaining);
                 } catch (InterruptedException ignored) {
+                    break;
                 }
             }
         } else {
@@ -589,6 +598,14 @@ public class DbusManagerImpl implements Manager {
         }
 
         removeReceiveHandler(receiveHandler);
+        receiveThread = null;
+    }
+
+    @Override
+    public void stopReceiveMessages() {
+        if (receiveThread != null) {
+            receiveThread.interrupt();
+        }
     }
 
     @Override
@@ -948,7 +965,7 @@ public class DbusManagerImpl implements Manager {
             };
             connection.addSigHandler(Signal.SyncMessageReceivedV2.class, signal, this.dbusSyncHandler);
         } catch (DBusException e) {
-            e.printStackTrace();
+            throw new RuntimeException(e);
         }
         signal.subscribeReceive();
     }
@@ -968,7 +985,7 @@ public class DbusManagerImpl implements Manager {
             connection.removeSigHandler(Signal.ReceiptReceivedV2.class, signal, this.dbusRcptHandler);
             connection.removeSigHandler(Signal.SyncMessageReceivedV2.class, signal, this.dbusSyncHandler);
         } catch (DBusException e) {
-            e.printStackTrace();
+            throw new RuntimeException(e);
         }
     }