diff --git a/lib/src/main/java/org/asamk/signal/manager/helper/ReceiveHelper.java b/lib/src/main/java/org/asamk/signal/manager/helper/ReceiveHelper.java index 8d44f4fb..f97bd18b 100644 --- a/lib/src/main/java/org/asamk/signal/manager/helper/ReceiveHelper.java +++ b/lib/src/main/java/org/asamk/signal/manager/helper/ReceiveHelper.java @@ -42,8 +42,8 @@ public class ReceiveHelper { private ReceiveConfig receiveConfig = new ReceiveConfig(false, false, false, false, false); private boolean hasCaughtUpWithOldMessages = false; - private boolean isWaitingForMessage = false; - private boolean shouldStop = false; + private volatile boolean isWaitingForMessage = false; + private volatile boolean shouldStop = false; private Callable authenticationFailureListener; private Callable caughtUpWithOldMessagesListener; @@ -71,6 +71,15 @@ public class ReceiveHelper { return isWaitingForMessage; } + /** + * Must only be called right before a new receive loop is started, while no other receive loop is running. + * Ending a receive loop doesn't clear the stop request, so a loop that was asked to stop can't continue + * receiving, e.g. by retrying in receiveMessagesContinuously after an IOException. + */ + public void clearStopRequest() { + this.shouldStop = false; + } + public void receiveMessagesContinuously(Manager.ReceiveMessageHandler handler) { while (!shouldStop) { try { @@ -111,7 +120,6 @@ public class ReceiveHelper { signalWebSocket.removeKeepAliveToken("receive"); signalWebSocket.disconnect(); webSocketStateDisposable.dispose(); - shouldStop = false; } } 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 2026c527..8658f365 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 @@ -1510,15 +1510,18 @@ public class ManagerImpl implements Manager { private static final AtomicInteger threadNumber = new AtomicInteger(0); private void startReceiveThreadIfRequired() { - if (receiveThread != null || isReceivingSynchronous) { + if (isReceiveThreadAlive() || isReceivingSynchronous) { return; } + context.getReceiveHelper().clearStopRequest(); receiveThread = Thread.ofPlatform().name("receive-" + threadNumber.getAndIncrement()).start(() -> { logger.debug("Starting receiving messages"); context.getReceiveHelper().receiveMessagesContinuously(this::passReceivedMessageToHandlers); logger.debug("Finished receiving messages"); synchronized (messageHandlers) { - receiveThread = null; + if (receiveThread == Thread.currentThread()) { + receiveThread = null; + } // Check if in the meantime another handler has been registered if (!messageHandlers.isEmpty()) { @@ -1529,6 +1532,13 @@ public class ManagerImpl implements Manager { }); } + private boolean isReceiveThreadAlive() { + // The receive thread clears receiveThread itself when it has finished, so it stays set while the thread is + // still stopping and no second receive thread can be started in the meantime. + // A receive thread that died from an uncaught exception never clears it. + return receiveThread != null && receiveThread.isAlive(); + } + private void passReceivedMessageToHandlers(MessageEnvelope envelope, Throwable e) { synchronized (messageHandlers) { Stream.concat(messageHandlers.stream(), weakHandlers.stream()).forEach(h -> { @@ -1551,7 +1561,6 @@ public class ManagerImpl implements Manager { return; } thread = receiveThread; - receiveThread = null; } stopReceiveThread(thread); @@ -1607,11 +1616,12 @@ public class ManagerImpl implements Manager { ReceiveMessageHandler handler ) throws IOException, AlreadyReceivingException { synchronized (messageHandlers) { - if (isReceiving()) { + if (isReceiving() || isReceiveThreadAlive()) { throw new AlreadyReceivingException("Already receiving message."); } isReceivingSynchronous = true; receiveThread = Thread.currentThread(); + context.getReceiveHelper().clearStopRequest(); } try { context.getReceiveHelper().receiveMessages(timeout, maxMessages, (envelope, e) -> { @@ -2036,7 +2046,6 @@ public class ManagerImpl implements Manager { weakHandlers.clear(); messageHandlers.clear(); thread = receiveThread; - receiveThread = null; } if (thread != null) { stopReceiveThread(thread);