Kotlin Flow

Flow is an asynchronous stream of values built on suspend functions. A cold flow does nothing until collect is called and runs its producer again for every collector; a hot stream (StateFlow, SharedFlow, Channel) emits whether or not anyone is listening. Cold: flow { }, a Room @Query that returns Flow. Two collectors mean two database queries or two network calls. Hot: StateFlow holds a current value, SharedFlow broadcasts to everyone subscribed right now. A late subscriber misses earlier values unless the flow replays them. **Flow vs Channel**: a Channel is a hot queue between coroutines: each element is received by exactly one consumer. Flow is the declarative stream type with operators; you rarely expose a raw Channel from a ViewModel. Cold to hot: stateIn and shareIn collect the cold flow once and share it with many collectors.

Both produce values lazily, one at a time, but a Sequence is synchronous: the consumer pulls each value with next() and its thread runs the producer until the next yield. A Flow is built on suspend functions, so its producer can wait on a timer or the network without blocking a thread. **Why delay does not compile there**: the sequence { } block runs on a SequenceScope marked @RestrictsSuspension, so it may call only yield and yieldAll. Anything else fails with "Restricted suspending functions can invoke member or extension suspending functions only on their restricted coroutine scope". The reason for the rule: next() is a plain function, so there is no caller coroutine to suspend; the only way to wait inside would be to block the caller's thread. What only Flow has: cancellation together with the collecting coroutine, a dispatcher switch with flowOn, and time-based operators such as debounce. Which to pick: a sequence for in-memory CPU work (a lazy pipeline over a big list, a generator); a flow for anything that waits. asFlow() turns a sequence into a flow.

Terminal operators start the collection: collect, first, toList, fold and the like are suspend functions, and launchIn collects in a new coroutine. map, filter and onEach only wrap the flow in a new one, so a chain that ends in onEach { } runs nothing. Cold until collected: an intermediate operator returns at once and the producer's code does not run. Every terminal call runs the whole chain again from the source. How to tell them apart: intermediate operators are plain functions that return a Flow, callable anywhere; terminal ones are suspend (or, like launchIn, take a scope), so they need a coroutine. Early exit: first() takes one value and cancels the upstream. On a flow that completes empty it throws NoSuchElementException, while firstOrNull() returns null. Waiting for a state: session.first { it.isReady } suspends until a StateFlow holds a matching value, and returns at once if it already does; wrap it in withTimeoutOrNull(5_000) to stop waiting. **launchIn(scope)**: shorthand for scope.launch { collect() }. It returns the Job, which you cancel to stop collecting. onEach and catch go before it, since nothing can follow it.

Use StateFlow for screen state: it always holds one current value, needs an initial value and skips equal values. Use SharedFlow, a configurable broadcast with no current value, for events every subscriber should see. **StateFlow**: .value is readable synchronously; a slow collector sees only the latest value; equal values are dropped, so update with copies, never by mutating a list in place. **SharedFlow**: MutableSharedFlow(replay, extraBufferCapacity, onBufferOverflow). With replay = 0, a value emitted while nobody collects is lost. The relation: a StateFlow behaves like a SharedFlow with replay = 1, DROP_OLDEST, an initial value and distinctUntilChanged. Both are hot: collecting them never completes on its own. Expose read-only types: a private MutableStateFlow, a public asStateFlow(). One-shot events (navigation, snackbar): a SharedFlow loses them when the UI is not collecting; many teams model them as state or use Channel with receiveAsFlow() for a single collector.

StateFlow drops a new value that equals the current one, and _cart.value = items assigns the very list it already holds. Nothing is emitted, so the screen keeps showing the old cart. Why: MutableStateFlow compares the new value with the current one by equals before emitting. Here both are the same object, so they are always equal, no matter how much the list changed inside. The fix: keep an immutable List in the flow and replace it: _cart.update { it + item } builds a new list, and update is a compare-and-set, so two concurrent add calls do not lose an item. Same root, second symptom: the collector holds the same MutableList the ViewModel mutates, so a DiffUtil or Compose list reading it during add can see a half-changed list or throw ConcurrentModificationException. Catch it next time: never put a MutableList or a data class with var fields into a StateFlow; update state with copy. A ViewModel test with Turbine (cart.test { vm.add(x); awaitItem() }) fails on the missing emission.

