diff --git a/stream-android-core/src/main/java/io/getstream/android/core/api/utils/Threading.kt b/stream-android-core/src/main/java/io/getstream/android/core/api/utils/Threading.kt index 2f9ae5f7..15e51324 100644 --- a/stream-android-core/src/main/java/io/getstream/android/core/api/utils/Threading.kt +++ b/stream-android-core/src/main/java/io/getstream/android/core/api/utils/Threading.kt @@ -163,7 +163,8 @@ public inline fun runOn(looper: Looper, crossinline block: () -> T): Result< } else { val latch = CountDownLatch(1) var result: Result? = null - Handler(looper).post { + val handler = Handler(looper) + val runnable = Runnable { try { result = Result.success(block()) } catch (t: Throwable) { @@ -172,8 +173,13 @@ public inline fun runOn(looper: Looper, crossinline block: () -> T): Result< latch.countDown() } } + handler.post(runnable) if (!latch.await(5, TimeUnit.SECONDS)) { + // Drop the message instead of leaving it queued. It would otherwise run once + // the looper frees up, long after this call has already failed and the caller + // has moved on, applying [block] against state it no longer expects. + handler.removeCallbacks(runnable) throw IllegalStateException("Timed out waiting to post to main thread") } result!!.getOrThrow() diff --git a/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt b/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt index 34c0d760..0954c785 100644 --- a/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt +++ b/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt @@ -16,6 +16,8 @@ package io.getstream.android.core.internal.observers.lifecycle +import android.os.Handler +import android.os.Looper import androidx.lifecycle.DefaultLifecycleObserver import androidx.lifecycle.Lifecycle import androidx.lifecycle.LifecycleOwner @@ -25,8 +27,8 @@ import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleListener import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleMonitor import io.getstream.android.core.api.subscribe.StreamSubscription import io.getstream.android.core.api.subscribe.StreamSubscriptionManager -import io.getstream.android.core.api.utils.runOnMainLooper import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger import kotlinx.coroutines.ExperimentalCoroutinesApi internal class StreamLifecycleMonitorImpl( @@ -37,23 +39,77 @@ internal class StreamLifecycleMonitorImpl( private val started = AtomicBoolean(false) + /** Attach/detach messages posted to the main looper but not yet executed. */ + private val pendingOnMain = AtomicInteger(0) + + /** Guards the [started] flip together with the decision to run inline or to post. */ + private val transitionLock = Any() + override fun subscribe( listener: StreamLifecycleListener, options: StreamSubscriptionManager.Options, ): Result = subscriptionManager.subscribe(listener, options) override fun start(): Result = runCatching { - if (!started.compareAndSet(false, true)) { - return@runCatching - } - runOnMainLooper { lifecycle.addObserver(this) }.getOrThrow() + transition(from = false, to = true) { lifecycle.addObserver(this) } } override fun stop(): Result = runCatching { - if (!started.compareAndSet(true, false)) { - return@runCatching + transition(from = true, to = false) { lifecycle.removeObserver(this) } + } + + /** + * Flips [started] from [from] to [to] and runs [block] on the main looper without waiting for + * it to complete. + * + * Attaching and detaching the observer has to happen on the main thread, but *waiting* for it + * must not: a caller that already occupies the main looper — a suspend `disconnect()` bridged + * with `runBlocking`, for example — would block the very thread it is waiting on, and the wait + * could only ever end in a timeout. + * + * Running inline when the caller is already on the main looper is only safe while nothing of + * ours is queued. Otherwise this call would jump ahead of an attach/detach posted earlier from + * another thread and invert the two, leaving [started] and the observer disagreeing. + * + * The flip and that decision are taken under [transitionLock] as one step. Apart they leave a + * window: a background `start()` that has won the flip but not yet posted looks like nothing is + * queued, so a `stop()` arriving on the main thread in between detaches inline and the attach + * lands after it — observer attached, [started] false. [block] itself runs outside the lock; + * the inline path holds the main thread, so nothing of ours can overtake it there anyway, and + * the lock stays clear of the listener callbacks the attach fans out to. + */ + private fun transition(from: Boolean, to: Boolean, block: () -> Unit) { + val mainLooper = Looper.getMainLooper() ?: error("Main looper is not initialized") + val runInline = + synchronized(transitionLock) { + if (!started.compareAndSet(from, to)) { + return + } + val inline = Looper.myLooper() === mainLooper && pendingOnMain.get() == 0 + if (!inline) { + postToMainLooper(mainLooper, block) + } + inline + } + if (runInline) { + block() + } + } + + private fun postToMainLooper(mainLooper: Looper, block: () -> Unit) { + pendingOnMain.incrementAndGet() + val posted = + Handler(mainLooper).post { + try { + block() + } finally { + pendingOnMain.decrementAndGet() + } + } + if (!posted) { + pendingOnMain.decrementAndGet() + error("Main looper is no longer accepting messages") } - runOnMainLooper { lifecycle.removeObserver(this) }.getOrThrow() } override fun onResume(owner: LifecycleOwner) { diff --git a/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt b/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt index db3e6371..ca271fae 100644 --- a/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt +++ b/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt @@ -27,12 +27,14 @@ import io.getstream.android.core.api.subscribe.StreamSubscriptionManager import io.getstream.android.core.api.subscribe.StreamSubscriptionManager.Options import io.getstream.android.core.api.subscribe.StreamSubscriptionManager.Options.Retention import io.getstream.android.core.testing.TestLogger +import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicReference import kotlin.concurrent.thread import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertNull import kotlin.test.assertTrue import org.junit.runner.RunWith import org.robolectric.RobolectricTestRunner @@ -145,10 +147,9 @@ class StreamLifecycleMonitorTest { completed.countDown() } - // Continuously process main looper tasks while waiting for thread to complete - while (!completed.await(10, TimeUnit.MILLISECONDS)) { - shadowLooper.idle() - } + // start() queues the registration and returns; it must not wait on the main looper. + assertTrue(completed.await(5, TimeUnit.SECONDS)) + shadowLooper.idle() assertEquals(mainLooper.thread, owner.addObserverThread.get()) @@ -173,10 +174,9 @@ class StreamLifecycleMonitorTest { completed.countDown() } - // Continuously process main looper tasks while waiting for thread to complete - while (!completed.await(10, TimeUnit.MILLISECONDS)) { - shadowLooper.idle() - } + // stop() queues the removal and returns; it must not wait on the main looper. + assertTrue(completed.await(5, TimeUnit.SECONDS)) + shadowLooper.idle() assertEquals(mainLooper.thread, owner.removeObserverThread.get()) } @@ -196,10 +196,8 @@ class StreamLifecycleMonitorTest { startCompleted.countDown() } - // Continuously process main looper tasks while waiting for thread to complete - while (!startCompleted.await(10, TimeUnit.MILLISECONDS)) { - shadowLooper.idle() - } + assertTrue(startCompleted.await(5, TimeUnit.SECONDS)) + shadowLooper.idle() assertEquals( mainLooper.thread, @@ -213,10 +211,8 @@ class StreamLifecycleMonitorTest { stopCompleted.countDown() } - // Continuously process main looper tasks while waiting for thread to complete - while (!stopCompleted.await(10, TimeUnit.MILLISECONDS)) { - shadowLooper.idle() - } + assertTrue(stopCompleted.await(5, TimeUnit.SECONDS)) + shadowLooper.idle() assertEquals( mainLooper.thread, @@ -226,6 +222,99 @@ class StreamLifecycleMonitorTest { } } + // Regression for AND-1468: start() used to block the caller on a five second latch while + // waiting for the main looper to run the registration. A caller that is itself holding the + // main looper can never release it, so the wait could only ever end in the timeout. Here + // the looper is simply never idled, which has the same shape. + @Test + fun `start returns without waiting for the main looper to run`() { + val owner = RecordingLifecycleOwner() + val monitor = StreamLifecycleMonitor(TestLogger, owner.lifecycle, newSubscriptionManager()) + val completed = CountDownLatch(1) + val mainLooper = Looper.getMainLooper() + + thread(start = true, name = "StreamLifecycleMonitorTest-unblocked") { + monitor.start().getOrThrow() + completed.countDown() + } + + assertTrue(completed.await(2, TimeUnit.SECONDS), "start must not wait on the main looper") + assertNull(owner.addObserverThread.get(), "the registration is queued, not yet run") + + Shadows.shadowOf(mainLooper).idle() + + assertEquals(mainLooper.thread, owner.addObserverThread.get()) + } + + // A registration that is already queued must not be overtaken by a later call that happens + // to run on the main thread, or the observer ends up in the opposite state to `started`. + @Test + fun `a main thread call does not overtake a registration queued from another thread`() { + val owner = RecordingLifecycleOwner() + val monitor = StreamLifecycleMonitor(TestLogger, owner.lifecycle, newSubscriptionManager()) + val started = CountDownLatch(1) + val shadowLooper = Shadows.shadowOf(Looper.getMainLooper()) + + thread(start = true, name = "StreamLifecycleMonitorTest-attach") { + monitor.start().getOrThrow() + started.countDown() + } + assertTrue(started.await(5, TimeUnit.SECONDS)) + + // The attach is still queued, so this stop() has to queue behind it rather than run + // inline on the main thread. + monitor.stop().getOrThrow() + shadowLooper.idle() + + assertEquals(listOf("add", "remove"), owner.calls.toList()) + } + + // The attach lands after start() returns, so a listener subscribed in between — which is what + // StreamNetworkAndLifecycleMonitorImpl.start() does — is already in place for the ON_RESUME + // that LifecycleRegistry replays to a newly added observer. Nothing acts on that replay: the + // recovery evaluator needs an earlier successful connection before it will reconnect, covered + // by StreamConnectionRecoveryEvaluatorImplTest. The delivery itself is behaviour now, so pin it + // here rather than leave it to be rediscovered. + @Test + fun `a listener subscribed before start receives the replayed foreground`() { + val owner = TestLifecycleOwner() + owner.registry.handleLifecycleEvent(Lifecycle.Event.ON_CREATE) + owner.registry.handleLifecycleEvent(Lifecycle.Event.ON_START) + owner.registry.handleLifecycleEvent(Lifecycle.Event.ON_RESUME) + + val monitor = StreamLifecycleMonitor(TestLogger, owner.lifecycle, newSubscriptionManager()) + val received = CopyOnWriteArrayList() + val listener = + object : StreamLifecycleListener { + override fun onForeground() { + received += "fg" + } + + override fun onBackground() { + received += "bg" + } + } + val subscription = + monitor + .subscribe(listener, Options(retention = Retention.KEEP_UNTIL_CANCELLED)) + .getOrThrow() + + val completed = CountDownLatch(1) + thread(start = true, name = "StreamLifecycleMonitorTest-replay") { + monitor.start().getOrThrow() + completed.countDown() + } + assertTrue(completed.await(5, TimeUnit.SECONDS)) + + assertEquals(emptyList(), received.toList(), "the attach is still queued") + + Shadows.shadowOf(Looper.getMainLooper()).idle() + + assertEquals(listOf("fg"), received.toList()) + + subscription.cancel() + } + private fun newSubscriptionManager(): StreamSubscriptionManager = StreamSubscriptionManager(TestLogger) @@ -240,16 +329,19 @@ class StreamLifecycleMonitorTest { private val registry = LifecycleRegistry(this) val addObserverThread = AtomicReference() val removeObserverThread = AtomicReference() + val calls = CopyOnWriteArrayList() override val lifecycle: Lifecycle = object : Lifecycle() { override fun addObserver(observer: LifecycleObserver) { addObserverThread.set(Thread.currentThread()) + calls.add("add") registry.addObserver(observer) } override fun removeObserver(observer: LifecycleObserver) { removeObserverThread.set(Thread.currentThread()) + calls.add("remove") registry.removeObserver(observer) } diff --git a/stream-android-core/src/test/java/io/getstream/android/core/api/utils/ThreadingTest.kt b/stream-android-core/src/test/java/io/getstream/android/core/api/utils/ThreadingTest.kt index 5471c22e..26925f16 100644 --- a/stream-android-core/src/test/java/io/getstream/android/core/api/utils/ThreadingTest.kt +++ b/stream-android-core/src/test/java/io/getstream/android/core/api/utils/ThreadingTest.kt @@ -17,10 +17,12 @@ package io.getstream.android.core.api.utils import android.os.Build +import android.os.Handler import android.os.HandlerThread import android.os.Looper import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicReference import kotlin.concurrent.thread @@ -28,6 +30,7 @@ import kotlin.coroutines.cancellation.CancellationException import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertFailsWith +import kotlin.test.assertFalse import kotlin.test.assertNotNull import kotlin.test.assertNull import kotlin.test.assertSame @@ -450,6 +453,36 @@ class ThreadingTest { assertTrue(exception.message?.contains("Timed out") == true) } + @Test + fun `runOn drops the pending message when it times out`() { + val handlerThread = HandlerThread("ThreadingTest-busy").apply { start() } + try { + val looper = handlerThread.looper + val release = CountDownLatch(1) + val executed = AtomicBoolean(false) + + // Occupy the looper for longer than runOn's five second bound. + Handler(looper).post { release.await(30, TimeUnit.SECONDS) } + + val exception = + assertFailsWith { + runOn(looper) { executed.set(true) }.getOrThrow() + } + assertTrue(exception.message?.contains("Timed out") == true) + + // Free the looper and let anything still queued run. The drain message is posted + // after the timed-out one, so it cannot overtake it. + release.countDown() + val drained = CountDownLatch(1) + Handler(looper).post { drained.countDown() } + assertTrue(drained.await(5, TimeUnit.SECONDS)) + + assertFalse(executed.get(), "a timed-out block must not run afterwards") + } finally { + handlerThread.quitSafely() + } + } + @Test fun `runOn handles concurrent calls to same looper`() { val handlerThread = HandlerThread("ThreadingTest-concurrent").apply { start() }