Hot flows: SharedFlow, StateFlow, and the WhileSubscribed timeout
StateFlow is a current value, SharedFlow is an event bus, and shareIn's stop timeout is the difference between one websocket and one per screen rotation.
Explore the full explanation 8 ideas
A cold `Flow` runs once per collector: two collectors mean two independent executions, two database queries, two websocket connections. That is the right default for a transformation pipeline and the wrong one for a shared source. Hot flows exist for the second case — they run **once**, independently of who is listening, and broadcast to every current subscriber. `SharedFlow` and `StateFlow` are the two built-in hot types, and `shareIn`/`stateIn` are the operators that turn any cold flow into one.
`MutableSharedFlow<T>(replay, extraBufferCapacity, onBufferOverflow)` is the event bus. Every `emit` is delivered to every active collector; the last `replay` emissions are kept and handed to collectors that arrive later. `replay = 0` is a pure event feed — a subscriber that was not there missed it, permanently. The buffering parameters matter more than they look: `emit` **suspends** when the buffer is full and the overflow policy is `SUSPEND`, which means a slow collector applies back-pressure to your producer, and one stuck subscriber can stall the whole bus. `tryEmit` plus a `DROP_OLDEST` policy is the shape you want when the producer must never be blocked by a consumer.
`StateFlow<T>` is a different thing wearing a similar interface: it is a **current value**, always present, and it has two behaviours a `SharedFlow` does not. It is **conflated** — a fast writer's intermediate values never reach a collector that has not resumed yet, so writing `1, 2, 3` with no suspension in between leaves only `3` observable. And it **deduplicates by equality** — setting the same value twice emits once, because `StateFlow` compares with `equals`. It is approximately `SharedFlow(replay = 1, onBufferOverflow = DROP_OLDEST)` plus `distinctUntilChanged` plus a mandatory initial value.
Those two behaviours are the source of a classic bug. Model a one-shot *event* — "show a toast", "navigate to checkout" — as a `StateFlow` and two identical events in a row fire once, because the second is deduplicated; and an event emitted while nobody was collecting is either replayed later at the wrong moment or lost. The rule that follows: **state is a `StateFlow`, events are a `SharedFlow`**. If you can meaningfully ask "what is the current value?", it is state. If the answer is "there isn't one, it just happened", it is an event.
`shareIn(scope, started, replay)` converts a cold flow into a hot one, and its `started` parameter is where the real decisions live. `SharingStarted.Eagerly` starts the upstream immediately and never stops it — your websocket stays open when nobody is looking. `Lazily` starts on the first collector and never stops. `WhileSubscribed(stopTimeoutMillis)` starts on the first collector and stops the upstream `stopTimeoutMillis` after the last one leaves — which is the only policy that is both cheap when idle and stable across brief gaps.
That timeout is worth understanding concretely. With `WhileSubscribed(5000)`, a screen rotation or a navigation that briefly drops the last collector does **not** tear down the upstream: the subscription survives the gap and the same connection is reused. With `WhileSubscribed(0)` the upstream is cancelled and restarted, so a cold source runs its whole setup again — reconnect, re-authenticate, re-query, and any counter it held resets. Five seconds is the value the Android community converged on for exactly this reason, and the same reasoning applies to any expensive, stateful upstream: a change stream, a paid API poll, a long-lived gRPC stream.
`stateIn` is `shareIn` with `replay = 1`, conflation, and an initial value — it produces a `StateFlow`, so late subscribers immediately see the current value rather than waiting for the next emission. `stateIn(scope, WhileSubscribed(5000), initial)` is the canonical ViewModel line, and it is worth being able to say why all three arguments are there: the scope owns the upstream's lifetime, the policy decides when it runs, and the initial value is what a subscriber sees before the first real emission arrives.
Two operational notes. First, the scope you pass owns the sharing coroutine, so it must outlive the subscribers — an application scope, a `viewModelScope`, `runTest`'s `backgroundScope` — never the scope of whoever happens to be collecting right now. Second, hot flows never complete, so a collector on one never returns: `toList()` on a `SharedFlow` hangs forever, and a test that collects one must `launch` it and cancel it. That is not a rough edge; it is the definition of hot, and remembering it saves an afternoon.
Let’s figure it out.
Ask me about this course or where to start.
import kotlinx.coroutines.*import kotlinx.coroutines.flow.*import kotlinx.coroutines.test.*@OptIn(ExperimentalCoroutinesApi::class)fun main() = runTest {val values = listOf(1, 2, 2, 3)val state = MutableStateFlow(0)val stateSeen = mutableListOf<Int>()val j1 = launch { state.collect { stateSeen.add(it) } }runCurrent() // let the collector subscribefor (v in values) state.value = v // no suspension between writesrunCurrent(); j1.cancel()val shared = MutableSharedFlow<Int>(replay = 2)val sharedSeen = mutableListOf<Int>()val j2 = launch { shared.collect { sharedSeen.add(it) } }runCurrent()for (v in values) shared.emit(v)runCurrent(); j2.cancel()println("stateFlow = " + stateSeen) // conflated + dedupedprintln("sharedFlow = " + sharedSeen) // every emission, duplicate tooprintln("replayCache = " + shared.replayCache)}
The two behaviours that separate StateFlow from SharedFlow: conflation (a fast writer's intermediates never arrive) and equality-based dedup.
Arena IDEimport kotlinx.coroutines.*import kotlinx.coroutines.flow.*import kotlinx.coroutines.test.*@OptIn(ExperimentalCoroutinesApi::class)suspend fun TestScope.probe(stopMs: Long): String {var starts = 0val upstream = flow {starts++ // one per upstream STARTvar i = 0while (true) { emit(i++); delay(100) }}val shared = upstream.shareIn(backgroundScope, // must outlive the collectorsSharingStarted.WhileSubscribed(stopTimeoutMillis = stopMs),replay = 0,)val seen = mutableListOf<Int>()var job = launch { shared.collect { seen.add(it) } }advanceTimeBy(250); job.cancel()advanceTimeBy(300) // the gapjob = launch { shared.collect { seen.add(it) } }advanceTimeBy(250); job.cancel()advanceTimeBy(1000)return "seen=" + seen + " upstreamStarts=" + starts}@OptIn(ExperimentalCoroutinesApi::class)fun main() = runTest {println("WhileSubscribed(500) -> " + probe(500))println("WhileSubscribed(0) -> " + probe(0))}
The stop timeout, measured. Same collectors, same gap — the only difference is whether the expensive upstream survives it.
Arena IDEclass OrdersViewModel(repo: OrderRepository) : ViewModel() {// STATE: there is always a current value, and duplicates are noise.val orders: StateFlow<List<Order>> =repo.observeOrders() // a cold Flow from the driver.stateIn(scope = viewModelScope, // owns the upstream's lifetimestarted = SharingStarted.WhileSubscribed(5_000), // survive rotationinitialValue = emptyList(), // what a new subscriber sees first)// EVENTS: no "current value", and two identical ones must both fire.private val _effects = MutableSharedFlow<UiEffect>(extraBufferCapacity = 8)val effects: SharedFlow<UiEffect> = _effects.asSharedFlow()fun refresh() = viewModelScope.launch {runCatching { repo.refresh() }.onFailure { _effects.emit(UiEffect.ShowToast("Refresh failed")) }}}
The canonical ViewModel shape: state as a StateFlow with stateIn, one-shot events as a SharedFlow. Reference only — the point is the two different types.
import kotlinx.coroutines.*import kotlinx.coroutines.flow.*fun main() = runBlocking {val bus = MutableSharedFlow<String>()val seen = mutableListOf<String>()// WRONG: bus.toList() would suspend forever — a SharedFlow has no end.val job = launch { bus.collect { seen.add(it) } }yield() // let the collector subscribebus.emit("a")bus.emit("b")yield()job.cancel() // the collector must be stoppedprintln("seen = " + seen)println("a SharedFlow is never 'done' — you cancel the collector instead")}
Hot flows never complete — so a collector never returns. Collect them in a launch you cancel, and never call toList() on one.
Arena IDE🧠 Check your understanding
0/1 · 0/1 answered1. Your ViewModel exposes navigation events as a `StateFlow<NavEvent?>`. A user taps the same button twice in a row and only navigates once. Why?