collect on a StateFlow never returns, because a hot flow never completes, so the coroutine stays on the first line and the second collect is never reached. Start each collection in its own launch inside the repeatOnLifecycle block. Why it looks fine: collect is an ordinary suspend call, and the code after it runs only when the flow completes. StateFlow.collect, inherited from SharedFlow, is even declared to return Nothing, so the compiler flags the second line with an "Unreachable code" warning: the hint to look for. Why it sometimes works: a cold flow that completes, such as a one-shot flow { } or a flowOf in a test fake, lets the next line run, so the bug can hide until the real repository flow is wired in. The fix: two launch children run concurrently; repeatOnLifecycle cancels both at onStop and starts both again at onStart. Better still: if the screen always needs both values, combine them into one UiState in the ViewModel and collect that one flow. In Compose every collectAsStateWithLifecycle() call has its own collector, so the trap does not appear there. Same root elsewhere: toList() or count() on a StateFlow hangs forever, and code after a collect in a LaunchedEffect never runs.

A Channel is a hot queue between coroutines: send puts an element in, receive takes it out, and each element reaches exactly one receiver. A Flow is a cold description that runs again for every collector; a channel exists on its own, even when nobody receives. **RENDEZVOUS** (the default for Channel(), capacity 0): send suspends until a receiver takes the element, a direct hand-off. trySend fails when no receiver is waiting. **BUFFERED**: a buffer of 64 by default; send suspends only when it is full. **CONFLATED**: keeps one element; a new send replaces the unread one and never suspends. For "only the latest matters". **UNLIMITED**: send never suspends, so a slow receiver lets memory grow without bound. Fan-out, not broadcast: two coroutines receiving from one channel split the elements between them; two collectors of a SharedFlow both get every value. Closing: after close(), a for (x in channel) loop ends once it has drained the rest; receive() on a closed, drained channel throws ClosedReceiveChannelException. A channel nobody closes keeps its receivers suspended forever. Where it fits: a work queue with one or a few consumers, or one-time UI events through receiveAsFlow(); consumeAsFlow() can be collected only once. Flow itself uses channels inside buffer, flowOn, channelFlow and callbackFlow.

flowOn changes the context only for the operators above it, up to the previous flowOn. So db.readAll() and map (1) run on IO, filter (2) runs on Default, and collect (3) runs on Main, where the collecting coroutine was launched. Upstream only: the collector's context never changes. That is why Flow has no observeOn: the downstream thread is simply where you call collect. How it works: at a dispatcher boundary flowOn runs the upstream in a separate coroutine and passes values through a channel, so it also adds a buffer (64 by default). **No withContext around emit**: a flow { } builder must emit from the collector's coroutine. Emitting from inside withContext fails with IllegalStateException: Flow invariant is violated. Calling withContext for a computation and emitting outside it is fine. Several producers: channelFlow { } allows send from other coroutines and contexts.

emit is called from coroutines started with launch, not from the coroutine that collects the flow, and flow { } forbids that: "Emission from another coroutine is detected". Use channelFlow { }, whose send is safe to call from any number of child coroutines. Why the rule exists: in flow { }, emit runs the collector's code directly. Two coroutines emitting at once, possibly on two threads, would run the downstream concurrently, and FlowCollector is not thread-safe. The runtime check says exactly that and suggests channelFlow. **How channelFlow fixes it**: the producers send into a channel and the collector reads from it in its own coroutine. The flow completes when the block and all its children have finished, and a failure in one child cancels the other. Order: results arrive in the order the pages finish, so page 2 may come first. When order matters, start both with async and emit the results one after another from the flow's own coroutine; emit directly inside coroutineScope { } is allowed, inside launch or async it is not. Other options: merge(flow { emit(api.page(1)) }, flow { emit(api.page(2)) }) also runs both concurrently. callbackFlow is the same channel-based builder for listener APIs.

