Fix SnackbarDispatcher dropping messages for hosts that subscribe after a post

Publish the current queue head via a StateFlow so any host that subscribes
after a message is posted (e.g. a screen recomposing as a flow pops back to it)
still observes it. The previous mutex-gated flow delivered each message to a
single parked collector, which could starve the host actually on screen and
drop the snackbar. Adds a regression test for the late-subscriber case.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jenna Vassar
2026-06-09 10:03:49 -07:00
parent 8479c8346e
commit b8daada2a3
2 changed files with 38 additions and 21 deletions
@@ -18,39 +18,35 @@ import androidx.compose.runtime.remember
import androidx.compose.ui.res.stringResource
import io.element.android.libraries.designsystem.theme.components.Snackbar
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.asStateFlow
/**
* A global dispatcher of [SnackbarMessage] to be displayed in [Snackbar] via a [SnackbarHostState].
*
* The current head of the queue is exposed as a [MutableStateFlow] so that every collector — including
* hosts that subscribe *after* a message was posted (e.g. a screen recomposing as a flow pops back to
* it) — observes the current message. An earlier mutex-gated implementation delivered each message to
* whichever single collector happened to be parked on the lock, which could starve the host that was
* actually on screen, dropping the snackbar.
*/
class SnackbarDispatcher {
private val queueMutex = Mutex()
private val snackBarMessageQueue = ArrayDeque<SnackbarMessage>()
val snackbarMessage: Flow<SnackbarMessage?> = flow {
while (currentCoroutineContext().isActive) {
queueMutex.lock()
emit(snackBarMessageQueue.firstOrNull())
}
}
private val currentMessage = MutableStateFlow<SnackbarMessage?>(null)
val snackbarMessage: Flow<SnackbarMessage?> = currentMessage.asStateFlow()
@Synchronized
fun post(message: SnackbarMessage) {
if (snackBarMessageQueue.isEmpty()) {
snackBarMessageQueue.add(message)
if (queueMutex.isLocked) queueMutex.unlock()
} else {
snackBarMessageQueue.add(message)
}
snackBarMessageQueue.add(message)
currentMessage.value = snackBarMessageQueue.firstOrNull()
}
@Synchronized
fun clear() {
if (snackBarMessageQueue.isNotEmpty()) {
snackBarMessageQueue.removeFirstOrNull()
if (queueMutex.isLocked) queueMutex.unlock()
}
snackBarMessageQueue.removeFirstOrNull()
currentMessage.value = snackBarMessageQueue.firstOrNull()
}
}
@@ -54,6 +54,27 @@ class SnackbarDispatcherTest {
}
}
@Test
fun `a host that subscribes after a message is posted still receives it`() = runTest {
val snackbarDispatcher = SnackbarDispatcher()
val message = SnackbarMessage(0)
// A first host is already collecting (mirrors a background screen still subscribed).
snackbarDispatcher.snackbarMessage.test {
assertThat(awaitItem()).isNull()
snackbarDispatcher.post(message)
assertThat(awaitItem()).isEqualTo(message)
// A second host subscribes only afterwards — e.g. the screen a flow pops back to once the
// message has already been posted. It must observe the current message rather than being
// starved waiting for the next post (regression: an earlier mutex-gated implementation
// delivered each post to a single parked collector, dropping the snackbar for any host
// that subscribed later).
snackbarDispatcher.snackbarMessage.test {
assertThat(awaitItem()).isEqualTo(message)
}
}
}
@Test
fun `given 2 message emissions, the next message is displayed only after a call to clear`() = runTest {
val snackbarDispatcher = SnackbarDispatcher()