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); diff --git a/lib/src/test/java/org/asamk/signal/manager/internal/ReceiveThreadTest.java b/lib/src/test/java/org/asamk/signal/manager/internal/ReceiveThreadTest.java new file mode 100644 index 00000000..63142cc1 --- /dev/null +++ b/lib/src/test/java/org/asamk/signal/manager/internal/ReceiveThreadTest.java @@ -0,0 +1,256 @@ +package org.asamk.signal.manager.internal; + +import org.asamk.signal.manager.Manager; +import org.asamk.signal.manager.Settings; +import org.asamk.signal.manager.api.MessageEnvelope; +import org.asamk.signal.manager.api.ServiceEnvironment; +import org.asamk.signal.manager.config.ServiceConfig; +import org.asamk.signal.manager.helper.AccountFileUpdater; +import org.asamk.signal.manager.helper.Context; +import org.asamk.signal.manager.helper.ReceiveHelper; +import org.asamk.signal.manager.storage.SignalAccount; +import org.asamk.signal.manager.util.KeyUtils; +import org.bouncycastle.jce.provider.BouncyCastleProvider; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; +import org.signal.core.models.ServiceId.ACI; +import org.whispersystems.signalservice.api.push.ServiceIdType; + +import java.nio.file.Path; +import java.security.Security; +import java.time.Duration; +import java.util.Optional; +import java.util.function.BooleanSupplier; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +@Timeout(60) +class ReceiveThreadTest { + + private static final Duration TIMEOUT = Duration.ofSeconds(10); + + @TempDir + Path directory; + + private ManagerImpl manager; + private FakeReceiveHelper receiveHelper; + + @BeforeEach + void setUp() throws Exception { + // ManagerImpl.close() creates the push service socket, whose trust store needs BouncyCastle + if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) { + Security.addProvider(new BouncyCastleProvider()); + } + + final var account = SignalAccount.createLinkedAccount(directory.toFile(), + "account", + ServiceEnvironment.STAGING, + Settings.DEFAULT); + account.setProvisioningData(null, + ACI.parseOrThrow("11111111-1111-4111-8111-111111111111"), + null, + "test-password", + new byte[]{1}, + KeyUtils.generateIdentityKeyPair(), + null, + KeyUtils.createProfileKey(), + null, + new byte[32], + null); + account.finishLinking(2, KeyUtils.generatePreKeysForType(account.getAccountData(ServiceIdType.ACI)), null); + manager = new ManagerImpl(account, + PathConfig.createDefault(directory.toFile()), + new AccountFileUpdater() { + @Override + public void updateAccountIdentifiers(final String number, final ACI aci) { + } + + @Override + public void removeAccount() { + } + }, + ServiceConfig.getServiceEnvironmentConfig(ServiceEnvironment.STAGING, "signal-cli-test"), + "signal-cli-test"); + + // Replace the receive helper, so the receive thread doesn't connect to the server + final var contextField = ManagerImpl.class.getDeclaredField("context"); + contextField.setAccessible(true); + final var context = (Context) contextField.get(manager); + receiveHelper = new FakeReceiveHelper(context); + final var receiveHelperField = Context.class.getDeclaredField("receiveHelper"); + receiveHelperField.setAccessible(true); + receiveHelperField.set(context, receiveHelper); + } + + @AfterEach + void tearDown() { + receiveHelper.setHoldStoppingLoops(false); + manager.close(); + } + + @Test + void handlerAddedWhileReceiveThreadIsStoppingDoesNotStartSecondReceiveThread() throws Exception { + final var first = newHandler(); + final var second = newHandler(); + + manager.addReceiveHandler(first); + assertTrue(receiveHelper.await(() -> receiveHelper.runningLoops == 1)); + + receiveHelper.setHoldStoppingLoops(true); + final var remover = Thread.ofPlatform().start(() -> manager.removeReceiveHandler(first)); + assertTrue(receiveHelper.await(() -> receiveHelper.stoppingLoops == 1)); + + // Like a client that reconnects while the receive thread for its previous connection is still stopping + manager.addReceiveHandler(second); + assertTrue(receiveHelper.await(() -> receiveHelper.startedThreads == 1)); + + receiveHelper.setHoldStoppingLoops(false); + assertTrue(remover.join(TIMEOUT)); + assertTrue(receiveHelper.await(() -> receiveHelper.startedLoops == 2 && receiveHelper.runningLoops == 1)); + assertTrue(receiveHelper.await(() -> receiveHelper.startedThreads == 2 && receiveHelper.maxRunningLoops == 1)); + + manager.removeReceiveHandler(second); + assertTrue(receiveHelper.await(() -> receiveHelper.runningLoops == 0)); + } + + @Test + void stopRequestedForStoppingReceiveThreadDoesNotStopNextReceiveThread() throws Exception { + final var first = newHandler(); + final var second = newHandler(); + final var third = newHandler(); + + manager.addReceiveHandler(first); + assertTrue(receiveHelper.await(() -> receiveHelper.runningLoops == 1)); + + receiveHelper.setHoldStoppingLoops(true); + final var firstRemover = Thread.ofPlatform().start(() -> manager.removeReceiveHandler(first)); + assertTrue(receiveHelper.await(() -> receiveHelper.stoppingLoops == 1)); + manager.addReceiveHandler(second); + final var secondRemover = Thread.ofPlatform().start(() -> manager.removeReceiveHandler(second)); + assertTrue(receiveHelper.await(() -> receiveHelper.stopRequests == 2)); + + receiveHelper.setHoldStoppingLoops(false); + assertTrue(firstRemover.join(TIMEOUT)); + assertTrue(secondRemover.join(TIMEOUT)); + assertTrue(receiveHelper.await(() -> receiveHelper.runningLoops == 0 && receiveHelper.startedThreads == 1)); + + manager.addReceiveHandler(third); + assertTrue(receiveHelper.await(() -> receiveHelper.startedLoops == 2 && receiveHelper.runningLoops == 1)); + assertTrue(receiveHelper.await(() -> receiveHelper.startedThreads == 2 && receiveHelper.maxRunningLoops == 1)); + } + + private static Manager.ReceiveMessageHandler newHandler() { + // An anonymous class instead of a lambda, so every handler is a separate instance + return new Manager.ReceiveMessageHandler() { + @Override + public void handleMessage(final MessageEnvelope envelope, final Throwable e) { + } + }; + } + + /** + * Instead of receiving messages from the server, a receive loop waits until it has been asked to stop. + * While holdStoppingLoops is set, a loop that has been asked to stop doesn't return yet, like a loop that + * is still busy when the last handler is removed. + */ + private static class FakeReceiveHelper extends ReceiveHelper { + + private final Object lock = new Object(); + private boolean stopRequested; + private boolean holdStoppingLoops; + private int stopRequests; + private int startedThreads; + private int startedLoops; + private int runningLoops; + private int maxRunningLoops; + private int stoppingLoops; + + FakeReceiveHelper(final Context context) { + super(context); + } + + @Override + public boolean requestStopReceiveMessages() { + super.requestStopReceiveMessages(); + synchronized (lock) { + stopRequested = true; + stopRequests++; + lock.notifyAll(); + } + // The fake loop isn't blocked in a read that needs to be interrupted + return false; + } + + @Override + public void clearStopRequest() { + super.clearStopRequest(); + synchronized (lock) { + stopRequested = false; + } + } + + @Override + public void receiveMessagesContinuously(final Manager.ReceiveMessageHandler handler) { + synchronized (lock) { + startedThreads++; + lock.notifyAll(); + } + super.receiveMessagesContinuously(handler); + } + + @Override + public void receiveMessages( + final Optional timeout, + final Integer maxMessages, + final Manager.ReceiveMessageHandler handler + ) { + synchronized (lock) { + startedLoops++; + runningLoops++; + maxRunningLoops = Math.max(maxRunningLoops, runningLoops); + lock.notifyAll(); + if (await(() -> stopRequested)) { + stoppingLoops++; + lock.notifyAll(); + await(() -> !holdStoppingLoops); + stoppingLoops--; + } + runningLoops--; + lock.notifyAll(); + } + } + + void setHoldStoppingLoops(final boolean hold) { + synchronized (lock) { + holdStoppingLoops = hold; + lock.notifyAll(); + } + } + + /** + * Waits until the condition, which is evaluated while holding the lock, is true. + * Returns false if it didn't become true within the timeout. + */ + boolean await(final BooleanSupplier condition) { + synchronized (lock) { + final var deadline = System.nanoTime() + TIMEOUT.toNanos(); + while (!condition.getAsBoolean()) { + final var remainingMillis = (deadline - System.nanoTime()) / 1_000_000; + if (remainingMillis <= 0) { + return false; + } + try { + lock.wait(remainingMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + return true; + } + } + } +}