zip prints 1A 2B: it pairs values by position and stops as soon as one flow ends, so 3 is never paired. combine prints 1A 2A 2B 3B: once both flows have a value, it emits the latest pair every time either of them emits. **zip: each pair waits for one new value from each side, so the faster flow is slowed to the pace of the slower one, and the leftover values of the longer flow are dropped. It fits values that belong together by position, which is rare in UI code. combine**: silent until every source has emitted once (nothing at 100 ms here), then it recomputes on every emission from any source. It fits screen state built from several sources, such as the products and the current filter. If a source may stay silent, give it a starting value with onStart, or the screen never leaves its initial state. Completion: zip completes when either flow completes and cancels the other; combine completes when all of them have. **merge**: for flows of the same type when you want every value as it arrives, with no pairing: local edits and push messages into one sync handler, several click sources into one event stream.

flatMapLatest: when a new query arrives it cancels the request for the old one, so stale results cannot overwrite fresh ones. Around it, debounce waits for a pause in typing and distinctUntilChanged drops a repeated query. **flatMapConcat: waits for each inner flow to finish, in order, so every keystroke's search runs one after another. flatMapMerge: runs inner flows concurrently and interleaves their results; a slow old response can arrive after the new one. flatMapLatest / mapLatest: cancel the previous inner flow or transform when a new value comes. collectLatest**: the same idea at the end of the chain: it cancels the collector body when a new value arrives, while plain collect processes every value in turn. The pipeline: debounce(300), then filter out queries that are too short, distinctUntilChanged(), flatMapLatest, and stateIn for the screen. Errors: put catch inside the inner flow. A catch after flatMapLatest turns the first failure into a value and ends the whole search stream.

debounce(300) emits a value only after 300 ms pass with no newer one, sample(300) emits the newest value once every 300 ms, and throttle-first emits the first value and ignores the rest for a window. Search box: debounce. Sensor: sample. Pay button: throttle-first. **debounce**: the last value before a pause wins, which is exactly "the user stopped typing". A source that never pauses never emits: a 60 Hz sensor through debounce(300) stays silent. **sample: a fixed-rate snapshot that drops the values between ticks. It fits redrawing from a fast source (a sensor, download progress) at a rate the UI can afford; the last value is lost if the flow completes before the next tick. Throttle-first**: the first tap acts immediately and the duplicates inside the window are dropped. debounce on a button would delay every payment by the window, and two taps 400 ms apart would still pay twice. kotlinx.coroutines has no throttleFirst; it is a few lines with flow { }. Not only a flow problem: also disable the button while the request runs and make the payment idempotent on the server, so a tap that slips through cannot charge twice. Details: debounce and sample are @FlowPreview, so the compiler warns until you add @OptIn(FlowPreview::class). In RxJava sample is also called throttleLast, and throttle-first is throttleFirst.

By default nothing is lost: emit suspends until the collector finishes the previous value, so the producer slows down to the collector's pace. buffer lets the two run concurrently, conflate keeps only the newest value, and collectLatest cancels the slow handler when a new value arrives. **buffer(capacity, onBufferOverflow)**: producer and collector run in separate coroutines with a channel between them, so the total time is the slower side instead of the sum. Overflow policy: SUSPEND, DROP_OLDEST or DROP_LATEST. **conflate(): skips intermediate values. Right for progress, sensor or location updates; wrong for events where each one matters. collectLatest { }: restarts the body for every new value; good when only the latest render or save matters. Built-in policies**: StateFlow is conflated by design; MutableSharedFlow takes extraBufferCapacity and an overflow policy. Side effect: flowOn adds a buffer at the dispatcher boundary.

catch only handles exceptions from upstream (the operators above it), so an exception in collect { } is below it and goes straight to the collecting coroutine. Order the chain as source, retryWhen, map, catch, and move risky collector work into onEach above catch. Exception transparency: a flow must not swallow downstream exceptions, so wrapping emit in try/catch inside a flow { } builder breaks the contract and can fail with IllegalStateException. **retry / retryWhen**: re-collect the upstream. retry(3) { it is IOException } is the short form; retryWhen { cause, attempt -> } with a delay inside gives exponential backoff. **catch { emit(fallback) }: turns a failure into a value, but the flow still completes afterwards. A long-lived stream such as search must catch inside each inner flow to stay alive. Order matters**: catch above retryWhen leaves nothing to retry, because the failure is already a value. **onCompletion { cause -> }**: works like finally: it sees the cause and lets it continue, it does not handle it.

