From: AsamK <asamk@gmx.de>
Date: Thu, 9 Nov 2023 18:22:58 +0000 (+0100)
Subject: Use improved shutdown for receive command
X-Git-Tag: v0.13.0~98
X-Git-Url: https://git.nmode.ca/signal-cli/commitdiff_plain/5d33f71d4d337e45c6051274b404424562503938?hp=1058e33f1208e1ca284d3fd4760380d68093741f

Use improved shutdown for receive command
---

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 41e08fb6..da69a8da 100644
--- a/lib/src/main/java/org/asamk/signal/manager/Manager.java
+++ b/lib/src/main/java/org/asamk/signal/manager/Manager.java
@@ -255,6 +255,8 @@ public interface Manager extends Closeable {
             Optional<Duration> timeout, Optional<Integer> maxMessages, ReceiveMessageHandler handler
     ) throws IOException, AlreadyReceivingException;
 
+    void stopReceiveMessages();
+
     void setReceiveConfig(ReceiveConfig receiveConfig);
 
     boolean isContactBlocked(RecipientIdentifier.Single recipient);
diff --git a/lib/src/main/java/org/asamk/signal/manager/internal/ManagerImpl.java b/lib/src/main/java/org/asamk/signal/manager/internal/ManagerImpl.java
index cbaaaab3..af001f40 100644
--- a/lib/src/main/java/org/asamk/signal/manager/internal/ManagerImpl.java
+++ b/lib/src/main/java/org/asamk/signal/manager/internal/ManagerImpl.java
@@ -1091,6 +1091,20 @@ public class ManagerImpl implements Manager {
         receiveMessages(timeout.orElse(Duration.ofMinutes(1)), timeout.isPresent(), maxMessages.orElse(null), handler);
     }
 
+    @Override
+    public void stopReceiveMessages() {
+        Thread thread = null;
+        synchronized (messageHandlers) {
+            if (isReceivingSynchronous) {
+                thread = receiveThread;
+                receiveThread = null;
+            }
+        }
+        if (thread != null) {
+            stopReceiveThread(thread);
+        }
+    }
+
     private void receiveMessages(
             Duration timeout, boolean returnOnTimeout, Integer maxMessages, ReceiveMessageHandler handler
     ) throws IOException, AlreadyReceivingException {
diff --git a/src/main/java/org/asamk/signal/commands/ReceiveCommand.java b/src/main/java/org/asamk/signal/commands/ReceiveCommand.java
index 228a0561..4a6f746a 100644
--- a/src/main/java/org/asamk/signal/commands/ReceiveCommand.java
+++ b/src/main/java/org/asamk/signal/commands/ReceiveCommand.java
@@ -8,6 +8,7 @@ import net.sourceforge.argparse4j.inf.Subparser;
 
 import org.asamk.signal.OutputType;
 import org.asamk.signal.ReceiveMessageHandler;
+import org.asamk.signal.Shutdown;
 import org.asamk.signal.commands.exceptions.CommandException;
 import org.asamk.signal.commands.exceptions.IOErrorException;
 import org.asamk.signal.commands.exceptions.UserErrorException;
@@ -67,6 +68,7 @@ public class ReceiveCommand implements LocalCommand, JsonRpcSingleCommand<Receiv
     public void handleCommand(
             final Namespace ns, final Manager m, final OutputWriter outputWriter
     ) throws CommandException {
+        Shutdown.installHandler();
         final var timeout = ns.getDouble("timeout");
         final var maxMessagesRaw = ns.getInt("max-messages");
         final var ignoreAttachments = Boolean.TRUE.equals(ns.getBoolean("ignore-attachments"));
@@ -80,6 +82,7 @@ public class ReceiveCommand implements LocalCommand, JsonRpcSingleCommand<Receiv
             };
             final var duration = timeout < 0 ? null : Duration.ofMillis((long) (timeout * 1000));
             final var maxMessages = maxMessagesRaw < 0 ? null : maxMessagesRaw;
+            Shutdown.registerShutdownListener(m::stopReceiveMessages);
             m.receiveMessages(Optional.ofNullable(duration), Optional.ofNullable(maxMessages), handler);
         } catch (IOException e) {
             throw new IOErrorException("Error while receiving messages: " + e.getMessage(), e);
diff --git a/src/main/java/org/asamk/signal/dbus/DbusManagerImpl.java b/src/main/java/org/asamk/signal/dbus/DbusManagerImpl.java
index e9fc3f9c..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