onStart runs before the upstream starts and can emit a first value, onEmpty runs when the upstream completes without emitting anything, and onCompletion runs after the flow ends for any reason: success, an exception or cancellation. A loading flag set in onStart is cleared in onCompletion. **onStart**: runs on every collection, so each retry placed after it and each restart of a WhileSubscribed share runs it again, and the spinner shows again. **onCompletion { cause -> }**: cause is null on success, the exception on failure, a CancellationException on cancellation. It sees failures from downstream too, and it does not handle them: like finally, it lets the exception continue to catch or to the collector. Why not elsewhere: clearing the flag in onEach or after the last value misses the error path, and clearing it in catch misses cancellation, which is not an error: a sync the user stops with a Cancel button leaves the spinner on screen. Endless flows: a Room query or a socket never completes, so its onCompletion runs only on cancellation. There, clear the flag on the first value, or model loading as a state that the next value replaces. **onEmpty { emit(NothingFound) }**: fires on zero emissions, not on an empty list. A Room query that returns emptyList() did emit, so check for an empty list in map instead.

Return Flow<T> when the data can change while someone is looking at it and the caller should see each change: a Room query, a DataStore preference, a socket. Use a suspend fun for a one-shot operation with one result: a network request, a write, a command. Flow for observation: a Room query that returns Flow re-emits after every write to its tables, so a screen that observes the database updates itself after any refresh or edit, with no refetch logic. That is the single source of truth pattern. suspend for one result: suspend fun placeOrder(cart: Cart): OrderId reads as a call, fails with an exception the caller catches, and runs exactly once. Wrapped in flow { emit(api.order(cart)) } it runs again for every collector and hides that only one value will ever come. The usual pair: fun observeArticles(): Flow<List<Article>> plus suspend fun refresh(), which fetches from the network and writes to Room; the new rows reach the screen through the flow. **No suspend fun that returns a Flow**: creating a cold flow does no work, so suspend there only forces callers into a coroutine for nothing; put the suspending work inside the flow. Keep UI states out: Loading and error banners are the ViewModel's business (onStart, catch); a repository that emits Resource.Loading ties every caller to one screen's idea of loading.

stateIn and shareIn collect a cold upstream once and re-broadcast it to many collectors; SharingStarted decides when that collection starts and stops. With WhileSubscribed(5_000) the upstream survives a rotation and stops 5 seconds after the last collector leaves. The three strategies Eagerly: subscribes immediately and never stops; the polling runs before anyone looks at the screen. Lazily: subscribes on the first collector and never stops; the flow keeps running after the user leaves. WhileSubscribed(5_000): subscribes on the first collector and stops 5 seconds after the last one leaves. A rotation recreates the collector well inside that window, so the Room query or socket is not restarted; a real exit passes it and the upstream stops. replayExpirationMillis (default: never) sets how long the cached value survives that stop; 0 makes a return reload instead of showing stale data. Bugs blamed on the wrong thing WhileSubscribed works only if the UI really unsubscribes: collectAsStateWithLifecycle() or repeatOnLifecycle. A plain collectAsState() never lets the subscriber count reach zero. A property with a get() that returns flow.stateIn(...) builds a new StateFlow and a new upstream collection on every access. Initialise it once. stateIn vs shareIn stateIn needs an initial value, conflates, and drops equal values; shareIn takes replay and does not drop equal values. Rule of thumb: WhileSubscribed(5_000) for screen state, Eagerly only for data the app needs whether or not a screen is open.

Usually inside a flow exposed with stateIn(viewModelScope, WhileSubscribed(5_000), initial): the load starts when the screen first collects, survives rotation and stops when nobody watches. init { } starts it at construction, and LaunchedEffect repeats it on every rotation. init { } The load starts when the ViewModel is constructed, before anything collects, even if the screen is never drawn. A collector launched there runs until onCleared(), so a socket or location flow keeps working while the app sits in the background. In a test the constructor has already started or queued the load before the test can arrange anything, and a retry means a load() function that guards against double loads. Its strength: plain imperative code that is easy to mix with user actions calling _state.update { }. LaunchedEffect(Unit) { vm.load() } It runs every time the composable enters the composition. A rotation recreates the composition but keeps the ViewModel, and so does a return to the screen from the back stack, so the load runs again each time. The UI now decides when data loads, against unidirectional data flow, and the ViewModel needs an "already loaded" flag: both are signs of the wrong place. stateIn with the load inside the flow Lazy: constructing the ViewModel costs nothing, and a test triggers the load by collecting (Turbine's test { }). WhileSubscribed(5_000) keeps the result across a rotation and stops the work in the background. A return after more than 5 seconds shows the cached value and runs the load again; use SharingStarted.Lazily if it should not. The cost: user actions have to become inputs of the chain. A retry is a trigger flow followed by flatMapLatest, a filter is a MutableStateFlow mixed in with combine. What is easy to write imperatively takes more thought here.

A SharedFlow has one buffer for all subscribers, and a slot is freed only once the slowest subscriber has read it. MutableSharedFlow() has no buffer, so emit suspends until every subscriber takes the value: the emitter moves at the slow subscriber's pace, and the fast ones wait with it. Without a buffer Each subscriber keeps its own position in the shared buffer. With one subscriber that needs 300 ms per value and one fast subscriber, emit returns only after the slow one takes the value, and the fast one gets at most one value ahead (the example shows it). tryEmit never suspends, so it returns false whenever emit would have to suspend. On a MutableSharedFlow<T>() with any subscriber that is every call: the classic "events never arrive" bug, and the reason event buses declare extraBufferCapacity. With no subscribers it returns true and the value is simply dropped. What the parameters change extraBufferCapacity = 64: the emitter and the fast subscribers may run up to 64 values ahead of the slowest one; past that emit suspends again. The buffer holds replay + extraBufferCapacity values. onBufferOverflow decides what a full buffer does: SUSPEND (the default) suspends emit, DROP_OLDEST drops the oldest value, DROP_LATEST drops the new one. With either drop policy emit never suspends, and a dropped value is lost for every subscriber, not only the slow one. A drop policy needs a buffer: MutableSharedFlow<Int>(onBufferOverflow = DROP_OLDEST) with no replay or extraBufferCapacity throws IllegalArgumentException. Keeping the slowness local Let the slow subscriber buffer for itself: bus.buffer(Channel.UNLIMITED).collect { slowWrite(it) } collects the bus in a fast coroutine and queues the work behind it, so the emitter and the others are not held back. StateFlow never stalls like this: it is conflated, and each subscriber reads the latest value. Use it when only the latest value matters.

Flow is an interface with one function, suspend fun collect(collector: FlowCollector<T>), and FlowCollector has one too, suspend fun emit(value: T). Collecting runs the producer's code, and each emit calls the collector's code directly, with no queue or thread in between. How a chain runs flow { } returns a flow whose collect runs your block with the downstream collector as its receiver. An operator such as map returns a new flow whose collect collects the flow above it and emits transformed values downstream. So source.map { }.filter { }.collect { } is a set of nested calls, and every value travels through a chain of emit calls on one call stack into the final lambda. That is why the default backpressure costs nothing: the producer stays suspended inside emit while the downstream processes the value. Only operators that need concurrency put a channel in between: buffer, flowOn, channelFlow, combine, flatMapMerge. What flow { } adds The collector is wrapped in a SafeCollector that checks every emit: it must come from the collecting coroutine, in the same context (context preservation), or it fails with "Flow invariant is violated". It also checks cancellation on every emit. flowOf(...) and asFlow() skip that check for speed, so a loop over them runs to the end after a cancel; .cancellable() adds it. Implementing Flow with an object compiles but skips these checks; the documentation calls the interface unstable for inheritance and points to flow { } or AbstractFlow. Writing an operator An extension on Flow<T> that returns flow { collect { value -> ... emit(...) } }. For stateless operators that emit zero or more values per input, transform { } is the shortcut. Keep per-collection state inside the flow { } block. A var declared in the function body outside it is shared by every collection of the returned flow: the second toList() in the example starts with the first one's last value. Never catch around emit (it would swallow downstream exceptions) and never emit from another coroutine or context: that is what channelFlow and flowOn are for.