Kotlin Flow Interview Questions

A Kotlin Flow is a cold, asynchronous stream that can emit multiple values over time — think of it like a pipe that carries a sequence of results, not just one. A suspend function returns one value and finishes. A Flow returns many values lazily, only when someone is collecting from it.

// suspend fun → one value, then done
suspend fun fetchUser(): User = api.getUser()

// Flow → multiple values over time
fun userUpdates(): Flow<User> = flow {
    while (true) {
        emit(api.getUser())   // emits a new value every 5s
        delay(5_000)
    }
}

// COLD — nothing runs until collected
val flow = userUpdates()  // no network call yet
flow.collect { user -> render(user) }  // NOW it starts

// LiveData vs Flow comparison:
// LiveData  → Android-only, main thread, no operators, observeForever leaks
// Flow      → pure Kotlin, any thread, full operator chain, KMP-compatible

Cold vs hot. A regular Flow is cold — each collector gets its own independent execution. If two screens collect the same Flow, two separate network requests are made. Hot flows (StateFlow, SharedFlow) share a single execution among all collectors. This distinction matters enormously for efficiency: a database query Flow is fine cold, but a WebSocket stream should be hot so you don't open two connections.

vs LiveData. LiveData requires LifecycleOwner, only runs on the main thread, has no native operators, and cannot be used outside Android. StateFlow is the modern drop-in replacement: it holds state, is thread-safe, works with all Flow operators, and is pure Kotlin — making it suitable for shared ViewModels in Kotlin Multiplatform projects. Google's architecture samples have migrated entirely from LiveData to StateFlow.

StateFlow is a hot flow with exactly one current value — it's a state holder. SharedFlow is a hot flow that can emit events without a persistent current value — it's an event bus. The mental model: StateFlow = the latest screen state. SharedFlow = one-time navigation or toast events.

// StateFlow — always has a value, replays 1 to new collectors
val uiState: StateFlow<UiState> = MutableStateFlow(UiState.Loading)
// New collector immediately receives UiState.Loading
// Skips IDENTICAL consecutive values (equality-based distinctUntilChanged)

// SharedFlow — no initial value, configurable replay
val events: SharedFlow<UiEvent> = MutableSharedFlow()
// New collector gets NOTHING until next emission
// Same event emitted to ALL active collectors simultaneously

// Practical ViewModel pattern:
class LoginViewModel : ViewModel() {
    // State — screen renders this continuously
    private val _uiState = MutableStateFlow<LoginState>(LoginState.Idle)
    val uiState = _uiState.asStateFlow()

    // Events — one-shot: navigate, show snackbar, etc.
    private val _events = MutableSharedFlow<LoginEvent>()
    val events = _events.asSharedFlow()

    fun login() {
        viewModelScope.launch {
            _uiState.value = LoginState.Loading
            val result = repo.login()
            if (result.isSuccess) {
                _events.emit(LoginEvent.NavigateToHome)  // one-shot
            } else {
                _uiState.value = LoginState.Error(result.errorMessage)
            }
        }
    }
}

StateFlow behaviour: replay = 1 (new collector always gets the latest value immediately), distinctUntilChanged by default (setting the same value twice emits only once), always readable via .value, equality check uses equals(). This makes it ideal for UI state — the screen can reconstruct itself from the latest emission alone.

SharedFlow behaviour: configurable replay (default 0 — no cached values for new collectors), no equality filtering (same value emitted twice → two emissions), no .value property. This makes it perfect for events: a navigation event should fire once; if the screen re-subscribes after rotation it should NOT re-navigate.

stateIn and shareIn are operators that convert a cold Flow into a hot StateFlow or SharedFlow, scoped to a CoroutineScope. Without them, every collector of a cold Flow would independently trigger the upstream — meaning two screens observing the same database Flow would run two separate queries. shareIn shares a single upstream execution among all collectors.

// Without stateIn — cold Flow, each collector runs independently
fun userData(): Flow<User> = dao.getUserFlow()
// Fragment A collects → DB query 1 started
// Fragment B collects → DB query 2 started (wasteful!)

// With stateIn — single upstream, multicast to all collectors
val userState: StateFlow<User?> = dao.getUserFlow()
    .stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5_000),
        initialValue = null
    )

// WhileSubscribed(5000) timeline:
// Collector arrives    → upstream Flow starts immediately
// Last collector leaves → 5-second countdown starts
// Within 5s: new collector arrives → upstream RESUMES (no restart)
// After 5s: no collector → upstream CANCELLED
// ↑ the 5s grace period covers configuration changes (rotation ~100-300ms)

// shareIn — converts cold Flow to SharedFlow
val locationUpdates = locationFlow()
    .shareIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), replay = 1)

// SharingStarted options:
// Eagerly          → starts immediately, never stops (use for app-wide streams)
// Lazily           → starts on first collector, never stops after that
// WhileSubscribed  → starts/stops with active subscribers + grace period

Why 5000ms specifically? Configuration changes (rotation, language switch, dark mode toggle) cause Android to destroy and recreate the Activity and Fragment. This process typically takes 100–300ms. Setting the grace period to 5000ms means the upstream flow (e.g., a network poll or database listener) stays active across the destruction/recreation window. The new subscriber reconnects and gets the same upstream — no wasted network request, no missed emission, and the StateFlow's last value is immediately available for the new UI to render while the upstream is still running.

stateIn vs shareIn: Use stateIn when you need a value that's always accessible via .value (UI state). Use shareIn when you want to multicast a stream without needing a current-value snapshot (real-time event streams, sensor data, WebSocket messages where replay semantics matter more than having a current value).

flowOn changes the upstream dispatcher — the context in which emissions are produced. The downstream collector continues on whatever context it was launched in. This is the correct way to move flow production to a background thread. withContext inside a flow builder is a common but subtly problematic pattern.

// ✅ CORRECT — flowOn shifts upstream production to IO
fun expensiveFlow(): Flow<Result> = flow {
    val data = db.query()   // runs on IO
    emit(data)
}.flowOn(Dispatchers.IO)  // applies to everything ABOVE it

// The collector still runs on Main:
viewModelScope.launch {       // Main dispatcher
    expensiveFlow().collect { result ->
        updateUi(result)       // executes on Main ✓
    }
}

// ❌ DON'T use withContext inside a flow builder
fun badFlow(): Flow<Result> = flow {
    withContext(Dispatchers.IO) {  // ⚠ breaks flow invariants
        emit(fetchData())            // emit from wrong context = exception
    }
}

// ✅ If you need IO inside flow builder, use withContext only for the work:
fun goodFlow(): Flow<Result> = flow {
    val data = withContext(Dispatchers.IO) { db.query() }  // work on IO
    emit(data)  // emit back on original context ✓
}

// flowOn is composable — multiple can stack
flow { emit(readDisk()) }
    .flowOn(Dispatchers.IO)          // readDisk() on IO
    .map { parse(it) }
    .flowOn(Dispatchers.Default)     // parse() on Default
    .collect { render(it) }          // render() on Main

flowOn works by inserting a channel between upstream and downstream, effectively decoupling them. The operator applies only to the operators above it in the chain, not below. Stacking multiple flowOn calls gives you fine-grained dispatcher control at each transformation stage — parse JSON on Default, write to database on IO, render on Main. This is how you build zero-compromise performance pipelines without manually managing coroutine scopes at each step.

This is the canonical debounce + flatMapLatest pattern. The requirement has two parts: wait until the user stops typing (debounce), then cancel any in-flight search when a new query arrives (flatMapLatest). Getting both right is what separates a polished search experience from a janky one.

// ViewModel — search with debounce + cancellation
class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")

    val results: StateFlow<SearchState> = _query
        .debounce(300)              // wait 300ms after last keystroke
        .filter { it.length >= 2 }   // ignore very short queries
        .distinctUntilChanged()      // skip if same query re-typed
        .flatMapLatest { query ->    // cancel previous search on new query
            flow {
                emit(SearchState.Loading)
                val data = api.search(query)
                emit(SearchState.Success(data))
            }.catch { e ->
                emit(SearchState.Error(e.message))
            }
        }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), SearchState.Idle)

    fun onQueryChanged(query: String) { _query.value = query }
}

// In the Fragment / Composable:
// editText.addTextChangedListener { viewModel.onQueryChanged(it.toString()) }
// OR in Compose: TextField(onValueChange = { viewModel.onQueryChanged(it) })

Why flatMapLatest, not flatMapMerge? flatMapLatest cancels the previous inner flow whenever a new value arrives — if the user types "an", then quickly adds "d", the "an" search is cancelled mid-flight. flatMapMerge would run all searches concurrently and results could arrive out of order (the "an" result might arrive after the "and" result, showing stale data). flatMapConcat would queue searches and never cancel, causing a backlog of stale queries. For real-time search, flatMapLatest is always the right choice.

debounce(300) prevents firing a search for every intermediate keystroke. "android" would otherwise trigger 7 API calls; with 300ms debounce, only one fires after the user pauses. Combine with distinctUntilChanged to also skip re-typing the same word. The full chain: debounce → filter(minLength) → distinctUntilChanged → flatMapLatest covers all the edge cases interviewers test for.

All three transform each emitted value into an inner Flow, but differ in how they handle concurrency: Latest cancels the previous inner flow, Merge runs all concurrently, Concat queues them sequentially. Choosing the wrong one causes either stale data, missed results, or queued-up backlog bugs.

// flatMapLatest — cancel previous, start fresh
// Use case: search, live filters, switching between tabs
viewModel.selectedTab
    .flatMapLatest { tab -> repo.getItemsForTab(tab) }
    .collect { render(it) }
// User switches tab → previous tab's Flow cancelled immediately

// flatMapMerge — run ALL concurrently, merge results as they arrive
// Use case: parallel file uploads, batch API calls
selectedFiles
    .flatMapMerge(concurrency = 3) { file -> uploadFlow(file) }
    .collect { result -> updateUploadStatus(result) }
// All files upload simultaneously (max 3 at a time), results arrive as ready

// flatMapConcat — wait for previous to complete, then start next
// Use case: ordered operations, dependent sequential steps
orderIds
    .flatMapConcat { id -> processOrderFlow(id) }
    .collect { updateOrderStatus(it) }
// Order 1 fully processed → Order 2 starts. Preserves order, no overlap.

// ⚠ Common bugs:
// Using flatMapMerge for search → results arrive out of order (stale UI)
// Using flatMapConcat for tabs → previous tab's slow load blocks new tab
// Using flatMapLatest for uploads → switching file cancels the upload!

Memory for the decision: Does new input make old work irrelevant? → flatMapLatest (search, filter, tab switch). Do you need all results but don't care about order? → flatMapMerge (batch uploads, parallel fetches). Do results need to arrive in strict order? → flatMapConcat (ordered operations, dependency chains). The concurrency parameter in flatMapMerge lets you throttle parallelism — flatMapMerge(concurrency = 1) is equivalent to flatMapConcat.

Use combine when you need the latest value from multiple flows. It emits whenever ANY source emits, pairing each emission with the most recent value from the other sources. The edge case: combine waits until ALL source flows have emitted at least once before producing any output.

// combine — emits when ANY source updates, uses LATEST from each
val uiState: StateFlow<ProfileUiState> = combine(
    userRepo.userFlow(),
    settingsRepo.settingsFlow(),
    subscriptionRepo.tierFlow()
) { user, settings, tier ->
    ProfileUiState(
        displayName = user.name,
        theme = settings.theme,
        isPremium = tier == Tier.Premium
    )
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), ProfileUiState.default())

// ⚠ Edge case: combine blocks until ALL flows emit once
// If settingsFlow delays 2s, no output for 2s — even if user arrived instantly
// Fix: give each flow a default with onStart or use StateFlow (always has value)
val safe = combine(
    userRepo.userFlow().onStart { emit(User.empty()) },
    settingsRepo.settingsFlow().onStart { emit(Settings.default()) }
) { user, settings -> merge(user, settings) }

// zip vs combine — important distinction:
// zip → pairs emissions ONE-TO-ONE, waits for partner emission
flowA.zip(flowB) { a, b -> Pair(a, b) }  // A=1,B=1 → (1,1); A=2,B=? → waits

// combine → uses LATEST from each, fires on any update
combine(flowA, flowB) { a, b -> Pair(a, b) }  // A=2, B still 1 → (2,1) emitted

combine vs zip — a frequent interview trap. zip pairs emissions positionally: it waits until both flows have produced a new value before emitting. If flowA emits 3 values but flowB only emits 1, zip will pause after the first pair until flowB emits again. This makes zip suitable for request-response pairing. combine fires whenever any source emits and uses the latest known value from all other sources — this is almost always what you want for UI state that depends on multiple data sources updating at independent rates.

The bug is caused by replay > 0 on the SharedFlow. When the Fragment is recreated after rotation, it re-subscribes to the SharedFlow and immediately receives the cached navigation event — triggering a second navigation. This is one of the most common Flow bugs in production apps.

// ❌ BUG — replay=1 re-sends navigation event after rotation
private val _events = MutableSharedFlow<NavEvent>(replay = 1)

// ✅ FIX 1 — replay=0 (default): events are fire-and-forget
private val _events = MutableSharedFlow<NavEvent>()  // replay=0
// New collectors get nothing — only collectors ACTIVE at emission time receive it

// ✅ FIX 2 — consume-once wrapper (more explicit)
data class ConsumedEvent<T>(val content: T, var consumed: Boolean = false) {
    fun getOrNull(): T? = if (consumed) null else content.also { consumed = true }
}

// ✅ FIX 3 — collect in lifecycle-aware scope (prevents resubscription during recreation)
// In Fragment — use repeatOnLifecycle(STARTED) not lifecycleScope.launch directly
viewLifecycleOwner.lifecycleScope.launch {
    viewLifecycleOwner.lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.events.collect { event ->
            when (event) {
                is NavEvent.GoToDetail -> findNavController().navigate(event.id)
            }
        }
    }
}
// repeatOnLifecycle STARTED: collection stops when app backgrounds (lifecycle below STARTED)

Why repeatOnLifecycle(STARTED)? Using lifecycleScope.launch { flow.collect{} } directly keeps collecting even when the app is backgrounded — wasting battery and risking crashes if the Fragment's view is destroyed. repeatOnLifecycle(STARTED) starts a new collection coroutine when the lifecycle reaches STARTED (foreground) and cancels it when it drops below STARTED (background). For navigation events, STARTED is correct — you only want to navigate when the screen is visible. Using RESUMED would miss events during brief transitions.

Backpressure occurs when the producer emits faster than the collector can process. Flow handles this with suspension by default — the producer waits for the collector. But sometimes waiting isn't acceptable. Three operators address this: buffer (queue emissions), conflate (keep only latest), collectLatest (cancel slow collector on new value).

// Default: producer waits for collector (backpressure = suspension)
flow { for (i in 1..100) { emit(i); delay(10) } }
    .collect { delay(100) }  // slow collector, producer waits

// buffer() — producer runs freely, queue builds up in channel
sensorFlow()
    .buffer(capacity = 64)    // up to 64 items queued
    .collect { processReading(it) }
// Use when: you must not miss any emission AND collector is occasionally slow

// conflate() — keep only the LATEST value, drop intermediates
stockPriceFlow()
    .conflate()              // if collector busy, skip old prices
    .collect { updateChart(it) }
// Use when: missing intermediate values is OK, only latest matters

// collectLatest — cancel ongoing collection when new value arrives
queryFlow()
    .collectLatest { query ->
        delay(300)              // cancelled if new query arrives before 300ms
        val result = api.search(query)
        updateResults(result)
    }
// Use when: new value makes current work irrelevant (debounce alternative)

// StateFlow automatically conflates — setting same value twice = one emission
// SharedFlow(extraBufferCapacity = N) — buffers N items before suspending

buffer vs conflate vs collectLatest — the decision is about what you can afford to lose. Buffer is for lossless pipelines where every emission matters (audio samples, analytics events). Conflate is for data where only the current value matters (UI updates, stock prices, location). collectLatest is for reactive triggers where new input makes current work stale (search, filter). In practice, StateFlow is the most common conflate scenario — it only holds the latest state and already conflates by design.

Use callbackFlow to wrap any callback-based API into a Flow. It provides a SendChannel to bridge the callback world with the suspension world. The critical part is the awaitClose block — without it, the Flow completes immediately after the callback is registered, and the producer leaks.

// Converting LocationManager callback to Flow
fun Context.locationFlow(): Flow<Location> = callbackFlow {
    val listener = LocationListener { location ->
        trySend(location)  // non-suspending send, drops if buffer full
    }

    locationManager.requestLocationUpdates(
        LocationManager.GPS_PROVIDER, 1000L, 10f, listener
    )

    // CRITICAL: awaitClose cleans up when collector cancels
    awaitClose {
        locationManager.removeUpdates(listener)  // no leak!
    }
}

// Pitfall 1: using send() instead of trySend() on Main thread
// send() is a suspend function — can't call from non-suspending callback
// trySend() returns a Result — check it if you care about drops

// Pitfall 2: missing awaitClose → channel closed immediately after registering
fun badFlow() = callbackFlow<Event> {
    api.setCallback { trySend(it) }
    // ❌ NO awaitClose — Flow completes, callback is never cleaned up
}

// Pitfall 3: emitting after close — trySend returns isFailure safely
// Pitfall 4: cold vs hot — each collector registers a NEW callback listener
// Fix: shareIn the callbackFlow to share one listener among collectors
val sharedLocation = locationFlow()
    .shareIn(appScope, SharingStarted.WhileSubscribed(), replay = 1)

callbackFlow vs flow builder: the regular flow { } builder cannot use non-suspending code that calls back on a different thread — the emit must happen on the same coroutine. callbackFlow creates a channel internally, allowing any thread to send values. The channel is backed by a buffer (default capacity 64) so fast callbacks don't immediately block. The awaitClose block runs when the Flow is cancelled — equivalent to lifecycle cleanup in a LifecycleObserver.

The problem is double execution — each Fragment collector independently triggers the Room Flow, opening two separate database observers. For a simple query this is wasteful; for a complex query with joins, this causes measurable performance impact. The fix is converting the cold Flow to a hot StateFlow using stateIn.

// ❌ Problem: two collectors = two independent Room observers
class UserViewModel : ViewModel() {
    val users: Flow<List<User>> = userDao.getAllUsers()
    // Fragment A collects → Room observer 1 registered
    // Fragment B collects → Room observer 2 registered
}

// ✅ Fix: stateIn converts to hot StateFlow — one upstream, N collectors
class UserViewModel : ViewModel() {
    val users: StateFlow<List<User>> = userDao.getAllUsers()
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5_000),
            initialValue = emptyList()
        )
    // Fragment A + B both collect → SINGLE Room observer
    // Rotation: 5s grace → no Room restart, no flicker
}

// ✅ Also add intermediate mapping before stateIn:
val sortedUsers: StateFlow<List<UserUi>> = userDao.getAllUsers()
    .map { users -> users.sortedBy { it.name }.map(UserUi::from) }
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), emptyList())
// map runs ONCE upstream — both Fragments see the same sorted+mapped result

A secondary problem: without stateIn, if both Fragments use repeatOnLifecycle(STARTED), they stop collecting when backgrounded. When they return, they re-collect and the cold Flow restarts from the beginning — emitting a flash of old data before the fresh query completes. StateFlow solves this: the 5s grace period keeps the Room observer alive across rotation, and the last value is delivered immediately on reconnect — no loading flash.

Flow has two places where exceptions can occur: upstream (inside the flow builder or operators) and downstream (inside the collect block). They behave very differently. The catch operator only catches upstream exceptions — exceptions thrown inside collect are NOT caught by it and propagate up to the coroutine.

// catch{} handles UPSTREAM exceptions only
apiFlow()
    .map { transform(it) }       // exception here → caught
    .catch { e ->
        emit(Result.failure(e))    // recover with default value
    }
    .collect { result ->
        dangerousUiOp(result)     // exception here → NOT caught by catch{}
    }

// ✅ Pattern 1: wrap collect block in try-catch
try {
    flow.collect { render(it) }
} catch (e: Exception) {
    showError(e)
}

// ✅ Pattern 2: use onEach + catch (preferred — keeps operators clean)
flow
    .onEach { render(it) }      // processing logic here
    .catch { e -> showError(e) } // catches both upstream AND onEach exceptions
    .launchIn(viewModelScope)   // starts collection as coroutine

// ✅ Pattern 3: model errors in the type system
sealed class Result<out T> {
    data class Success<T>(val data: T) : Result<T>()
    data class Error(val exception: Throwable) : Result<Nothing>()
}
fun safeFlow(): Flow<Result<Data>> = flow {
    emit(Result.Success(api.fetch()))
}.catch { e -> emit(Result.Error(e)) }
// Collector receives Success or Error — no exceptions escape

The onEach + catch + launchIn pattern is particularly clean because onEach runs in the upstream chain, so its exceptions are caught by catch. Modelling errors as sealed class values (Result/Either) is the most robust approach — it makes error handling explicit in the type system, and no uncaught exception can crash the coroutine. The StateFlow approach in ViewModels takes this further: the ViewModel maps exceptions to a UiState.Error value and emits it — the UI renders the error state rather than crashing.

Testing Flows requires handling their asynchronous, time-dependent nature. The standard approach combines runTest (coroutines test runner) with Turbine (a Flow testing library that provides a clean assertion API). Without Turbine, you'd need brittle toList() collectors that require the flow to terminate — useless for infinite StateFlows.

// Dependency: testImplementation("app.cash.turbine:turbine:1.1.0")

class SearchViewModelTest {
    private val fakeRepo = FakeSearchRepo()
    private val viewModel = SearchViewModel(fakeRepo)

    // Test StateFlow emissions with Turbine
    @Test
    fun `search emits Loading then Success`() = runTest {
        viewModel.results.test {
            assertEquals(SearchState.Idle, awaitItem())    // initial value

            viewModel.onQueryChanged("android")
            assertEquals(SearchState.Loading, awaitItem())  // loading state
            val result = awaitItem()
            assertTrue(result is SearchState.Success)

            cancelAndIgnoreRemainingEvents()  // clean up infinite flow
        }
    }

    // Test that debounce works
    @Test
    fun `rapid typing debounces to single API call`() = runTest {
        viewModel.onQueryChanged("a")
        viewModel.onQueryChanged("an")
        viewModel.onQueryChanged("and")
        advanceTimeBy(350)  // advance past debounce window

        assertEquals(1, fakeRepo.searchCallCount)  // only one API call
    }

    // Test error handling
    @Test
    fun `API error emits Error state`() = runTest {
        fakeRepo.shouldThrow = true
        viewModel.results.test {
            viewModel.onQueryChanged("android")
            skipItems(2)  // skip Idle, Loading
            assertTrue(awaitItem() is SearchState.Error)
            cancelAndIgnoreRemainingEvents()
        }
    }
}

Why Turbine? Without it, testing a StateFlow requires creating a collector coroutine, manually managing it, using timing-sensitive advanceUntilIdle(), and building a mutable list of emissions — fragile and verbose. Turbine wraps this with awaitItem() (suspends until next emission), awaitComplete(), awaitError(), skipItems(n), and cancelAndIgnoreRemainingEvents() for infinite flows. Tests become declarative and easy to read. Also set Dispatchers.setMain(UnconfinedTestDispatcher()) in your @Before to replace the Main dispatcher in unit tests.

StateFlow internally uses an atomic reference and a lock-free notification mechanism. The value is stored in a @Volatile field backed by an atomic state object. Updates use compare-and-set under the hood to guarantee thread safety without heavy synchronization. The equality skip is implemented as a structural equality check before notifying subscribers.

// StateFlow internals (simplified conceptual model)
// Actual impl: kotlinx.coroutines.flow.StateFlowImpl

// 1. Value storage — atomic + volatile
// @Volatile var _value: T  ← readable from any thread without locking
// setState() uses synchronised block to update and notify

// 2. Equality check — skips identical consecutive values
data class User(val id: Int, val name: String)

val state = MutableStateFlow(User(1, "Alice"))
state.value = User(1, "Alice")  // NO emission — equals() returns true
state.value = User(1, "Bob")    // emits — equals() returns false

// ⚠ Trap: mutable objects and equals()
data class MutableList... // if you mutate contents, == may still be true
val list = mutableListOf("a")
val state2 = MutableStateFlow(list)
list.add("b")
state2.value = list   // same reference → == true → NO emission! 🐛
// Fix: always use immutable data classes and copy semantics
state2.update { currentList -> currentList + "b" }  // new list = new emission

// 3. update() is atomic compare-and-set
val counter = MutableStateFlow(0)
counter.update { it + 1 }  // atomic: reads current, computes next, CAS
// Retries automatically if another thread updated concurrently
// Safe to call from multiple threads simultaneously

// 4. Notification — subscribers are stored as linked list of continuations
// On value update, all waiting collect{} continuations are resumed

The mutable object trap is a production bug that catches many developers: if you store a mutable list or class in StateFlow and mutate it in-place, then reassign the same reference, StateFlow sees the same object reference, calls equals() which returns true (same contents = same reference), and skips the notification. The fix is immutable data: use Kotlin data classes, copy-on-write semantics, or persistentListOf from kotlinx.collections.immutable. update { } is the correct way to modify StateFlow values atomically — it uses compare-and-set retry loop so concurrent updates are safe.

Parallel network requests with Flow require launching both concurrently. The pattern is combine on two Flows that each wrap their own suspend call, or using zip if both need to arrive together. For one-shot parallel fetches, using async/await inside a single flow is also valid.

// Pattern 1: combine two Flows (live-updating, DB-backed)
val profileUiState = combine(
    userRepository.userFlow(userId),   // room flow, updates on DB change
    orderRepository.ordersFlow(userId) // room flow, updates on DB change
) { user, orders ->
    ProfileState.Ready(user, orders)
}.onStart { emit(ProfileState.Loading) }
 .catch { e -> emit(ProfileState.Error(e)) }
 .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), ProfileState.Loading)

// Pattern 2: one-shot parallel network calls inside a flow
val profileFlow: Flow<ProfileState> = flow {
    emit(ProfileState.Loading)
    val (user, orders) = coroutineScope {
        val userDeferred = async { api.getUser(userId) }
        val ordersDeferred = async { api.getOrders(userId) }
        Pair(userDeferred.await(), ordersDeferred.await())
    }
    emit(ProfileState.Ready(user, orders))
}.catch { e -> emit(ProfileState.Error(e.message)) }

// ⚠ Anti-pattern: sequential fetches (no parallelism)
val bad = flow {
    val user = api.getUser(userId)      // waits 500ms
    val orders = api.getOrders(userId)  // waits another 500ms = 1000ms total
    emit(ProfileState.Ready(user, orders))
}

Pattern 1 (combine on live Flows) is best when both data sources are Room DAOs — you get a live-updating UI that automatically re-renders whenever the database changes, and both queries run in parallel on separate coroutines. Pattern 2 (coroutineScope + async inside flow) is best for one-shot network-only fetches where you need explicit parallelism. The key insight: coroutineScope { async + await } inside a flow builder runs both requests concurrently while respecting structured concurrency — if either fails, the other is cancelled automatically.

collect is a suspend function that blocks the current coroutine until the flow completes. launchIn(scope) is a terminal operator that launches the collection as a new coroutine in the given scope, returning a Job — the current coroutine continues immediately. They're functionally equivalent but differ in concurrency model.

// collect — suspends caller until flow terminates
viewModelScope.launch {
    myFlow.collect { value ->
        processValue(value)
    }
    // code here runs AFTER flow completes (or never, for infinite flows)
}

// launchIn — fire-and-forget, returns Job
myFlow
    .onEach { value -> processValue(value) }
    .launchIn(viewModelScope)  // returns Job immediately
// code here continues immediately — doesn't wait for flow

// ✅ launchIn preferred when launching multiple flows in init{}
init {
    userFlow.onEach { _userState.value = it }.launchIn(viewModelScope)
    settingsFlow.onEach { _settings.value = it }.launchIn(viewModelScope)
    // both flow concurrently — if you used collect, they'd be sequential!
}

// ✅ collect preferred when you need sequential result after flow completes
val allItems = mutableListOf<Item>()
finiteFlow.collect { allItems.add(it) }
processAll(allItems)  // runs after flow finishes

// ⚠ Common mistake: calling collect twice in the same launch for different flows
viewModelScope.launch {
    flowA.collect { ... }  // blocks here for infinite flow
    flowB.collect { ... }  // never reaches this!
}

The most dangerous mistake is calling collect twice in a single launch block for two infinite flows — the second collect never starts because the first never completes. This is why launchIn exists: it makes the concurrency explicit. The pattern flow.onEach{}.catch{}.launchIn(scope) is idiomatic for ViewModel init blocks that need to observe multiple sources in parallel.

The key is combining lifecycle-aware collection with stateIn's WhileSubscribed. The polling Flow in the ViewModel runs only while there's an active subscriber — and the Fragment's repeatOnLifecycle(STARTED) cancels collection when backgrounded, causing the subscriber count to drop to zero.

// ViewModel — polling flow with automatic lifecycle suspension
class DashboardViewModel : ViewModel() {
    private fun pollingFlow(): Flow<DashboardData> = flow {
        while (true) {
            emit(api.getDashboardData())
            delay(30_000)   // 30s poll interval
        }
    }

    val dashboardState: StateFlow<DashboardData?> = pollingFlow()
        .catch { /* log, retry */ }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5_000),  // stops 5s after last collector
            initialValue = null
        )
}

// Fragment — lifecycle-aware collection
viewLifecycleOwner.lifecycleScope.launch {
    viewLifecycleOwner.lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.dashboardState.collect { data ->
            render(data)
        }
    }
}

// Lifecycle flow:
// App foreground → repeatOnLifecycle starts collecting → WhileSubscribed sees subscriber → polling starts
// App background → repeatOnLifecycle cancels collect → no subscribers → 5s grace → polling stops
// App foreground again → repeatOnLifecycle restarts → polling resumes (no restart = within 5s)

// ✅ For retry on error:
pollingFlow()
    .retryWhen { cause, attempt ->
        cause is IOException && attempt < 3
    }
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), null)

This architecture — polling Flow + stateIn(WhileSubscribed) + repeatOnLifecycle(STARTED) — creates a perfectly lifecycle-coupled polling mechanism without any explicit lifecycle callbacks, manual start/stop flags, or LifecycleObserver boilerplate. The three components work together: WhileSubscribed controls upstream lifecycle, repeatOnLifecycle controls collection lifecycle, and the StateFlow buffers the 5s gap to avoid redundant restarts on rotation. Add retryWhen for resilient polling that auto-retries on network errors.

A Channel is a hot, concurrent communication primitive — a blocking queue for coroutines. A Flow is a cold, declarative pipeline. The mental model: a Channel is a pipe with two ends (send/receive). A Flow is a recipe that runs when collected. In modern Android code, you almost never use Channel directly — you use Flow, SharedFlow, or StateFlow instead. But understanding Channels explains how Flow operators like buffer and callbackFlow work internally.

// Channel — hot, send/receive, one-to-one (or broadcast)
val channel = Channel<Int>(capacity = Channel.BUFFERED)

// Producer coroutine
viewModelScope.launch {
    for (i in 1..10) channel.send(i)
    channel.close()
}

// Consumer coroutine
viewModelScope.launch {
    for (item in channel) process(item)  // iterates until close()
}

// Channel types:
// RENDEZVOUS (0)   — sender suspends until receiver ready (zero buffer)
// BUFFERED         — 64 element buffer (default)
// UNLIMITED        — unbounded buffer, never suspends sender (OOM risk)
// CONFLATED        — keeps only latest, sender never suspends

// Flow vs Channel — when to use each:
// Flow   → declarative pipeline, cold, composable operators, most Android cases
// Channel → hot producer-consumer, cross-coroutine communication, one-receiver

// Modern alternative — prefer SharedFlow over broadcast Channel
// MutableSharedFlow() replaced BroadcastChannel (deprecated since 1.5.0)

// ✅ Real use case: Channel in a ViewModel for command pattern
private val commandChannel = Channel<Command>(Channel.UNLIMITED)
val commandFlow = commandChannel.receiveAsFlow()  // expose as Flow

Channel vs SharedFlow for events: Both work for one-shot events, but they have different backpressure semantics. Channel.UNLIMITED never suspends the sender — events are always queued. SharedFlow with extraBufferCapacity and an onBufferOverflow strategy is more explicit. For ViewModel → UI events, SharedFlow(replay=0) is the modern idiomatic choice. Channels are still valuable for internal ViewModel orchestration — a "command channel" pattern where the ViewModel processes one command at a time using receiveAsFlow() converted to a regular Flow.

Configure the SharedFlow with extraBufferCapacity and a buffer overflow strategy. This buffers events while the collector is temporarily unavailable (e.g., during a configuration change gap) without losing any. The choice of overflow strategy determines what happens when the buffer fills up.

// SharedFlow for lossless analytics events
private val _analyticsEvents = MutableSharedFlow<AnalyticsEvent>(
    replay = 0,                          // new collectors don't get old events
    extraBufferCapacity = 64,            // buffer 64 events while collector is unavailable
    onBufferOverflow = BufferOverflow.DROP_OLDEST  // if buffer full, drop oldest
)

// Emit never suspends — analytics must not block business logic
fun trackEvent(event: AnalyticsEvent) {
    _analyticsEvents.tryEmit(event)  // returns false only if buffer full + DROP strategy
}

// BufferOverflow options:
// SUSPEND         — emit() suspends until space available (back-pressure)
// DROP_OLDEST     — make room by dropping oldest buffered item
// DROP_LATEST     — discard the new item (tryEmit returns false)

// For critical events (no loss allowed): use extraBufferCapacity = Channel.UNLIMITED
// BUT: unlimited buffer = OOM risk under heavy load

// For navigation events (must not lose, few events): replay=0 + buffer=1
private val _navEvents = MutableSharedFlow<NavEvent>(
    extraBufferCapacity = 1,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)
// Survives brief no-collector gap (config change) without losing the navigation event

The extraBufferCapacity parameter is separate from the replay cache. Replay stores items for new subscribers. extraBufferCapacity stores items for existing subscribers that are temporarily slow. For analytics, you typically want replay=0 (new screens shouldn't replay old events) with a generous extra buffer and DROP_OLDEST overflow. For navigation events, replay=0 with extraBufferCapacity=1 ensures the event survives the ~100ms configuration change gap without replaying it indefinitely to future screens.

The offline-first pattern with Flow is elegant: Room emits the cached data immediately, then your network call updates the database, which triggers Room to re-emit with the fresh data. The UI collects a single Flow that automatically delivers both the cached and fresh values in sequence.

// Repository — offline-first with Flow
class UserRepository(
    private val dao: UserDao,
    private val api: UserApi
) {
    fun getUserFlow(id: String): Flow<User> = flow {
        // Step 1: emit cached data instantly (may be null/stale)
        val cached = dao.getUserById(id)
        if (cached != null) emit(cached)

        // Step 2: fetch fresh data from network
        try {
            val fresh = api.getUser(id)
            dao.insertUser(fresh)   // update cache
            emit(fresh)              // emit fresh data
        } catch (e: IOException) {
            if (cached == null) throw e  // no cache AND no network = error
            // else: silently use cached data
        }
    }

    // ✅ Better: use Room Flow directly — it emits whenever DB changes
    fun getUserWithAutoUpdate(id: String): Flow<User> {
        // Trigger refresh asynchronously
        return dao.getUserFlowById(id)  // Room Flow: emits on every DB update
            .onStart {
                // Start network refresh alongside Flow collection
                coroutineScope {
                    launch(Dispatchers.IO) {
                        try {
                            val fresh = api.getUser(id)
                            dao.insertUser(fresh)  // Room emits updated value automatically
                        } catch (e: Exception) { /* log */ }
                    }
                }
            }
    }
}

The Room Flow approach is the cleanest: dao.getUserFlow(id) emits every time the database row changes. Trigger a background refresh onStart, which writes to the DB, which automatically triggers Room to re-emit with fresh data. The UI collector sees: (1) cached value immediately on subscription, (2) fresh value when the network completes — with zero timing code written. This is the single-source-of-truth pattern: the database is the truth, the network only writes to it.

collectAsState() collects forever — even when the app is backgrounded — because it ties the collection to the composable's lifetime, not the Android lifecycle. collectAsStateWithLifecycle() from lifecycle-runtime-compose pauses collection when the lifecycle drops below STARTED and resumes when it returns. For any Android app, the lifecycle-aware version should always be preferred.

// ❌ collectAsState — no lifecycle awareness
@Composable
fun UserScreen(viewModel: UserViewModel = hiltViewModel()) {
    val uiState by viewModel.uiState.collectAsState()
    // Collects even when app is backgrounded — wastes resources
    // If uiState triggers network call, it fires even in background
}

// ✅ collectAsStateWithLifecycle — lifecycle-coupled
// dependency: implementation("androidx.lifecycle:lifecycle-runtime-compose:2.8+")
@Composable
fun UserScreen(viewModel: UserViewModel = hiltViewModel()) {
    val uiState by viewModel.uiState.collectAsStateWithLifecycle()
    // Pauses when app backgrounds, resumes on foreground
    // Works with stateIn(WhileSubscribed(5000)) to stop upstream too
}

// The full picture — three lifecycle-aware layers working together:
// 1. collectAsStateWithLifecycle() — stops collection when backgrounded
// 2. stateIn(WhileSubscribed(5000)) — stops upstream 5s after last collector
// 3. viewModelScope — cancelled when ViewModel is destroyed

// When you combine all three:
// Background → collectAsStateWithLifecycle stops → 5s later → upstream stops
// Foreground → collectAsStateWithLifecycle restarts → upstream resumes
// ViewModel destroyed → viewModelScope cancelled → everything cleaned up

// Custom lifecycle state (RESUMED for in-foreground only work):
viewModel.uiState.collectAsStateWithLifecycle(
    minActiveState = Lifecycle.State.RESUMED  // only collect when fully in foreground
)

The practical impact is measurable in battery life and network usage. A location-tracking flow using collectAsState() never stops when the user switches apps — the GPS keeps firing, the UI keeps recomposing, and the battery drains. With collectAsStateWithLifecycle() + stateIn(WhileSubscribed), the location flow stops within 5 seconds of backgrounding. This is the kind of optimization that distinguishes a professional Android developer from someone who just makes things work.

Kotlin Flow provides retry and retryWhen operators that make retry logic clean and composable. retryWhen gives you the cause and attempt number, letting you implement exponential backoff with jitter — the production-quality pattern for avoiding thundering herd problems.

// retryWhen — full control over retry logic
fun fetchWithRetry(): Flow<Data> = flow {
    emit(api.getData())
}.retryWhen { cause, attempt ->
    val shouldRetry = cause is IOException && attempt < 4
    if (shouldRetry) {
        // Exponential backoff with jitter: 1s, 2s, 4s, 8s (±500ms jitter)
        val backoff = (2.pow(attempt.toInt()) * 1000L)
            .coerceAtMost(30_000L)  // cap at 30s
        val jitter = (-500..500).random().toLong()
        delay(backoff + jitter)
    }
    shouldRetry
}

// Full production pattern with state tracking:
fun resilientFetch(): Flow<NetworkState> = flow {
    emit(NetworkState.Loading)
    emit(NetworkState.Success(api.getData()))
}.retryWhen { cause, attempt ->
    if (cause is HttpException && cause.code() == 401) {
        return@retryWhen false  // don't retry auth errors
    }
    if (attempt >= 3) return@retryWhen false
    delay(1000L * (2.pow(attempt.toInt())))
    true
}.catch { e ->
    emit(NetworkState.Error(e.message))  // all retries exhausted
}

// ✅ Combine with onEach to emit Retrying state:
flow { emit(api.getData()) }
    .onStart { emit(State.Loading) }
    .retryWhen { _, attempt -> emit(State.Retrying(attempt)); delay(2000); true }

Exponential backoff with jitter is critical in production. Without jitter, all clients that fail simultaneously will retry at the same intervals, hammering the server in synchronized waves (the "thundering herd" problem). Adding ±500ms random jitter spreads the retries across time. The coerceAtMost(30_000) cap prevents the backoff from growing indefinitely. For non-retryable errors (HTTP 401, 403), return false immediately to fail fast rather than wasting three retry attempts on guaranteed failures.

Yes — if you use _state.value = _state.value + something (read-modify-write), you have a classic race condition. Two coroutines reading the same value and writing independently can cause one update to overwrite the other. The fix is update { }, which is an atomic compare-and-set operation that retries if the value was modified between read and write.

// ❌ RACE CONDITION — read-modify-write is NOT atomic
val _count = MutableStateFlow(0)

// Two coroutines run concurrently:
// Coroutine A: reads 0, prepares to write 1
// Coroutine B: reads 0, prepares to write 1
// Coroutine A: writes 1
// Coroutine B: writes 1   ← lost one increment!
viewModelScope.launch { _count.value = _count.value + 1 }  // ❌
viewModelScope.launch { _count.value = _count.value + 1 }  // ❌

// ✅ FIX — update{} is atomic (compare-and-set with retry)
viewModelScope.launch { _count.update { it + 1 } }  // safe
viewModelScope.launch { _count.update { it + 1 } }  // safe
// If both read 0, first writer succeeds; second detects stale read → retries
// Result: always 2, never 1

// ✅ For complex state objects — update the whole state atomically
data class UiState(val items: List<Item>, val isLoading: Boolean)
val _state = MutableStateFlow(UiState(emptyList(), false))

// ❌ Two fields updated separately — brief inconsistent state
_state.value = _state.value.copy(isLoading = true)   // other thread sees partial update
_state.value = _state.value.copy(items = newItems)

// ✅ Single atomic update — always consistent
_state.update { it.copy(items = newItems, isLoading = false) }

update { } internally reads the current value, applies your transform, then uses an atomic compare-and-set (CAS) operation to write. If another thread modified the value between the read and the CAS, the CAS fails and the entire operation retries with the new current value. This is lock-free concurrency — no mutex, no synchronized block, no deadlock risk. For complex state objects with multiple fields, always use a single update { it.copy(...) } call to atomically update all fields at once, eliminating the window where the state could be seen in a partially-updated form.

Real-time chat requires combining a long-lived WebSocket connection (hot stream) with persistent storage (Room) and lifecycle-aware UI collection. The key insight: the WebSocket delivers raw messages, Room is the single source of truth, and a SharedFlow bridges the two with lifecycle safety.

// 1. DATA LAYER — WebSocket as callbackFlow
class ChatWebSocketManager {
    fun messageFlow(roomId: String): Flow<ChatMessage> = callbackFlow {
        val ws = OkHttpClient().newWebSocket(request, object : WebSocketListener() {
            override fun onMessage(ws: WebSocket, text: String) {
                trySend(parseMessage(text))
            }
            override fun onFailure(ws: WebSocket, t: Throwable, ...) {
                close(t)  // close flow with error → triggers retryWhen
            }
        })
        awaitClose { ws.close(1000, "Lifecycle end") }
    }.retryWhen { cause, attempt ->
        delay(1000L * (2.pow(attempt.toInt())).coerceAtMost(30000L))
        true
    }
}

// 2. REPOSITORY — write to DB, expose DB flow
class ChatRepository(
    private val ws: ChatWebSocketManager,
    private val dao: MessageDao
) {
    // Persist incoming messages — Room flow auto-emits on DB changes
    suspend fun startListening(roomId: String) {
        ws.messageFlow(roomId)
            .collect { msg -> dao.insertMessage(msg) }  // write to single source of truth
    }

    fun messagesFlow(roomId: String): Flow<List<ChatMessage>> =
        dao.getMessagesFlow(roomId)  // Room Flow, auto-updates on insert
}

// 3. VIEWMODEL — convert to StateFlow
class ChatViewModel : ViewModel() {
    val messages = repo.messagesFlow(roomId)
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), emptyList())

    init {
        viewModelScope.launch(Dispatchers.IO) { repo.startListening(roomId) }
    }
}

// 4. UI — lifecycle-aware collection in Compose
val messages by viewModel.messages.collectAsStateWithLifecycle()

The architecture is cleanly layered. callbackFlow wraps the WebSocket with automatic reconnect via retryWhen. Room is the single source of truth — messages are written to DB as they arrive, and the UI always reads from the DB, not directly from the WebSocket. This means offline reading works (messages persist), and the UI gets a clean Flow from Room that handles all the complexity of the WebSocket lifecycle. WhileSubscribed(5000) keeps the Room observer alive during rotation, and collectAsStateWithLifecycle pauses collection when the app backgrounds.

This code review tests pattern recognition across multiple Flow anti-patterns. There are 5 bugs — finding all of them demonstrates production-level Flow expertise.

// ❌ PR CODE WITH MULTIPLE BUGS — can you find them?
class BadFlowViewModel : ViewModel() {

    // Bug 1:
    val events = MutableSharedFlow<NavEvent>(replay = 1)

    // Bug 2:
    val users: Flow<List<User>> = dao.getAllUsers()

    // Bug 3:
    fun loadData() = viewModelScope.launch {
        dao.getAllUsers().collect { userList ->
            _state.value = userList
        }
        api.getAnnouncements().collect { items ->  // Bug 3
            _announcements.value = items
        }
    }

    // Bug 4:
    fun updateCount() {
        _count.value = _count.value + 1
    }

    // Bug 5:
    fun observeSearch(query: StateFlow<String>) {
        viewModelScope.launch {
            query
                .flatMapLatest { api.search(it) }  // Bug 5
                .collect { _results.value = it }
        }
    }
}

// ✅ FIXED VERSION:
class GoodFlowViewModel : ViewModel() {
    // Fix 1: replay=0 for navigation events
    val events = MutableSharedFlow<NavEvent>()

    // Fix 2: stateIn to avoid cold Flow double-execution
    val users = dao.getAllUsers()
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), emptyList())

    // Fix 3: launchIn for concurrent collection
    init {
        dao.getAllUsers().onEach { _state.value = it }.launchIn(viewModelScope)
        api.getAnnouncements().onEach { _announcements.value = it }.launchIn(viewModelScope)
    }

    // Fix 4: atomic update{}
    fun updateCount() { _count.update { it + 1 } }

    // Fix 5: add debounce + catch before flatMapLatest
    fun observeSearch(query: StateFlow<String>) {
        query
            .debounce(300)
            .flatMapLatest { q ->
                flow { emit(api.search(q)) }
                    .catch { emit(emptyList()) }
            }
            .onEach { _results.value = it }
            .launchIn(viewModelScope)
    }
}

Bug 1 — SharedFlow replay=1 for navigation events causes double-navigation on rotation. Fix: replay=0. Bug 2 — cold Flow exposed from ViewModel means each UI collector triggers a separate Room query. Fix: stateIn. Bug 3 — two collect calls in one coroutine are sequential, not concurrent; the second never starts because the first never ends. Fix: two separate launchIn calls. Bug 4 — non-atomic read-modify-write on StateFlow; concurrent updates can lose increments. Fix: update{}. Bug 5 — flatMapLatest without debounce fires on every keystroke; also no error handling means one failed search kills the entire operator chain. Fix: debounce + catch inside the inner flow.

This requires flatMapLatest to cancel in-flight requests, plus an intermediate Loading state emitted before the API call. The tricky part is emitting Loading inside the flatMapLatest inner flow so it's also cancelled when a newer query arrives — preventing stale loading spinners.

sealed class AutocompleteState {
    object Idle        : AutocompleteState()
    object Loading     : AutocompleteState()
    data class Results(val items: List<String>) : AutocompleteState()
    data class Error(val msg: String)      : AutocompleteState()
}

val autocomplete: StateFlow<AutocompleteState> = _query
    .debounce(250)
    .filter { it.length >= 2 }
    .distinctUntilChanged()
    .flatMapLatest { query ->
        flow {
            emit(AutocompleteState.Loading)    // ← inside inner flow
            val results = api.suggest(query)     // cancellable network call
            emit(AutocompleteState.Results(results))
        }.catch { e ->
            emit(AutocompleteState.Error(e.message ?: "Error"))
        }
    }
    .onStart { emit(AutocompleteState.Idle) }
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), AutocompleteState.Idle)

// When user types "an" then "and" within 250ms:
// "an" → debounce fires → Loading emitted → api.suggest("an") starts
// "and" arrives → flatMapLatest CANCELS "an" flow (Loading + network call)
// "and" → debounce fires → fresh Loading emitted → api.suggest("and") starts
// No stale Loading state visible to user for cancelled "an" request ✓

The critical insight is emitting Loading inside the inner flow { } block within flatMapLatest, not outside it. If you emitted Loading before flatMapLatest, it would persist even after the new query cancelled the old request — showing a spinner while the new query's debounce timer is still running. Emitting Loading inside the inner flow means it gets cancelled along with the network request when a newer query arrives. Each query gets its own independent Loading → Results/Error lifecycle.

Flow cancellation is cooperative — the producer must check for cancellation at suspension points. Every emit() call in a flow builder checks cancellation automatically. But if a flow builder has a tight loop with no suspension points, it won't respond to cancellation until a suspension point is reached.

// Flow cancellation is automatic at every emit() call
val countFlow = flow {
    for (i in 1..Int.MAX_VALUE) {
        emit(i)   // ← checks cancellation here, throws if cancelled
    }
}

// ⚠ Tight CPU loop — cancellation NOT checked between emissions
val badFlow = flow {
    var i = 0
    while (true) {
        i++
        // No emit(), no delay() — cancellation ignored!
        if (i % 1000 == 0) emit(i)  // only checked every 1000 iterations
    }
}

// ✅ Fix: add ensureActive() or currentCoroutineContext().ensureActive()
val goodFlow = flow {
    var i = 0
    while (true) {
        currentCoroutineContext().ensureActive()  // ← manual check
        i++
        if (i % 1000 == 0) emit(i)
    }
}

// onCompletion — runs on cancellation, completion, OR error
flow
    .onCompletion { cause ->
        if (cause is CancellationException) println("Cancelled")
        else if (cause != null) println("Error: $cause")
        else println("Completed normally")
    }
    .collect { ... }

// CancellationException must NOT be caught by catch{}
// Flow.catch{} automatically re-throws CancellationException
// Never catch CancellationException and swallow it — breaks structured concurrency

The catch operator automatically re-throws CancellationException — it will never swallow a cancellation signal. This is by design: if a flow is cancelled, the cancellation must propagate upward to properly unwind the coroutine hierarchy. If you write a custom flow operator using try-catch, you must also re-throw CancellationException explicitly, or you'll break structured concurrency and create coroutine leaks. The onCompletion operator is the correct place to detect whether a flow ended normally, via cancellation, or via error — the cause parameter distinguishes all three.

MVI maps perfectly onto Flow primitives: intents arrive as a SharedFlow, state is a StateFlow, side effects (one-time events) are a SharedFlow with replay=0. The ViewModel processes intents, updates state, and fires side effects — all through typed Flow streams.

// MVI types
sealed class LoginIntent {
    data class EmailChanged(val email: String) : LoginIntent()
    data class PasswordChanged(val pw: String) : LoginIntent()
    object Submit : LoginIntent()
}

data class LoginState(val email: String = "", val isLoading: Boolean = false, val error: String? = null)

sealed class LoginEffect {
    object NavigateToHome : LoginEffect()
    data class ShowSnackbar(val msg: String) : LoginEffect()
}

class LoginViewModel : ViewModel() {
    private val _intents = MutableSharedFlow<LoginIntent>()

    private val _state = MutableStateFlow(LoginState())
    val state = _state.asStateFlow()

    private val _effects = MutableSharedFlow<LoginEffect>()  // replay=0
    val effects = _effects.asSharedFlow()

    init {
        _intents
            .onEach { processIntent(it) }
            .launchIn(viewModelScope)
    }

    fun send(intent: LoginIntent) {
        viewModelScope.launch { _intents.emit(intent) }
    }

    private suspend fun processIntent(intent: LoginIntent) {
        when (intent) {
            is LoginIntent.EmailChanged -> _state.update { it.copy(email = intent.email) }
            is LoginIntent.Submit -> {
                _state.update { it.copy(isLoading = true) }
                val result = repo.login(_state.value.email)
                if (result.isSuccess) _effects.emit(LoginEffect.NavigateToHome)
                else _effects.emit(LoginEffect.ShowSnackbar("Login failed"))
                _state.update { it.copy(isLoading = false) }
            }
            else -> {}
        }
    }
}

The separation of state (StateFlow — current screen state, always available, survives rotation) and effects (SharedFlow replay=0 — one-shot, fire-and-forget) is the critical MVI Flow design decision. Navigation and snackbar messages are effects: they should fire once regardless of rotation. The loading indicator is state: if the screen rotates mid-login, it should immediately show the loading spinner based on the replayed StateFlow value.

These lifecycle operators let you inject behaviour at key points in a flow's lifetime without modifying the source. onStart fires before the first emission. onCompletion fires when the flow ends (normally, cancelled, or errored). onEmpty fires if the flow completes without emitting any values.

// onStart — emit a default/loading value before upstream begins
dao.getProductsFlow()
    .onStart { emit(emptyList()) }  // instant empty state while DB loads
    .collect { render(it) }

// onStart for showing a loading state
repo.fetchFlow()
    .map { UiState.Success(it) }
    .onStart { emit(UiState.Loading) }   // Loading shown immediately
    .catch { emit(UiState.Error(it)) }

// onCompletion — cleanup, analytics, logging
uploadFlow
    .onCompletion { cause ->
        when {
            cause == null                      -> analytics.logUploadComplete()
            cause is CancellationException  -> analytics.logUploadCancelled()
            else                               -> analytics.logUploadFailed(cause)
        }
        hideProgressBar()
    }
    .collect { progress -> updateProgressBar(progress) }

// onEmpty — handle zero-result cases explicitly
repo.searchFlow(query)
    .onEmpty { emit(SearchState.NoResults) }
    .collect { render(it) }
// Without onEmpty, a flow that completes without emitting leaves UI stale

// onStart fires even if the flow immediately throws — before any processing
// onCompletion fires BEFORE catch — you see the original error in cause

Execution order matters: onStart runs before any upstream operator. onCompletion runs after the flow terminates (before catch if there's an exception). onEmpty runs only if the flow completed normally with zero emissions — it doesn't fire on errors or cancellation. A common production pattern is chaining all three: onStart { emit(Loading) }.onEmpty { emit(NoResults) }.catch { emit(Error(it)) } — this makes every terminal state explicit and prevents the UI from being stuck in a stale state.

A countdown timer as a Flow uses a flow { } builder with delay() for each tick. Pause/resume maps directly to StateFlow-controlled flatMapLatest — when paused, switch to an empty flow; when running, switch to the tick flow.

class TimerViewModel : ViewModel() {
    private val _isRunning = MutableStateFlow(false)

    private fun countdownFlow(totalSeconds: Int): Flow<Int> = flow {
        for (remaining in totalSeconds downTo 0) {
            emit(remaining)
            if (remaining > 0) delay(1_000)
        }
    }

    // Pause/resume by switching between active and empty flow
    val timerState: StateFlow<Int> = _isRunning
        .flatMapLatest { running ->
            if (running) countdownFlow(60) else flowOf()  // empty flow when paused
        }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), 60)

    fun start()  { _isRunning.value = true  }
    fun pause()  { _isRunning.value = false }
    fun toggle() { _isRunning.update { !it } }
}

// Preserving elapsed time across pause:
private var _elapsed = 0
val resumableTimer = _isRunning
    .flatMapLatest { running ->
        if (!running) flowOf()
        else flow {
            var t = _elapsed
            while (t <= 60) {
                emit(t++)
                _elapsed = t         // save position before delay
                delay(1_000)
            }
        }
    }

The flowOf() (empty flow) when paused is the key: flatMapLatest cancels the countdown flow when _isRunning becomes false, and the empty flow emits nothing. When _isRunning becomes true again, flatMapLatest starts a new countdown flow from the current value. For a resumable timer (pick up where left off), store the elapsed position in a ViewModel property before each delay — when the flow is cancelled (pause) the position is saved, and the next start picks it up. The StateFlow's stateIn ensures the last tick value is immediately available to the UI on resubscription.

fold is a terminal operator that reduces the entire flow to a single value after it completes. scan (also called runningFold) is an intermediate operator that emits each accumulated intermediate value as the flow progresses. For infinite flows or state accumulation, scan is the correct tool — fold would never complete.

// fold — terminal, waits for flow to complete
val total = flowOf(1, 2, 3).fold(0) { acc, v -> acc + v }
// total = 6, but nothing emitted until flow ends

// scan — intermediate, emits each step
flowOf(1, 2, 3).scan(0) { acc, v -> acc + v }
// Emits: 0, 1, 3, 6  (initial value, then each accumulated result)

// Production use case 1: accumulating user actions into state
data class CartState(val items: List<Item> = emptyList(), val total: Double = 0.0)

cartActionFlow                          // Flow<CartAction>
    .scan(CartState()) { state, action ->
        when (action) {
            is CartAction.AddItem    -> state.copy(items = state.items + action.item)
            is CartAction.RemoveItem -> state.copy(items = state.items - action.item)
            is CartAction.Clear      -> CartState()
        }
    }
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), CartState())
    // Each action produces a new complete state — pure Redux/MVI state machine

// Production use case 2: running total in analytics
eventFlow.scan(0) { count, _ -> count + 1 }   // live event counter

// runningFold is identical to scan — just an alias added in Kotlin 1.9

scan is the natural choice for implementing a state reducer pattern (Redux/MVI) on top of Flow. A stream of actions is scanned with an initial state and a reducer function — each action produces a new state. This pattern makes every state transition explicit and testable: given this sequence of actions, assert this sequence of states. The initial value in scan (the seed) is always emitted first, which conveniently provides the initial UI state without a separate stateIn initialValue.

Rate limiting (throttle) differs from debounce: debounce waits for a pause in events, throttle enforces a minimum interval between processed events regardless of pauses. Kotlin Flow has sample for periodic sampling and debounce for silence-triggered emission. For strict rate limiting, combine conflate with a timed delay.

// sample — emit the most recent value every N ms (periodic sampling)
sensorFlow.sample(2_000).collect { sendToServer(it) }
// Emits at most once per 2s, taking the latest value in each window
// If no emission in window → nothing emitted (not guaranteed 2s intervals)

// throttleFirst — emit first value, then wait before accepting next
// Kotlin Flow has no built-in throttleFirst — implement with conflate + delay
fun <T> Flow<T>.throttleFirst(windowMs: Long): Flow<T> = flow {
    var lastEmit = 0L
    collect { value ->
        val now = System.currentTimeMillis()
        if (now - lastEmit >= windowMs) {
            lastEmit = now
            emit(value)
        }
    }
}

// Rate-limited API calls: conflate keeps latest, delay enforces minimum gap
fun rateLimitedApiCalls(events: Flow<Event>): Flow<Result> = flow {
    events.conflate().collect { event ->    // only process latest pending
        emit(api.process(event))              // actual API call
        delay(2_000)                         // enforce 2s minimum gap
    }
}
// While processing event N, new events accumulate in conflate buffer
// Only the LATEST pending event is processed when delay expires
// Guarantees: ≤1 call per 2s, always processes most recent event

debounce vs sample vs throttleFirst: Debounce fires after a silence — typing speed determines timing. Sample fires on a fixed clock — always every N ms regardless of events. ThrottleFirst fires immediately on first event, then ignores subsequent events for N ms — good for button clicks to prevent double-taps. The conflate + delay pattern enforces a minimum gap while always keeping the latest: if 10 events arrive in 2 seconds, one API call is made for the most recent event, and another 2 seconds must pass before the next — the oldest 9 events are dropped.

Flow leaks happen when a collection coroutine outlives its intended scope — typically caused by collecting in a scope that doesn't end when the UI is destroyed, or by a hot Flow keeping its upstream alive when it should have stopped.

// Leak 1: collecting in GlobalScope or a scope that outlives the screen
class LeakyFragment : Fragment() {
    override fun onViewCreated(...) {
        // ❌ GlobalScope — never cancelled when Fragment is destroyed
        GlobalScope.launch { viewModel.flow.collect { render(it) } }

        // ❌ lifecycleScope without repeatOnLifecycle — keeps collecting when backgrounded
        lifecycleScope.launch { viewModel.flow.collect { render(it) } }

        // ✅ viewLifecycleOwner.lifecycleScope + repeatOnLifecycle — correctly scoped
        viewLifecycleOwner.lifecycleScope.launch {
            viewLifecycleOwner.lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
                viewModel.flow.collect { render(it) }
            }
        }
    }
}

// Leak 2: hot SharedFlow keeping upstream alive with Eagerly
val leakyFlow = callbackRepo.sensorFlow()
    .shareIn(viewModelScope, SharingStarted.Eagerly)
// ↑ Starts immediately and NEVER stops — GPS listener runs forever
// ✅ Fix: use WhileSubscribed to stop when no active collectors
val fixedFlow = callbackRepo.sensorFlow()
    .shareIn(viewModelScope, SharingStarted.WhileSubscribed(5_000))

// Leak 3: callbackFlow without awaitClose — callback registered but never unregistered
fun brokenFlow() = callbackFlow<Event> {
    api.setCallback { trySend(it) }
    // ❌ missing awaitClose — Flow completes, but callback stays registered
}

// Diagnosis with LeakCanary:
// LeakCanary reports: Fragment → anonymous coroutine → Flow → Repository
// The leak path shows which scope outlived its owner
// Add -Dkotlinx.coroutines.debug to see coroutine names in stack traces

Flow leaks are sneaky because they don't always appear in LeakCanary immediately — the coroutine may hold references to lambdas that capture the Fragment, keeping it alive after destruction. The three prevention rules: (1) always collect in viewLifecycleOwner.lifecycleScope (not lifecycleScope) with repeatOnLifecycle; (2) use WhileSubscribed not Eagerly for hot flows; (3) always include awaitClose in callbackFlow. Adding -Dkotlinx.coroutines.debug in your run configuration names all coroutines in stack traces, making it much easier to identify leaked collections.

Flow-based pagination can be implemented manually with a MutableStateFlow of page number, or via Jetpack Paging 3 which wraps its PagingData in a Flow. For custom pagination, the pattern is: a trigger flow (scroll events or load-more button) drives flatMapLatest/flatMapConcat to fetch successive pages and accumulate them.

// Manual cursor-based pagination with Flow
class FeedViewModel : ViewModel() {
    private val _loadTrigger = MutableSharedFlow<Unit>(extraBufferCapacity = 1)
    private var cursor: String? = null

    val items: StateFlow<List<FeedItem>> = _loadTrigger
        .onStart { emit(Unit) }           // load first page immediately
        .flatMapConcat {                    // sequential pages (no overlap)
            flow { emit(api.getFeed(cursor)) }
        }
        .scan(emptyList()) { acc, page ->
            cursor = page.nextCursor      // save cursor for next page
            acc + page.items             // accumulate all pages
        }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), emptyList())

    fun loadNextPage() { _loadTrigger.tryEmit(Unit) }
}

// Paging 3 — recommended for complex pagination
class FeedPagingSource : PagingSource<String, FeedItem>() {
    override suspend fun load(params: LoadParams<String>): LoadResult<String, FeedItem> {
        return try {
            val page = api.getFeed(params.key)
            LoadResult.Page(page.items, prevKey = null, nextKey = page.nextCursor)
        } catch (e: Exception) { LoadResult.Error(e) }
    }
    override fun getRefreshKey(...) = null
}

// ViewModel with Paging 3
val feedPager: Flow<PagingData<FeedItem>> = Pager(PagingConfig(pageSize = 20)) {
    FeedPagingSource()
}.flow.cachedIn(viewModelScope)  // cache survives rotation

The scan operator is the key to accumulating pages: each new page is appended to the existing list without requiring external mutable state. flatMapConcat (not flatMapMerge) ensures pages load sequentially — if the user triggers loadNextPage twice rapidly, the second load waits for the first to complete, preventing duplicate pages. For complex scenarios (network + database, placeholders, retry, prepend/append indicators), Jetpack Paging 3 handles all of this and should be the default choice. The cachedIn(viewModelScope) call is critical — without it, the PagingData Flow is recreated on every rotation, losing the loaded pages.

Feature flags are app-global state that can change at any time. The ideal architecture is a @Singleton repository holding a StateFlow<FeatureFlags>, updated whenever remote config changes. Each screen observes the relevant flags via map + distinctUntilChanged to receive only changes that affect them.

// Singleton feature flag repository
class FeatureFlagRepository @Inject constructor(
    private val remoteConfig: FirebaseRemoteConfig
) {
    private val _flags = MutableStateFlow(loadFlags())
    val flags: StateFlow<FeatureFlags> = _flags.asStateFlow()

    // Refresh flags periodically or on app foreground
    fun refresh() {
        remoteConfig.fetchAndActivate().addOnSuccessListener {
            _flags.update { loadFlags() }  // atomic update → all observers notified
        }
    }

    private fun loadFlags() = FeatureFlags(
        newCheckout = remoteConfig.getBoolean("new_checkout"),
        darkModeV2  = remoteConfig.getBoolean("dark_mode_v2")
    )

    // Per-flag flows — observers only notified when THEIR flag changes
    val newCheckoutEnabled: Flow<Boolean> = flags
        .map { it.newCheckout }
        .distinctUntilChanged()  // no recomposition if other flags change
}

// ViewModel subscribing to a specific flag
class CheckoutViewModel @Inject constructor(
    private val flags: FeatureFlagRepository
) : ViewModel() {

    val useNewCheckout: StateFlow<Boolean> = flags.newCheckoutEnabled
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), false)
}

// combine flags with data for a conditional experience
val checkoutUi = combine(flags.newCheckoutEnabled, cartFlow) { newUi, cart ->
    if (newUi) CheckoutUi.newDesign(cart) else CheckoutUi.legacyDesign(cart)
}

The distinctUntilChanged after map is critical: without it, every flag refresh (even one that only changes an unrelated flag) would trigger all flag observers. With distinctUntilChanged on the individual flag flow, a screen observing newCheckoutEnabled only reacts when that specific flag changes — not when any other flag is refreshed. The combine pattern then lets you build conditional UI that immediately reflects both flag changes and data changes.

map transforms each emission 1-to-1 — one input, one output. transform is the general-purpose operator that can emit zero, one, or multiple values per input element, and can itself use Flow operators. It's the building block that map, filter, and flatMap are implemented on top of.

// map — 1-to-1 transformation
flow.map { user -> user.name }  // one name per user

// transform — emit 0, 1, or N values per input

// filter using transform (emit 0 or 1)
flow.transform { user ->
    if (user.isActive) emit(user)   // only active users
}

// expand each input to multiple outputs (1-to-N)
orderFlow.transform { order ->
    order.items.forEach { item -> emit(item) }  // flatten order items
}

// emit intermediate steps (1-to-N for loading states)
requestFlow.transform { request ->
    emit(State.Loading)                 // before processing
    emit(State.Success(api.call(request)))  // after result
}

// transformLatest — like flatMapLatest but with transform ergonomics
queryFlow.transformLatest { query ->
    emit(State.Loading)
    delay(300)  // debounce inside
    emit(State.Success(api.search(query)))
}
// transformLatest cancels previous block when new value arrives
// Combines debounce + flatMapLatest + emit in one clean block

transformLatest is particularly powerful: it cancels the previous transform block when a new value arrives, similar to flatMapLatest but with simpler syntax for single-output transforms. A common pattern: emit Loading, then delay (acting as a debounce), then fetch and emit result — all in one transformLatest block. If a new query arrives before the delay expires, the entire block (including the Loading emission and the delay) is cancelled and restarted. This produces cleaner loading state management than debounce + flatMapLatest as separate operators.

Undo/redo is a state history problem. The cleanest Flow-based approach uses a stack of past states stored in a StateFlow, pushing new states on every action and popping on undo. scan is perfect for building the history as actions flow in.

data class HistoryState<T>(
    val past: List<T> = emptyList(),   // states before current
    val current: T,
    val future: List<T> = emptyList()  // states for redo
) {
    val canUndo = past.isNotEmpty()
    val canRedo = future.isNotEmpty()
}

sealed class HistoryAction<T> {
    data class Apply<T>(val newState: T) : HistoryAction<T>()
    class  Undo<T> : HistoryAction<T>()
    class  Redo<T> : HistoryAction<T>()
}

class EditorViewModel(initial: DocumentState) : ViewModel() {
    private val _actions = MutableSharedFlow<HistoryAction<DocumentState>>()

    val history = _actions
        .scan(HistoryState(current = initial)) { h, action ->
            when (action) {
                is HistoryAction.Apply -> HistoryState(
                    past    = h.past + h.current,
                    current = action.newState,
                    future  = emptyList()     // new action clears redo stack
                )
                is HistoryAction.Undo -> if (h.canUndo) HistoryState(
                    past    = h.past.dropLast(1),
                    current = h.past.last(),
                    future  = listOf(h.current) + h.future
                ) else h
                is HistoryAction.Redo -> if (h.canRedo) HistoryState(
                    past    = h.past + h.current,
                    current = h.future.first(),
                    future  = h.future.drop(1)
                ) else h
            }
        }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), HistoryState(current = initial))

    fun apply(state: DocumentState) { viewModelScope.launch { _actions.emit(HistoryAction.Apply(state)) } }
    fun undo() { viewModelScope.launch { _actions.emit(HistoryAction.Undo()) } }
    fun redo() { viewModelScope.launch { _actions.emit(HistoryAction.Redo()) } }
}

This design is a pure state machine: actions flow in, history states flow out. Every state transition is an immutable data transformation — no mutation, no side effects. Testing is trivial: emit a sequence of Apply/Undo/Redo actions and assert the emitted HistoryState sequence with Turbine. The scan seed (initial HistoryState) is immediately emitted and serves as the starting UI state. Applying a new action correctly clears the redo stack — exactly the semantics users expect when they type after undoing.

With a cold Flow, each collector independently executes the upstream — three collectors mean three network calls. This is wasteful, can cause inconsistent data (each call returns slightly different data), and overloads the server. The fix is converting to a hot flow with shareIn or stateIn.

// ❌ Cold Flow — 3 collectors = 3 network calls
fun userFlow(): Flow<User> = flow { emit(api.getUser()) }

// Fragment A collects → network call 1
// Fragment B collects → network call 2 (possibly different result!)
// BottomSheet collects → network call 3

// ✅ Fix A: stateIn in ViewModel — one upstream, N collectors
val user: StateFlow<User?> = userFlow()
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), null)
// ONE network call, result shared to all collectors via multicast

// ✅ Fix B: shareIn with replay=1 — hot SharedFlow, result cached for late subscribers
val sharedUser: SharedFlow<User> = userFlow()
    .shareIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), replay = 1)
// Replay=1 ensures late subscribers (e.g. BottomSheet opening after load) get the value

// How multicasting works internally:
// shareIn/stateIn subscribe once to the upstream cold Flow
// Emissions are fanned out to all active downstream collectors simultaneously
// This is a single-producer / multiple-consumer pattern

// ⚠ Trap: shareIn in the repository without a scope — who owns it?
// Repository flows should be COLD — convert to hot in the ViewModel with viewModelScope
// OR use a application-scoped flow for app-wide data (inject via Hilt @Singleton)

The multicast danger is subtle: three calls to the same cold Flow at roughly the same time can return inconsistent data — call 1 returns user with version A, call 3 returns version B after a server-side update. The UI components then render different states simultaneously. Beyond correctness, it's a network efficiency issue — one network call should serve all three components. The ViewModel is the right place to convert to hot because viewModelScope ties the upstream lifecycle to the ViewModel lifetime, not to any specific UI component.

Flow is pure Kotlin — it compiles for all KMP targets (Android, iOS, Desktop, Web) without modification. The challenge is consuming Flow from Swift on iOS, where the Kotlin coroutine runtime is not natively accessible. The standard solutions are either wrapping Flow in a Swift-friendly callback, or using the SKIE or KMP-NativeCoroutines library.

// Shared KMP code (commonMain) — pure Flow, works everywhere
class UserRepository {
    val userFlow: StateFlow<User?> = MutableStateFlow(null)
    fun usersFlow(): Flow<List<User>> = dao.getAllUsers()
}

// Android — collect as normal
viewModel.users.collect { render(it) }

// iOS challenge: Swift cannot directly await Kotlin Flows
// Swift async/await ≠ Kotlin coroutines natively

// Option 1: KMP-NativeCoroutines library — wraps Flow as Swift AsyncSequence
// In Kotlin (shared): annotate with @NativeCoroutines
class UserViewModel : ViewModel() {
    @NativeCoroutines
    val usersFlow: Flow<List<User>> = repo.usersFlow()
}
// In Swift: for await user in userViewModel.usersFlow { ... }

// Option 2: SKIE library — auto-generates Swift-friendly wrappers
// Automatically converts Flow<T> to AnyFlow<T> callable from Swift

// Option 3: Manual wrapper (no dependency)
fun <T> watchFlow(flow: Flow<T>, onEach: (T) -> Unit, onError: (Throwable) -> Unit): Cancellable {
    val job = CoroutineScope(Dispatchers.Main).launch {
        flow.catch { onError(it) }.collect { onEach(it) }
    }
    return object : Cancellable { override fun cancel() = job.cancel() }
}
// Swift: let cancel = watchFlow(flow: repo.usersFlow(), onEach: { ... }, onError: { ... })

StateFlow on iOS: StateFlow.value is accessible from Swift synchronously — you can read the current state without needing coroutines. But observing StateFlow updates from Swift requires one of the above approaches. The SKIE library is the most ergonomic — it generates Swift-compatible wrappers automatically at compile time with no annotations needed. KMP-NativeCoroutines requires annotation but gives more control. The manual wrapper is dependency-free but requires more boilerplate.

Stale-while-revalidate (SWR) is the HTTP caching pattern adapted for mobile: serve stale data immediately for zero perceived latency, then fetch fresh data and silently update. With Room + Flow, this is nearly automatic — Room emits cached data first, then re-emits after the background refresh writes to the database.

// Repository — SWR pattern
class NewsRepository(
    private val dao: NewsDao,
    private val api: NewsApi
) {
    // Room Flow: emits cached data immediately, then on every DB update
    fun newsFlow(): Flow<List<Article>> = dao.getArticlesFlow()
        .onStart {
            // Trigger background refresh (doesn't block the Room Flow)
            refreshInBackground()
        }

    private fun refreshInBackground() {
        // Fire-and-forget: refresh happens without blocking collectors
        // Room notifies all active collectors when insert completes
        externalScope.launch(Dispatchers.IO) {
            try {
                val fresh = api.getLatestArticles()
                dao.insertAll(fresh)  // triggers Room Flow to re-emit ✓
            } catch (e: IOException) {
                // Network unavailable — silently use cache, no error UI
            }
        }
    }
}

// ViewModel — staleness indicator
data class NewsUiState(val articles: List<Article>, val isRefreshing: Boolean = false)

val uiState = combine(
    repo.newsFlow(),
    _isRefreshing
) { articles, refreshing ->
    NewsUiState(articles, refreshing)
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), NewsUiState(emptyList(), true))

// Timeline (user opens app):
// T=0ms:   Room emits cached articles instantly → UI shows immediately
// T=0ms:   Background network refresh starts
// T=800ms: Fresh articles stored in Room → Room re-emits → UI updates silently

The key is that onStart triggers the background refresh without blocking the Room Flow — both happen concurrently. The Room Flow emits the cached data first (zero latency for the user), and as soon as the background refresh writes to the database, Room automatically emits the updated data. The UI sees two emissions for one screen load: first cached, then fresh. Adding a isRefreshing boolean via combine lets you show a subtle pull-to-refresh indicator without blocking the content, which is much better UX than showing a loading spinner over empty content.

Flow.collect is a suspend function — it should never block the main thread. ANRs from Flow collection are almost always caused by heavy work inside the collect block (synchronous operations), or Flow operators running on the wrong dispatcher.

// ANR Cause 1: heavy work inside collect on Main thread
flow.collect { data ->
    // ❌ These block Main thread → ANR if > 5s
    val processed = heavyJsonProcessing(data)  // CPU-intensive on Main
    writeToDisk(data)                          // IO on Main
    updateBitmap(data)                         // memory-intensive on Main
}

// ✅ Fix: move heavy work off Main thread
flow
    .map { heavyJsonProcessing(it) }    // transform on Default
    .flowOn(Dispatchers.Default)        // map runs on Default, collect stays on Main
    .collect { render(it) }              // only UI work on Main

// ANR Cause 2: runBlocking inside a flow operator
flow.map { runBlocking { api.fetch(it) } }  // ❌ blocks Main coroutine thread
// Fix: use suspend inside map, no runBlocking
flow.map { api.fetch(it) }                  // ✅ suspends, doesn't block

// ANR Cause 3: Flow collected without lifecycle — runs during onPause/onStop
// During ANR: Fragment paused (e.g. dialog shown), flow still processing
// ✅ Fix: repeatOnLifecycle(STARTED) stops collection during pause

// Diagnosis steps:
// 1. ANR traces (traces.txt): look for "main" thread in BLOCKED state
// 2. Android Vitals: ANR cluster → stack trace shows blocking call
// 3. StrictMode: detects disk I/O and network on main thread at dev time
// 4. Profiler: CPU usage spike on main thread correlating with Flow emission

ANRs from Flow are insidious because the Flow API looks async — developers assume collect won't block. But the collect block itself runs on whichever coroutine calls it — if that's the Main dispatcher, any blocking work in the collect lambda will block the main thread. The correct rule: the collect lambda should only contain UI work (binding data to views, updating StateFlow). Any processing — JSON parsing, bitmap decoding, filtering — belongs in upstream operators with flowOn(Dispatchers.Default).

produceIn converts a Flow to a ReceiveChannel, and consumeAsFlow converts a Channel back to a Flow. These bridging operators let you use both primitives in the same pipeline — useful when integrating legacy Channel-based code with modern Flow-based code.

// produceIn — Flow → ReceiveChannel
val channel: ReceiveChannel<Int> = myFlow.produceIn(scope)
// channel is now a hot Channel backed by the Flow
// Use when: legacy code expects a ReceiveChannel, or for select{} patterns

// consumeAsFlow — Channel → Flow
val channel = Channel<Event>()
val flow: Flow<Event> = channel.consumeAsFlow()
// consumeAsFlow: single consumer only — second collect throws exception

// receiveAsFlow — multiple consumers allowed
val sharedFlow: Flow<Event> = channel.receiveAsFlow()
// Each item delivered to ONE collector (round-robin if multiple)

// select{} — receive from multiple channels
val ch1 = flow1.produceIn(scope)
val ch2 = flow2.produceIn(scope)

scope.launch {
    while (true) {
        select<Unit> {
            ch1.onReceive { handleFirst(it) }
            ch2.onReceive { handleSecond(it) }
        }
    }
}
// select{} processes whichever channel has data first
// The closest Kotlin equivalent to Go's select statement

// Modern recommendation: use combine() instead of select{} for most cases
combine(flow1, flow2) { a, b -> handle(a, b) }  // cleaner, no Channel conversion needed

consumeAsFlow() vs receiveAsFlow() is an important distinction: consumeAsFlow creates a flow that can only be collected once (second collect throws), enforcing single-consumer semantics. receiveAsFlow allows multiple collectors but distributes items — collector A gets item 1, collector B gets item 2 (not both get all items). For broadcasting to multiple collectors use SharedFlow, not receiveAsFlow. In modern code, direct Channel interaction is rare — use Flow operators like combine, merge, and flatMapLatest instead of convert-to-channel-and-select.

Random test failures indicate a race condition — the emission order depends on coroutine scheduling, which varies between runs. StateFlow guarantees that all collectors see a consistent final state, but the intermediate states are non-deterministic when updated from concurrent coroutines. The fix is ensuring state updates are serialised.

// ❌ Flaky test: concurrent updates to StateFlow
@Test
fun `state transitions are Loading then Success`() = runTest {
    val state = MutableStateFlow<UiState>(UiState.Idle)

    // Two coroutines on Default dispatcher — order non-deterministic
    launch(Dispatchers.Default) { state.value = UiState.Loading }
    launch(Dispatchers.Default) { state.value = UiState.Success(data) }

    state.test {
        assertEquals(UiState.Loading, awaitItem())  // may fail — Success could come first
    }
}

// ✅ Fix 1: use UnconfinedTestDispatcher — makes coroutines run eagerly and deterministically
@Test
fun `state transitions are deterministic`() = runTest(UnconfinedTestDispatcher()) {
    // All coroutines run synchronously in call order in UnconfinedTestDispatcher
}

// ✅ Fix 2: serialise state updates — single coroutine updates state sequentially
fun loadData() = viewModelScope.launch {        // single coroutine, sequential
    _state.value = UiState.Loading              // guaranteed before Success
    val data = withContext(Dispatchers.IO) { api.fetch() }
    _state.value = UiState.Success(data)
}

// ✅ Fix 3: use StateFlow.update{} for atomic transforms
// update{} is atomic CAS — won't interleave with other updates
_state.update { it.copy(isLoading = true) }

// Test setup best practice:
@Before
fun setup() {
    Dispatchers.setMain(UnconfinedTestDispatcher())  // deterministic test dispatching
}
@After
fun tearDown() { Dispatchers.resetMain() }

The root cause of flaky Flow tests is almost always concurrency: coroutines on real thread-pool dispatchers (Default, IO) run truly parallel, making the emission order non-deterministic. UnconfinedTestDispatcher makes coroutines run eagerly in the calling thread's order, making tests deterministic. StandardTestDispatcher queues all coroutines and only runs them when you call advanceUntilIdle() or advanceTimeBy() — giving you complete control over execution order. For state transitions that must be sequential, the architectural fix is keeping updates in a single coroutine (sequential launch) rather than multiple concurrent coroutines.

Kotlin 2024-25 brought several meaningful Flow improvements: new operators, collectAsStateWithLifecycle becoming the standard in Compose, and the K2 compiler generating more efficient state machine bytecode for Flow operator chains.

// 1. Flow.toList(), toSet() — collect into collection (Kotlin 1.9+)
val items: List<Item> = finiteFlow.toList()

// 2. Flow.first(predicate) — suspends until first matching emission
val ready = statusFlow.first { it == Status.Ready }

// 3. StateFlow.subscriptionCount — debug how many collectors are active
stateFlow.subscriptionCount.collect { count -> log("Collectors: $count") }

// 4. collectAsStateWithLifecycle now universal default in Compose
// Google's architecture samples, codelabs, and Now in Android all updated
// Use it everywhere — not just when you're worried about lifecycle

// 5. K2 compiler (Kotlin 2.0+) — better Flow operator chain bytecode
// map.filter.flatMapLatest chains compile to more efficient state machines
// Fewer allocations, smaller APK, faster coroutine resumption

// 6. Turbine 1.x — multi-flow testing with turbineScope
turbineScope {
    val flowA = viewModel.stateA.testIn(this)
    val flowB = viewModel.stateB.testIn(this)
    flowA.awaitItem()  // assert independently
    flowB.awaitItem()
    flowA.cancel(); flowB.cancel()
}

// 2025 production standard checklist:
// ✅ collectAsStateWithLifecycle (not collectAsState) in Compose
// ✅ stateIn(WhileSubscribed(5000)) — not Eagerly or Lazily
// ✅ update{} for all StateFlow mutations (not .value =)
// ✅ Turbine 1.x for all Flow tests
// ✅ SharedFlow(replay=0) for navigation events
// ✅ callbackFlow + awaitClose for all callback-to-Flow bridges

The most impactful 2025 shift is cultural: collectAsStateWithLifecycle is now the only recommended way to collect Flow in Compose — not "the lifecycle-aware option." Google updated every sample, codelab, and architecture guide. The K2 compiler improvements benefit Flow chains silently — faster startup, smaller APK — requiring no code changes. The Turbine 1.x turbineScope API enables testing multiple flows simultaneously in a single test, which was previously awkward and required complex coroutine coordination.

This is the capstone system design question. A real e-commerce app needs: live product listing (Room Flow + WebSocket), cart state (singleton StateFlow), and checkout one-shot events (SharedFlow). Each domain gets the right Flow primitive for its access pattern.

// 1. PRODUCT LISTING — offline-first with live stock updates
class ProductRepository {
    fun productsFlow(category: String): Flow<List<Product>> =
        dao.getProductsFlow(category).onStart { backgroundRefresh(category) }

    val stockUpdates = callbackFlow<StockUpdate> {
        val ws = ws.connect("wss://stock.api/live") { trySend(it) }
        awaitClose { ws.close() }
    }.retryWhen { _, attempt -> delay(2000L * attempt); true }
}

// 2. CART — singleton, app-global, instant propagation
class CartRepository {   // @Singleton via Hilt
    private val _cart = MutableStateFlow(Cart.empty())
    val cartFlow = _cart.asStateFlow()
    fun addItem(item: Item) { _cart.update { it.add(item) } }
}

// 3. PRODUCT VIEWMODEL — combine products + cart badge
class ProductListViewModel : ViewModel() {
    val uiState = combine(
        productRepo.productsFlow(category),
        cartRepo.cartFlow.map { it.itemCount }.distinctUntilChanged()
    ) { products, cartCount ->
        ProductListUiState(products, cartCount)
    }.onStart { emit(ProductListUiState.loading()) }
     .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), ProductListUiState.loading())
}

// 4. CHECKOUT VIEWMODEL — state + one-shot events
class CheckoutViewModel : ViewModel() {
    private val _state = MutableStateFlow(CheckoutState())
    val state = _state.asStateFlow()  // survives rotation

    private val _events = MutableSharedFlow<CheckoutEvent>()  // replay=0: fire-once
    val events = _events.asSharedFlow()

    fun pay() = viewModelScope.launch {
        _state.update { it.copy(isProcessing = true) }
        when (val result = paymentRepo.charge()) {
            is PayResult.Success -> _events.emit(CheckoutEvent.GoToConfirmation)
            is PayResult.Failure -> _events.emit(CheckoutEvent.ShowError(result.msg))
        }
        _state.update { it.copy(isProcessing = false) }
    }
}

// Flow type decision table:
// Room data (products)  → cold Flow → stateIn in ViewModel
// App-wide state (cart) → @Singleton MutableStateFlow
// Streaming (WebSocket) → callbackFlow + retryWhen + shareIn
// One-shot UI events    → MutableSharedFlow(replay=0)

The decision table at the bottom is the core answer. Each domain maps to the correct Flow primitive: Room Flow for persistent data (cold, DB-backed, auto-updates on insert), Singleton StateFlow for app-global mutable state (hot, always readable, instant propagation to all screens), callbackFlow for streaming connections (lifecycle-managed via awaitClose, resilient via retryWhen), and SharedFlow replay=0 for one-shot events (no rotation replay bug). The combine in ProductListViewModel shows how to merge two independent streams into a single coherent UiState — the cart badge updates instantly when items are added from any screen, without any pub/sub boilerplate.

zip pairs values positionally one-to-one — it waits for both flows to emit before producing output. combine emits whenever any source emits, using the latest value from all others. The choice determines whether you're synchronising two streams or reacting to independent updates.

// zip — positional pairing, waits for BOTH to emit
val nums = flowOf(1, 2, 3)
val strs = flowOf("A", "B")

nums.zip(strs) { n, s -> "$n$s" }.collect { println(it) }
// Prints: "1A", "2B"  — stops when shorter flow ends
// "3" is NEVER emitted — no partner from strs

// combine — latest from each, emits on ANY update
val price = flowOf(10, 20, 30)
val qty   = flowOf(2, 5)

combine(price, qty) { p, q -> p * q }.collect { println(it) }
// Possible: 20, 40, 100, 150 — combines latest each time either emits

// Real Android examples:

// zip: request/response pairing (upload file → get signed URL)
fileIds.zip(uploadResults) { id, url -> Pair(id, url) }

// combine: live dashboard with independent refresh rates
combine(
    stockRepo.priceFlow(),     // updates every 1s
    portfolioRepo.holdingsFlow() // updates on trade
) { price, holdings -> holdings * price }

// ⚠ zip blocks if one flow is slower — the fast flow's items queue up
// ⚠ combine blocks until ALL sources emit at least once — use onStart{emit(default)}

When zip blocks: if flowA emits every 10ms but flowB emits every 1s, zip emits once per second — flowA's values accumulate in a buffer waiting for flowB partners. This is intentional for synchronised pairing (request/response, test assertion matching) but catastrophic if you accidentally use zip where combine was intended. The termination behaviour is also different: zip ends when the shorter flow ends; combine ends only when all flows end.

This is a systematic diagnosis question. The most common causes are: StateFlow equality filtering skipping updates, collecting on the wrong scope, the flow never being collected, or the UI observing a different instance than the one being updated.

// Debugging checklist — work top to bottom

// 1. Is the flow actually emitting? Add onEach logging
viewModel.uiState
    .onEach { Log.d("FLOW", "Emission: $it") }  // before collect
    .collect { render(it) }

// 2. StateFlow equality filter — same value → no emission
data class State(val items: MutableList<Item>)  // ❌ mutable
_state.value = _state.value  // same reference → skipped
// Fix: use immutable data + copy()
_state.update { it.copy(items = it.items + newItem) }

// 3. Collected on wrong scope → stops on lifecycle change
// ❌ this stops collecting after first pause
lifecycleScope.launch { flow.collect { render(it) } }
// ✅ use repeatOnLifecycle
lifecycleScope.launch {
    lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
        flow.collect { render(it) }
    }
}

// 4. Wrong ViewModel instance — scoped to wrong NavBackStackEntry
// Fragment A and Fragment B may get different ViewModel instances
// ✅ verify scope: hiltViewModel() vs activityViewModels() vs navGraphViewModels()

// 5. stateIn with Lazily — Flow not started until first collector
// If you access .value before any collection, it's still the initialValue
// ✅ check: WhileSubscribed starts on first collect, Eagerly starts immediately

// 6. Missed emission with SharedFlow replay=0
// Emission fired before collector was active — event lost
// ✅ add extraBufferCapacity=1 or ensure collector is ready before emit

The mutable data bug is the hardest to spot: if your state holds a mutable list and you add an item to the list then reassign the same reference, equals() compares the new and old state — since it's the same list object with the same reference, they compare as equal and no emission happens. The fix is always immutable data classes with copy() or using persistentListOf from kotlinx.collections.immutable. The systematic approach — add onEach logging first — immediately tells you whether the problem is in the producer (no emissions) or the consumer (emissions happening, UI not responding).

snapshotFlow converts Compose snapshot state (anything read via remember, mutableStateOf, etc.) into a cold Flow. It observes the snapshot system and emits whenever any state read inside its block changes. This bridges the Compose reactive system with the coroutine Flow world.

// snapshotFlow — observe Compose state as a Flow
@Composable
fun SearchScreen(viewModel: SearchViewModel) {
    var query by remember { mutableStateOf("") }
    val listState = rememberLazyListState()

    // Bridge Compose state → Flow → ViewModel
    LaunchedEffect(viewModel) {
        snapshotFlow { query }         // observe query changes
            .debounce(300)
            .filter { it.length >= 2 }
            .distinctUntilChanged()
            .collect { viewModel.onSearch(it) }
    }

    // Detect when user scrolls to bottom of a list
    LaunchedEffect(listState) {
        snapshotFlow { listState.layoutInfo.visibleItemsInfo.lastOrNull()?.index }
            .filterNotNull()
            .distinctUntilChanged()
            .filter { idx -> idx >= listState.layoutInfo.totalItemsCount - 3 }
            .collect { viewModel.loadNextPage() }
    }

    LazyColumn(state = listState) { /* items */ }
}

// snapshotFlow reads snapshot state synchronously inside the block
// Emits: initial value immediately, then on every snapshot commit that changes the read values
// Automatically applies distinctUntilChanged — won't re-emit identical values
// Must be called from a coroutine context (LaunchedEffect, etc.)

The key difference from a regular Flow: snapshotFlow reads Compose snapshot state (registered via the Compose snapshot system), not regular Kotlin variables. It automatically re-runs whenever any State<T> read inside its block is invalidated by a recomposition. The flow only emits when the result of the block changes — it's implicitly distinctUntilChanged. The infinite scroll detection pattern (snapshotFlow on lazyListState) is the canonical production use case — it lets you trigger pagination from Compose scroll state using standard Flow operators like debounce and filter.

Use flowWithLifecycle (from lifecycle-runtime-ktx) or build a custom lifecycle-gated flow with callbackFlow and a LifecycleEventObserver. The built-in operator is cleaner for most cases.

// ✅ flowWithLifecycle — gate any Flow to lifecycle state
// dependency: lifecycle-runtime-ktx 2.6+
val foregroundData = repo.sensorFlow()
    .flowWithLifecycle(lifecycle, Lifecycle.State.STARTED)
    // Upstream pauses when lifecycle < STARTED
    // Resumes when lifecycle returns to STARTED

// In Fragment:
viewLifecycleOwner.lifecycleScope.launch {
    repo.locationFlow()
        .flowWithLifecycle(viewLifecycleOwner.lifecycle)
        .collect { location -> updateMap(location) }
}

// ✅ Custom: lifecycle events as a Flow
fun Lifecycle.eventsFlow(): Flow<Lifecycle.Event> = callbackFlow {
    val observer = LifecycleEventObserver { _, event -> trySend(event) }
    addObserver(observer)
    awaitClose { removeObserver(observer) }
}

// Usage: react to specific lifecycle events
lifecycle.eventsFlow()
    .filter { it == Lifecycle.Event.ON_RESUME }
    .collect { refreshData() }

// ⚠ flowWithLifecycle vs repeatOnLifecycle:
// flowWithLifecycle → operator on a SINGLE flow (cleaner for one flow)
// repeatOnLifecycle → block collecting MULTIPLE flows (one scope for all)
// For two+ flows, prefer repeatOnLifecycle to avoid multiple lifecycleScope.launch calls

flowWithLifecycle and repeatOnLifecycle achieve the same goal but have different ergonomics. flowWithLifecycle is an operator — it chains cleanly when you have a single flow to collect. repeatOnLifecycle is a scope builder — it's cleaner when you're collecting multiple flows simultaneously, since all collections share one lifecycle-aware coroutine scope. Mixing both can lead to redundant lifecycle coupling. The rule: one flow → flowWithLifecycle; multiple flows → one repeatOnLifecycle block with multiple collect calls or launchIn.

Shared data between ViewModels always flows through the data layer — never directly between ViewModels. The repository holds the shared stream as a hot SharedFlow or StateFlow; each ViewModel subscribes independently. This keeps ViewModels decoupled while sharing a single upstream source.

// ❌ Anti-pattern: ViewModels referencing each other
class CheckoutViewModel(
    private val cartViewModel: CartViewModel  // ❌ tight coupling
) : ViewModel()

// ✅ Correct: shared repository as single source of truth
class CartRepository {
    // Singleton (Hilt @Singleton) — one instance app-wide
    private val _cartState = MutableStateFlow(Cart.empty())
    val cartFlow: StateFlow<Cart> = _cartState.asStateFlow()

    fun addItem(item: Item) { _cartState.update { it.add(item) } }
}

// Both ViewModels inject the same repository
class CartViewModel @Inject constructor(repo: CartRepository) : ViewModel() {
    val cart = repo.cartFlow
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), Cart.empty())
}

class CheckoutViewModel @Inject constructor(repo: CartRepository) : ViewModel() {
    val orderSummary = repo.cartFlow
        .map { cart -> cart.toOrderSummary() }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), OrderSummary.empty())
}

// Alternative: NavGraph-scoped ViewModel for sibling screens in the same flow
// val sharedVm: CheckoutFlowViewModel by navGraphViewModels(R.id.checkout_graph)
// Lives exactly as long as the nav graph — survives rotation, dies on back

For screens within the same navigation flow (checkout step 1 → step 2 → step 3), the NavGraph-scoped ViewModel is the cleanest solution: one ViewModel scoped to the checkout nav graph holds the shared state, and each screen accesses it via navGraphViewModels. The ViewModel is automatically cleared when the user exits the checkout flow. For app-wide shared state (auth status, cart, user profile), a @Singleton repository with a StateFlow is correct — it survives the entire app session and any ViewModel can subscribe to it independently.

Kotlin Flow doesn't have a built-in timeout operator, but the pattern is straightforward using withTimeout inside a flow builder, or race between a data flow and a timer flow. withTimeout throws TimeoutCancellationException when the deadline expires.

// withTimeout inside flow — throws on timeout
fun fetchWithTimeout(): Flow<Data> = flow {
    withTimeout(5_000) {         // 5-second deadline
        emit(api.getData())
    }
}.catch { e ->
    if (e is TimeoutCancellationException) emit(Data.timeout())
    else throw e
}

// timeout for streaming flow — cancel if no emission for N ms
fun <T> Flow<T>.timeoutBetweenEmissions(ms: Long): Flow<T> = flow {
    collect { value ->
        withTimeout(ms) {
            emit(value)
        }
    }
}

// Race pattern: data vs timeout (no withTimeout dependency)
fun fetchOrTimeout(): Flow<Result<Data>> = merge(
    flow { emit(Result.success(api.getData())) },
    flow { delay(5_000); emit(Result.failure(TimeoutException())) }
).take(1)   // first one wins, other is cancelled

// Practical ViewModel usage
val uiState = flow {
    withTimeout(10_000) { emit(repo.loadHeavyData()) }
}.map   { UiState.Success(it) }
 .onStart { emit(UiState.Loading) }
 .catch  { e ->
     val msg = if (e is TimeoutCancellationException) "Request timed out" else "Error"
     emit(UiState.Error(msg))
 }.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), UiState.Loading)

The merge + take(1) race pattern is elegant for non-cooperative timeouts — useful when the upstream cannot be cancelled (legacy APIs). The first flow to emit wins; the other is cancelled by take(1). Always distinguish TimeoutCancellationException from other exceptions in catch — they should produce a "timeout" error message, not a generic "something went wrong." Never use Thread.sleep() for timeouts — it blocks the thread and ignores cancellation.

Geofence detection on a location stream uses filter + distinctUntilChangedBy to detect state transitions (outside → inside) without firing on every location update inside the fence.

data class GeofenceState(val insideFence: Boolean, val fenceId: String)

val geofenceTransitions: Flow<GeofenceState> = locationFlow()
    .map { loc ->
        val inside = geofences.any { fence -> fence.contains(loc) }
        GeofenceState(inside, if (inside) geofences.find { it.contains(loc) }!!.id else "")
    }
    .distinctUntilChangedBy { it.insideFence }  // only on enter/exit transition
    .filter { it.insideFence }                   // only on ENTER (not exit)

// Observe and trigger notification
geofenceTransitions
    .onEach { state ->
        notificationManager.showGeofenceNotification(state.fenceId)
    }
    .flowOn(Dispatchers.Default)  // geofence math on background thread
    .launchIn(serviceScope)        // ForegroundService scope — survives screen off

// For entry + exit events:
locationFlow()
    .map { loc -> geofences.firstOrNull { it.contains(loc) } }
    .distinctUntilChanged()      // null→fence = enter, fence→null = exit
    .zipWithNext()               // pair consecutive emissions
    .collect { (prev, curr) ->
        if (prev == null && curr != null) onEnter(curr)
        if (prev != null && curr == null) onExit(prev)
    }

The distinctUntilChangedBy { it.insideFence } is the key: without it, every GPS update inside the geofence (hundreds per hour) would trigger a notification. With it, only the first update that changes the insideFence flag from false→true fires the notification. The zipWithNext() pattern for entry/exit detection is clean: it pairs each emission with the next one, letting you detect direction of transition from the pair values.

A cold Flow is implemented as an object holding a lambda (the flow body). Calling collect executes that lambda with a Collector as receiver. There's no background coroutine, no channel, no shared state — just a lambda called on the collector's coroutine. A hot Flow (StateFlow/SharedFlow) maintains an internal state object and a subscriber list, independent of any collector.

// Cold Flow — compiled to roughly this:
val coldFlow: Flow<Int> = object : Flow<Int> {
    override suspend fun collect(collector: FlowCollector<Int>) {
        // flow { emit(1); emit(2) } body runs HERE
        collector.emit(1)
        collector.emit(2)
    }
}
// collect() == calling the lambda. No separate coroutine is started.
// Two collectors = lambda runs twice = two independent executions

// Hot Flow (StateFlow) — maintains state independent of collectors
// Internally: atomic value + CopyOnWriteArrayList of SlotArrays (subscriber slots)
// Emission: update value → wake all suspended collectors via slot resumption
// Collector: suspend on a slot until resumed by emission

// Flow operator chain — each operator wraps the previous:
flow.map { it * 2 }.filter { it > 4 }
// Compiled to roughly:
val mapped   = MapFlow(upstream = flow, transform = { it * 2 })
val filtered = FilterFlow(upstream = mapped, predicate = { it > 4 })
// No data moves until collect() is called on filtered
// collect(filtered) → collect(mapped) → collect(flow) — chain runs inside-out

// K2 compiler optimisation (Kotlin 2.0+):
// Simple operator chains (map, filter) may be inlined/fused
// Fewer intermediate Flow wrapper objects allocated
// CPS state machine for suspend functions is more compact

Understanding the cold Flow as "a lambda called by collect" explains many behaviours: why two collectors get independent executions (the lambda runs twice), why there's no memory overhead until collect is called, and why flowOn inserts a channel between the lambda and the collector (it's the only way to run the lambda on a different coroutine/dispatcher than the collector). Hot flows work completely differently — they have internal state and a registry of suspended collectors that are resumed on each emission.

Form validation with Flow uses a StateFlow per field, validation logic as operators, and combine to derive the overall form validity. Each field's validity is a derived StateFlow — the submit button observes the combined validity.

class RegistrationViewModel : ViewModel() {
    val email    = MutableStateFlow("")
    val password = MutableStateFlow("")
    val phone    = MutableStateFlow("")

    // Per-field validation — derived StateFlows
    val emailError: StateFlow<String?> = email
        .debounce(400)
        .map { e -> if ("@" !in e) "Invalid email" else null }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), null)

    val passwordError: StateFlow<String?> = password
        .debounce(400)
        .map { p -> when { p.length < 8 -> "Min 8 chars"; else -> null } }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), null)

    // Submit enabled only when ALL fields are valid
    val canSubmit: StateFlow<Boolean> = combine(
        emailError, passwordError
    ) { emailErr, pwErr ->
        emailErr == null && pwErr == null
            && email.value.isNotEmpty()
            && password.value.isNotEmpty()
    }.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), false)

    fun submit() {
        if (!canSubmit.value) return
        viewModelScope.launch { api.register(email.value, password.value) }
    }
}

// Compose UI — each field observes its own error, button observes canSubmit
val emailErr by vm.emailError.collectAsStateWithLifecycle()
val canSubmit by vm.canSubmit.collectAsStateWithLifecycle()
Button(onClick = vm::submit, enabled = canSubmit) { Text("Register") }

The debounce on each field prevents showing validation errors while the user is still typing. Each field gets its own independent debounce, so slow typists in one field don't delay validation in another. The combine on error StateFlows re-evaluates button state whenever any field changes. Because the error flows are hot StateFlows (via stateIn), the combine starts immediately with initial null errors — the button starts disabled and only enables once all validations pass.

FCM callbacks arrive in FirebaseMessagingService.onMessageReceived() — a non-coroutine callback that can fire when the app is in any state. The bridge uses a @Singleton SharedFlow in the repository, written from the Service and observed in the ViewModel.

// @Singleton notification repository — bridges Service and ViewModel
class NotificationRepository @Inject constructor() {
    private val _messages = MutableSharedFlow<PushMessage>(
        extraBufferCapacity = 16,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )
    val messages: SharedFlow<PushMessage> = _messages

    // Called from FCM Service — non-suspending, fire-and-forget
    fun onMessageReceived(msg: PushMessage) {
        _messages.tryEmit(msg)  // never suspends, returns false if buffer full
    }
}

// FCM Service — inject repository and forward to SharedFlow
class MyFcmService : FirebaseMessagingService() {
    @Inject lateinit var repo: NotificationRepository

    override fun onMessageReceived(message: RemoteMessage) {
        repo.onMessageReceived(PushMessage.from(message))
        // Also persist to Room for offline access:
        // GlobalScope.launch(IO) { dao.insert(PushMessage.from(message)) }
    }
}

// ViewModel — observe SharedFlow from repository
class InboxViewModel @Inject constructor(repo: NotificationRepository) : ViewModel() {
    val liveMessages = repo.messages
        .onEach { msg -> dao.insert(msg) }   // persist each received message
        .launchIn(viewModelScope)

    val allMessages = dao.getMessagesFlow()
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), emptyList())
}

Using tryEmit (not emit) from the Service is critical — emit is a suspend function that cannot be called from a non-coroutine context. tryEmit returns false if the buffer is full; with extraBufferCapacity = 16 and DROP_OLDEST overflow, this only happens under extreme load. The Singleton ensures the same SharedFlow instance is shared between the Service and all ViewModels. Also persist each received notification to Room so missed notifications (app killed) are available when the app reopens.

A wizard is a stateful multi-step flow. The cleanest design uses a single NavGraph-scoped ViewModel holding a StateFlow of the wizard state, with explicit step transitions and per-step validation.

data class WizardState(
    val step: Step = Step.Name,
    val name: String = "",
    val email: String = "",
    val planId: String? = null,
    val isLoading: Boolean = false
)
enum class Step { Name, Email, Plan, Confirm }

class OnboardingViewModel : ViewModel() {
    private val _state = MutableStateFlow(WizardState())
    val state = _state.asStateFlow()

    private val _events = MutableSharedFlow<WizardEvent>()
    val events = _events.asSharedFlow()

    // canAdvance derived from current step's validation
    val canAdvance: StateFlow<Boolean> = state
        .map { s ->
            when (s.step) {
                Step.Name  -> s.name.length >= 2
                Step.Email -> "@" in s.email
                Step.Plan  -> s.planId != null
                Step.Confirm -> true
            }
        }
        .distinctUntilChanged()
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), false)

    fun next() = viewModelScope.launch {
        val s = _state.value
        if (s.step == Step.Confirm) {
            _state.update { it.copy(isLoading = true) }
            val result = runCatching { api.createAccount(s) }
            _state.update { it.copy(isLoading = false) }
            if (result.isSuccess) _events.emit(WizardEvent.Finish)
            else _events.emit(WizardEvent.ShowError(result.exceptionOrNull()!!))
        } else {
            _state.update { it.copy(step = Step.values()[s.step.ordinal + 1]) }
        }
    }
    fun back() { _state.update { it.copy(step = Step.values()[it.step.ordinal - 1]) } }
    fun setName(v: String)   { _state.update { it.copy(name = v) } }
    fun setEmail(v: String)  { _state.update { it.copy(email = v) } }
}

The NavGraph-scoped ViewModel survives across all wizard screens — each step accesses the same ViewModel via navGraphViewModels(R.id.onboarding_graph). When the user presses Back and returns to a previous step, their data is already populated from the StateFlow. The ViewModel is automatically cleared when the user exits the onboarding graph (pressing back past the first step or completing the wizard). The canAdvance derived StateFlow means the Next button's enabled state is always declarative and automatically correct for the current step.

Neither StateFlow nor SharedFlow survive process death — they are in-memory objects. When the OS kills the process, all in-memory state is lost. Handling process death correctly requires persisting critical state to SavedStateHandle (for transient UI state) or Room/DataStore (for durable data).

// What is LOST on process death:
// MutableStateFlow values
// MutableSharedFlow buffered items
// ViewModel properties (ViewModels are also destroyed)
// Any in-memory cache

// What SURVIVES process death:
// Room database data
// DataStore values
// SharedPreferences
// SavedStateHandle (via Activity's saved instance state bundle)

// ✅ Pattern: persist transient UI state in SavedStateHandle
class SearchViewModel @Inject constructor(
    private val savedState: SavedStateHandle
) : ViewModel() {

    // SavedStateHandle.getStateFlow — backed by saved state AND a StateFlow!
    val query: StateFlow<String> =
        savedState.getStateFlow("query", initialValue = "")
    // On process death + restore: query is automatically restored from bundle

    fun setQuery(q: String) {
        savedState["query"] = q  // writes to both SavedStateHandle AND StateFlow
    }

    // Results derived from the process-death-safe query
    val results = query
        .debounce(300)
        .flatMapLatest { repo.search(it) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), emptyList())
}
// After process death: user re-opens app → previous query auto-restored →
// search re-runs automatically (query StateFlow emits the restored value)

SavedStateHandle.getStateFlow() is the ideal bridge: it wraps a SavedStateHandle key as a StateFlow, so you get the reactive declarative API of StateFlow with the process-death persistence of SavedStateHandle. Writing to the key via savedState[key] = value updates both the bundle (persisted) and the StateFlow (reactive). This single line replaces the old pattern of manually observing SavedStateHandle.getLiveData() and converting it. Any derived Flow that reads from this StateFlow will automatically re-run after process death and restoration.

A typing indicator requires two things: detecting typing events from the WebSocket stream, and auto-clearing the indicator if no events arrive within a timeout window. This is a classic debounce-to-false pattern.

// Typing indicator — show on event, auto-hide after silence
val isTyping: StateFlow<Boolean> = wsMessages
    .filter { it.type == "typing" && it.userId == partnerId }
    .transformLatest {
        emit(true)           // show indicator immediately
        delay(3_000)        // wait 3s for next typing event
        emit(false)          // hide if no new event
    }
    .onStart { emit(false) }
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), false)

// transformLatest is the key:
// Event arrives → emit(true) → start 3s delay
// New event arrives within 3s → previous block CANCELLED, restart
// No event for 3s → delay completes → emit(false) → indicator hidden

// Sending your OWN typing events to the WebSocket
val localTyping = MutableStateFlow("")  // current draft text
localTyping
    .filter { it.isNotEmpty() }
    .debounce(200)                    // don't spam on every keystroke
    .onEach { ws.sendTypingEvent(roomId) }
    .launchIn(viewModelScope)

transformLatest is the perfect operator for "show immediately, auto-expire" patterns. Each new typing event cancels the previous 3-second countdown and starts fresh — so a user typing continuously keeps the indicator visible indefinitely without a new network event for each keystroke. The indicator only disappears 3 seconds after the last typing event, matching the UX expectation. The outbound typing debounce prevents flooding the WebSocket — sending a typing event at most every 200ms regardless of how fast the user types.

Kotlin Flow doesn't include windowed or chunked as built-in operators (they exist on collections and sequences), but they can be built using buffer, scan, or custom flow operators. Batching is essential for analytics, bulk database writes, and UI list updates.

// chunked — collect N items then emit as a list
fun <T> Flow<T>.chunked(size: Int): Flow<List<T>> = flow {
    val buffer = mutableListOf<T>()
    collect { item ->
        buffer.add(item)
        if (buffer.size == size) {
            emit(buffer.toList())
            buffer.clear()
        }
    }
    if (buffer.isNotEmpty()) emit(buffer.toList())  // flush remainder
}

// timed batching — emit every N seconds OR when size reached
fun <T> Flow<T>.batchedWithTimeout(size: Int, timeoutMs: Long): Flow<List<T>> =
    channelFlow {
        val batch = mutableListOf<T>()
        var job: Job? = null
        collect { item ->
            batch.add(item)
            job?.cancel()
            job = launch { delay(timeoutMs); send(batch.toList()); batch.clear() }
            if (batch.size >= size) { job?.cancel(); send(batch.toList()); batch.clear() }
        }
        if (batch.isNotEmpty()) send(batch.toList())
    }

// Production: batch analytics events — send every 20 events OR every 30 seconds
analyticsEventFlow
    .batchedWithTimeout(size = 20, timeoutMs = 30_000)
    .onEach { batch -> analytics.sendBatch(batch) }
    .launchIn(appScope)

// Production: batch Room inserts for performance
incomingDataFlow
    .chunked(50)
    .onEach { batch -> dao.insertAll(batch) }  // one DB transaction per 50 items
    .launchIn(viewModelScope)

Batching analytics events is one of the most impactful production optimisations: sending 100 events as 5 batches of 20 uses 95% fewer HTTP connections than sending each individually. The timed batch (emit after N items OR after timeout, whichever comes first) handles the edge case of low-traffic periods — events don't queue forever waiting for a full batch. For Room inserts, batching into 50-item transactions is dramatically faster than 50 individual inserts because each insert in isolation commits a database transaction.

Background sync requires: a local mutation queue (Room), an online/offline detector (Flow), and a sync worker that drains the queue when connectivity returns. The reactive pipeline connects all three.

// Sync architecture with Flow
class SyncRepository(
    private val dao: SyncQueueDao,
    private val api: Api,
    private val networkFlow: Flow<Boolean>
) {
    // React to connectivity: drain queue when online
    fun startSync(): Flow<SyncResult> = networkFlow
        .filter { it }                     // only when online
        .flatMapLatest {
            dao.getPendingItemsFlow()     // Room: emits whenever queue changes
        }
        .filter { it.isNotEmpty() }
        .flatMapConcat { pending ->        // process one batch at a time
            flow {
                pending.forEach { item ->
                    val result = runCatching { api.sync(item) }
                    if (result.isSuccess) {
                        dao.markSynced(item.id)
                        emit(SyncResult.Success(item.id))
                    } else {
                        // Conflict resolution: server wins, update local
                        val err = result.exceptionOrNull()
                        if (err is ConflictException) {
                            dao.update(item.id, err.serverValue)
                            dao.markSynced(item.id)
                        } else {
                            dao.incrementRetry(item.id)  // back-off
                        }
                        emit(SyncResult.Error(item.id, err))
                    }
                }
            }
        }

    // Mutations enqueue locally — always succeed immediately
    suspend fun enqueue(mutation: Mutation) = dao.insert(SyncItem.from(mutation))
}

flatMapLatest on the network flow means connectivity changes immediately cancel any in-progress sync (going offline) and restart the sync pipeline (coming back online). flatMapConcat on the pending items flow ensures one item is fully synced (success or fail) before the next starts — preventing interleaved partial syncs. Room's reactive Flow automatically re-emits the pending queue after each markSynced, driving the pipeline forward without polling. WorkManager should wrap this for guaranteed delivery when the app is killed — the Flow handles the reactive in-app experience.

merge combines multiple flows into one, emitting values from all sources concurrently as they arrive. There are no ordering guarantees — whichever flow emits first wins. This is the right tool when you have multiple independent sources that contribute to the same stream.

// merge — concurrent, no ordering guarantees
val allNotifications = merge(
    pushNotificationFlow,      // from FCM
    inAppMessageFlow,          // from WebSocket
    localReminderFlow          // from AlarmManager
).collect { notification -> show(notification) }
// All three sources active simultaneously, emissions interleave freely

// Extension: Flow.mergeWith
pushFlow.mergeWith(localFlow)  // merge two flows

// combine vs merge — key distinction
// combine: waits for ALL sources to emit, pairs latest values
// merge: any source emits → immediately forwarded downstream

// Real use case: aggregating errors from multiple subsystems
val errorFlow = merge(
    networkLayer.errors(),
    databaseLayer.errors(),
    authLayer.errors()
).map { err -> AppError.from(err) }
 .shareIn(appScope, SharingStarted.Eagerly, replay = 0)

// ⚠ merge does NOT cancel if one source errors
// One source throwing cancels the entire merged flow
// Fix: wrap each source in catch{} before merging
val safe = merge(
    flowA.catch { emit(Default) },
    flowB.catch { emit(Default) }
)

The error propagation behaviour is the most important detail: if any merged source throws an uncaught exception, the entire merged flow terminates. In a notification aggregator, a broken WebSocket connection should not kill the local reminder flow. The fix is wrapping each source with catch { } before merging, so errors are handled per-source. For high-reliability error streams, use a SharedFlow(replay=0) as the merge target and route each source through it with separate launchIn calls — source failures are isolated.

Optimistic UI requires immediately updating local state, firing the network request in the background, and restoring the previous state on failure. With StateFlow this is clean: snapshot the current state, apply the optimistic update, attempt the API call, rollback on error.

fun toggleLike(postId: String) = viewModelScope.launch {
    // 1. Snapshot current state for rollback
    val previous = _state.value

    // 2. Apply optimistic update immediately — UI responds at once
    _state.update { state ->
        state.copy(posts = state.posts.map {
            if (it.id == postId) it.copy(liked = !it.liked, likes = it.likes + if (it.liked) -1 else 1)
            else it
        })
    }

    // 3. Confirm with server
    val result = runCatching { api.toggleLike(postId) }

    // 4. Rollback on failure, emit error event
    if (result.isFailure) {
        _state.value = previous   // restore exact previous state
        _events.emit(UiEvent.ShowSnackbar("Failed to update like"))
    }
}

// Room-backed version: optimistic local DB write, sync in background
fun archiveEmail(id: String) = viewModelScope.launch {
    dao.setArchived(id, true)         // Room emits new list instantly
    val ok = runCatching { api.archive(id) }
    if (ok.isFailure) {
        dao.setArchived(id, false)     // Room re-emits with rollback
        _events.emit(UiEvent.ShowError("Archive failed"))
    }
}

The Room-backed version is especially elegant: writing the optimistic state to the local DB triggers the Room Flow to re-emit immediately — all collectors see the update in one Frame with no extra StateFlow management. On rollback, writing the restored state triggers another Room emission, and collectors see the correction transparently. This pattern scales well: every screen observing the same Room Flow automatically sees both the optimistic update and the rollback without any additional coordination.

distinctUntilChanged() skips an emission if the whole object equals the previous emission. distinctUntilChangedBy { selector } skips an emission only if the selected property is equal — useful when you only care about a specific field changing, not the entire object.

data class User(val id: Int, val name: String, val lastSeen: Long)

// distinctUntilChanged — skips if ENTIRE object is equal
userFlow
    .distinctUntilChanged()
    .collect { render(it) }
// lastSeen updating every second → re-renders every second

// distinctUntilChangedBy — skips if selected KEY is equal
userFlow
    .distinctUntilChangedBy { it.name }  // only emit when name changes
    .collect { renderName(it.name) }
// lastSeen changes → skipped (name unchanged) — no redundant renders

// Real use case: avoid redundant API calls when irrelevant fields update
settingsFlow
    .distinctUntilChangedBy { it.language }  // only on language change
    .flatMapLatest { settings -> api.getLocalizedContent(settings.language) }

// Custom equality with distinctUntilChanged(comparator)
userFlow.distinctUntilChanged { old, new ->
    old.id == new.id && old.name == new.name  // ignore lastSeen
}

// StateFlow already applies distinctUntilChanged internally
// Adding .distinctUntilChanged() on a StateFlow is a no-op — already deduplicated
// BUT: adding .map{}.distinctUntilChanged() after map IS useful
cartFlow
    .map { it.total }
    .distinctUntilChanged()    // only emit when total changes, not on every cart mutation
    .collect { updateTotalLabel(it) }

The map + distinctUntilChanged combination is one of the most performance-relevant Flow patterns in Android: extract only the property you care about, then deduplicate. A cart StateFlow might emit every time any item changes (quantity, selection, coupon), but the checkout button only needs to know the total. cartFlow.map { it.total }.distinctUntilChanged() means the checkout button only recomposes when the total changes — not on every cart mutation. This directly reduces unnecessary UI recompositions in Compose.

This combines callbackFlow (ConnectivityManager callbacks), periodic RTT pings, and Flow operators to produce a reactive connection quality stream. The key is separating connectivity detection (system callbacks) from quality measurement (periodic pings).

enum class ConnectionQuality { OFFLINE, DEGRADED, GOOD }

// Step 1: network availability as Flow
fun Context.networkAvailabilityFlow(): Flow<Boolean> = callbackFlow {
    val cm = getSystemService(ConnectivityManager::class.java)
    val cb = object : NetworkCallback() {
        override fun onAvailable(n: Network) { trySend(true) }
        override fun onLost(n: Network)      { trySend(false) }
    }
    cm.registerDefaultNetworkCallback(cb)
    awaitClose { cm.unregisterNetworkCallback(cb) }
}

// Step 2: RTT ping flow (only when connected)
fun pingFlow(): Flow<Long> = flow {
    while (true) {
        val rtt = measureTimeMillis { api.ping() }
        emit(rtt)
        delay(10_000)  // measure every 10s
    }
}

// Step 3: combine into quality flow
val connectionQuality: StateFlow<ConnectionQuality> =
    networkAvailabilityFlow()
        .flatMapLatest { connected ->
            if (!connected) flowOf(ConnectionQuality.OFFLINE)
            else pingFlow().map { rtt ->
                when {
                    rtt < 200  -> ConnectionQuality.GOOD
                    rtt < 1000 -> ConnectionQuality.DEGRADED
                    else       -> ConnectionQuality.OFFLINE
                }
            }.catch { emit(ConnectionQuality.DEGRADED) }
        }
        .distinctUntilChanged()
        .stateIn(appScope, SharingStarted.Eagerly, ConnectionQuality.GOOD)

Using flatMapLatest here means when the network drops, the ping flow is immediately cancelled (no point pinging when offline) and OFFLINE is emitted instantly. When the network reconnects, a fresh ping flow starts. distinctUntilChanged() prevents re-notifying the UI when the quality stays the same across pings. SharingStarted.Eagerly is correct here because connection quality is app-global and should always be monitored — not just when a screen is active.

Pull-to-refresh requires a manual refresh trigger alongside the automatic data stream. The cleanest pattern is a refresh-trigger StateFlow that flatMapLatest switches between, causing the data flow to restart while preserving the accumulated state for the UI.

class FeedViewModel : ViewModel() {
    private val _refreshTrigger = MutableStateFlow(0)  // increment to refresh

    val uiState: StateFlow<FeedState> = _refreshTrigger
        .flatMapLatest {
            repo.feedFlow()
                .map { FeedState.Success(it) }
                .onStart { emit(FeedState.Loading) }
                .catch   { emit(FeedState.Error(it)) }
        }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), FeedState.Loading)

    fun refresh() { _refreshTrigger.update { it + 1 } }
}

// In Compose: SwipeRefresh + collectAsStateWithLifecycle
@Composable
fun FeedScreen(vm: FeedViewModel = hiltViewModel()) {
    val state by vm.uiState.collectAsStateWithLifecycle()
    val isRefreshing = state is FeedState.Loading

    SwipeRefresh(state = rememberSwipeRefreshState(isRefreshing), onRefresh = vm::refresh) {
        when (state) {
            is FeedState.Success -> FeedList((state as FeedState.Success).items)
            is FeedState.Error   -> ErrorView()
            is FeedState.Loading -> LoadingSpinner()
        }
    }
}

// Debounce rapid refresh taps (accidental double-swipe)
fun refresh() {
    if (uiState.value is FeedState.Loading) return  // already refreshing
    _refreshTrigger.update { it + 1 }
}

Incrementing an integer StateFlow to trigger refreshes is elegant: each new value causes flatMapLatest to cancel the previous flow and start fresh. The new flow emits Loading immediately (via onStart), then data or error. The guard against double-refresh — checking if state is already Loading — prevents redundant network calls from rapid swipes. This pattern also handles error retry: the same refresh() function works whether the user pulls to refresh or taps a "Retry" button after an error.

filterIsInstance<T>() is a type-safe filter that keeps only emissions that are instances of type T, casting them automatically. It's ideal for event-bus flows carrying a sealed class hierarchy where different consumers care about different subtypes.

sealed class AppEvent {
    data class Navigate(val route: String) : AppEvent()
    data class ShowSnackbar(val message: String) : AppEvent()
    data class Analytics(val event: String, val params: Map<String, Any>) : AppEvent()
    object Logout : AppEvent()
}

// Singleton event bus in app module
@Singleton
class AppEventBus @Inject constructor() {
    private val _events = MutableSharedFlow<AppEvent>(
        replay = 0,
        extraBufferCapacity = 64
    )
    val events: SharedFlow<AppEvent> = _events.asSharedFlow()

    suspend fun emit(event: AppEvent) = _events.emit(event)
    fun tryEmit(event: AppEvent) = _events.tryEmit(event)
}

// Navigation handler — only cares about Navigate events
class MainActivity : AppCompatActivity() {
    private val eventBus: AppEventBus by inject()

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        lifecycleScope.launch {
            eventBus.events
                .filterIsInstance<AppEvent.Navigate>()
                .collect { nav ->
                    navController.navigate(nav.route)
                }
        }
    }
}

// Analytics handler — only cares about Analytics events
class AnalyticsManager @Inject constructor(
    private val eventBus: AppEventBus
) {
    fun observe(scope: CoroutineScope) {
        eventBus.events
            .filterIsInstance<AppEvent.Analytics>()
            .onEach { analytics ->
                firebaseAnalytics.logEvent(analytics.event, analytics.params.toBundle())
            }
            .launchIn(scope)
    }
}

// Under the hood — filterIsInstance is a simple inline extension:
// inline fun <reified R> Flow<*>.filterIsInstance() = filter { it is R }.map { it as R }
// Reified T means zero reflection overhead — bytecode uses instanceof check

Performance implications: filterIsInstance is inline with a reified type parameter, so the JVM generates a direct instanceof check with no reflection. Cost is O(1) per emission. The shared flow fans out to all subscribers; each subscriber independently filters its subtype. No allocation overhead beyond the event object itself. The only cost is the extra subscriber coroutine per handler — which is negligible.

All three are stateful accumulation operators, but differ in initial value handling and aliasing. scan (alias for runningFold) takes an initial value and emits it immediately before any upstream emission. runningReduce has no initial value and emits starting from the second element — similar to reduce but streaming.

data class ScoreEvent(val userId: String, val delta: Int)
data class LeaderboardEntry(val userId: String, val totalScore: Int)

// Live score stream from WebSocket
fun leaderboardFlow(
    scoreEvents: Flow<ScoreEvent>
): Flow<List<LeaderboardEntry>> {
    return scoreEvents
        .runningFold(
            initial = emptyMap<String, Int>()  // emitted immediately as first value
        ) { accumulator, event ->
            accumulator + (event.userId to (accumulator.getOrDefault(event.userId, 0) + event.delta))
        }
        .map { scoreMap ->
            scoreMap.entries
                .sortedByDescending { it.value }
                .map { LeaderboardEntry(it.key, it.value) }
        }
}

// scan is identical — just an alias for runningFold
val runningTotal: Flow<Int> = scoreChannel
    .scan(0) { acc, event -> acc + event.delta }

// runningReduce — no initial value, first element is "initial"
val maxSoFar: Flow<Int> = scoresFlow
    .runningReduce { acc, value -> maxOf(acc, value) }
// WARNING: runningReduce emits nothing if upstream is empty!

// ViewModel wires it to UI
val leaderboard: StateFlow<List<LeaderboardEntry>> = leaderboardFlow(webSocketFlow)
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

Key distinction: runningFold/scan always emits the initial value first, making it safe for StateFlow (collector sees a value immediately). runningReduce skips the initial emit and is unsafe if the upstream can be empty — the flow completes without any emission. Use runningFold for UI state accumulation, runningReduce only for pure stream reduction where empty inputs are impossible.

WorkManager and Kotlin Flow integrate via WorkInfo LiveData (or Flow) that the ViewModel observes. Inside the worker, progress is published via setProgress(); the UI observes WorkManager.getWorkInfoByIdFlow() and reads the progress data.

// CoroutineWorker reports incremental progress
class MediaSyncWorker(ctx: Context, params: WorkerParameters) :
    CoroutineWorker(ctx, params) {

    override suspend fun doWork(): Result {
        val items = fetchItemsToSync()
        items.forEachIndexed { index, item ->
            syncItem(item)
            val progress = workDataOf(
                "progress" to ((index + 1) * 100 / items.size),
                "current" to item.name
            )
            setProgress(progress)  // pushes WorkInfo update
        }
        return Result.success()
    }
}

// ViewModel observes WorkInfo as Flow
class SyncViewModel @Inject constructor(
    private val workManager: WorkManager
) : ViewModel() {

    private var syncWorkId: UUID? = null

    val syncProgress: StateFlow<SyncState> = flow {
        val id = syncWorkId ?: return@flow
        workManager.getWorkInfoByIdFlow(id)
            .collect { info ->
                when (info?.state) {
                    WorkInfo.State.RUNNING -> {
                        val pct = info.progress.getInt("progress", 0)
                        val cur = info.progress.getString("current") ?: ""
                        emit(SyncState.InProgress(pct, cur))
                    }
                    WorkInfo.State.SUCCEEDED -> emit(SyncState.Done)
                    WorkInfo.State.FAILED -> emit(SyncState.Error)
                    else -> emit(SyncState.Idle)
                }
            }
    }.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), SyncState.Idle)

    fun startSync() {
        val request = OneTimeWorkRequestBuilder<MediaSyncWorker>().build()
        workManager.enqueue(request)
        syncWorkId = request.id
    }
}

// Compose UI
val state by viewModel.syncProgress.collectAsStateWithLifecycle()
when (val s = state) {
    is SyncState.InProgress -> LinearProgressIndicator(progress = s.percent / 100f)
    SyncState.Done -> Text("Sync complete")
    else -> {}
}

The key insight: getWorkInfoByIdFlow() returns a cold Flow backed by a Room query — it emits a new WorkInfo every time WorkManager persists a state change. This works even when the app is foregrounded because WorkManager writes to its Room DB regardless. Progress updates are throttled by WorkManager's internal batching (~1 update per few hundred ms under load).

Turbine's test{} block suspends the Flow in a controlled environment, letting you assert items, errors, and completion. For SharedFlow with replay=0 (one-shot events), you verify exactly one emission then cancelAndIgnoreRemainingEvents().

// ViewModel under test
class LoginViewModel(private val authRepo: AuthRepository) : ViewModel() {
    private val _effects = MutableSharedFlow<LoginEffect>(replay = 0)
    val effects: SharedFlow<LoginEffect> = _effects.asSharedFlow()

    fun onLoginClicked(email: String, password: String) {
        viewModelScope.launch {
            val success = authRepo.login(email, password)
            if (success) _effects.emit(LoginEffect.NavigateToHome)
            else _effects.emit(LoginEffect.ShowError("Invalid credentials"))
        }
    }
}

// Test using Turbine + MockK + StandardTestDispatcher
class LoginViewModelTest {
    private val authRepo = mockk<AuthRepository>()
    private lateinit var viewModel: LoginViewModel

    @BeforeEach fun setup() {
        Dispatchers.setMain(StandardTestDispatcher())
        viewModel = LoginViewModel(authRepo)
    }

    @Test fun `successful login emits NavigateToHome exactly once`() = runTest {
        coEvery { authRepo.login(any(), any()) } returns true

        viewModel.effects.test {
            viewModel.onLoginClicked("user@test.com", "pass123")

            val item = awaitItem()
            assertThat(item).isEqualTo(LoginEffect.NavigateToHome)

            // No more emissions — SharedFlow replay=0 means no buffered items
            expectNoEvents()
            cancelAndIgnoreRemainingEvents()
        }
    }

    @Test fun `failed login emits ShowError, not NavigateToHome`() = runTest {
        coEvery { authRepo.login(any(), any()) } returns false

        viewModel.effects.test {
            viewModel.onLoginClicked("user@test.com", "wrong")

            val item = awaitItem()
            assertThat(item).isInstanceOf(LoginEffect.ShowError::class.java)
            cancelAndIgnoreRemainingEvents()
        }
    }

    // Test that effect is NOT replayed to a late subscriber
    @Test fun `effect not replayed to new subscriber`() = runTest {
        coEvery { authRepo.login(any(), any()) } returns true
        viewModel.onLoginClicked("a@b.com", "pass")
        advanceUntilIdle()  // let emission happen before subscribing

        viewModel.effects.test {
            expectNoEvents()  // late subscriber misses the event — replay=0
            cancelAndIgnoreRemainingEvents()
        }
    }
}

The third test is the critical one: it verifies that replay=0 means a subscriber who arrives late misses already-emitted events. This is the contract you depend on to avoid re-navigating when the screen re-subscribes after configuration change.

DataStore's data property is already a cold Flow<Preferences> backed by file I/O on a dedicated dispatcher. Wrap it in a repository that maps raw Preferences to typed domain objects and exposes them as StateFlow for the ViewModel layer.

// DataStore keys
private val THEME_KEY = stringPreferencesKey("theme")
private val NOTIFICATIONS_KEY = booleanPreferencesKey("notifications_enabled")
private val FONT_SIZE_KEY = intPreferencesKey("font_size")

data class UserPreferences(
    val theme: Theme = Theme.SYSTEM,
    val notificationsEnabled: Boolean = true,
    val fontSize: Int = 16
)

class PreferencesRepository @Inject constructor(
    private val dataStore: DataStore<Preferences>
) {
    // Cold Flow — maps raw prefs to typed domain object
    val preferences: Flow<UserPreferences> = dataStore.data
        .catch { e ->
            if (e is IOException) emit(emptyPreferences())
            else throw e
        }
        .map { prefs ->
            UserPreferences(
                theme = Theme.valueOf(prefs[THEME_KEY] ?: Theme.SYSTEM.name),
                notificationsEnabled = prefs[NOTIFICATIONS_KEY] ?: true,
                fontSize = prefs[FONT_SIZE_KEY] ?: 16
            )
        }

    suspend fun setTheme(theme: Theme) {
        dataStore.edit { it[THEME_KEY] = theme.name }
    }

    suspend fun setNotifications(enabled: Boolean) {
        dataStore.edit { it[NOTIFICATIONS_KEY] = enabled }
    }

    suspend fun setFontSize(size: Int) {
        dataStore.edit { it[FONT_SIZE_KEY] = size.coerceIn(12, 24) }
    }
}

// ViewModel — converts cold Flow to hot StateFlow
class SettingsViewModel @Inject constructor(
    private val prefs: PreferencesRepository
) : ViewModel() {

    val preferences: StateFlow<UserPreferences> = prefs.preferences
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5000),
            initialValue = UserPreferences()  // default until first disk read
        )

    fun onThemeChange(theme: Theme) {
        viewModelScope.launch { prefs.setTheme(theme) }
    }
}

// Compose — preferences are instantly reactive
val prefs by viewModel.preferences.collectAsStateWithLifecycle()
DarkTheme(enabled = prefs.theme == Theme.DARK) { /* app content */ }

DataStore guarantees atomic writes and will never corrupt data mid-write (unlike SharedPreferences). The catch { emit(emptyPreferences()) } handles the first-run case or corrupt files gracefully. The stateIn with an in-memory default means the UI renders immediately without waiting for the disk read to complete.

LiveData and StateFlow have several subtle behavioral differences that bite developers during migration. The key ones: null safety, threading model, distinct-until-changed behavior, and lifecycle integration.

// DIFFERENCE 1: Null safety
// LiveData can hold null — StateFlow<T> cannot unless T is nullable
val liveData: MutableLiveData<User> = MutableLiveData()  // initial null
val stateFlow: MutableStateFlow<User?> = MutableStateFlow(null)  // must be explicit
// Or use a sealed state wrapper to avoid nullable StateFlow:
sealed class Async<out T> {
    object Loading : Async<Nothing>()
    data class Success<T>(val data: T) : Async<T>()
    data class Error(val e: Throwable) : Async<Nothing>()
}

// DIFFERENCE 2: Threading — LiveData posts to main; StateFlow does not
// LiveData (safe to call from any thread):
liveData.postValue(user)  // auto-switches to main thread
// StateFlow (must be on correct dispatcher or use update{}):
_stateFlow.value = user  // safe — StateFlow.value is thread-safe atomic
// ✅ StateFlow.value setter IS thread-safe — no postValue equivalent needed

// DIFFERENCE 3: Equality / distinctUntilChanged behavior
// LiveData emits every setValue() regardless of equality
liveData.value = sameUser  // observer fires even if value == current value
// StateFlow skips emission if new value == current value
_stateFlow.value = sameUser  // NO emission if structurally equal (equals())
// FIX: if you relied on LiveData re-emitting same value, use SharedFlow or force change:
// _events.emit(event) // SharedFlow always emits

// DIFFERENCE 4: Lifecycle — LiveData is lifecycle-aware, StateFlow is not
// LiveData observer stops when lifecycle is STOPPED
viewModel.liveData.observe(this) { /* auto lifecycle-safe */ }
// StateFlow must use repeatOnLifecycle or collectAsStateWithLifecycle
lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.stateFlow.collect { /* lifecycle-safe */ }
    }
}

// DIFFERENCE 5: Initial value — StateFlow always has a value, LiveData may not
// If you check liveData.value in tests before posting, it may be null
// StateFlow.value is never null (for non-nullable T) — always initialized

// Migration helper: asLiveData() bridge for gradual migration
val legacyLiveData: LiveData<User> = viewModel.userFlow.asLiveData()

The most common migration bug: code that relied on LiveData re-emitting the same object to trigger an update (e.g., a mutable list mutated in place then posted). StateFlow's equality check means this update is swallowed. Fix: use immutable data structures, or call _stateFlow.update { it.copy(...) } to guarantee a new reference.

asLiveData() is a bridge that wraps any Flow in a LiveData. Internally it creates an EmittedValueLiveData that starts the Flow collection in a coroutine when the LiveData becomes active (has observers) and cancels it when inactive. It uses a configurable timeout (default 5 seconds) before actually stopping collection, similar to WhileSubscribed(5000).

// Internal mechanics (simplified from androidx.lifecycle:lifecycle-livedata-ktx)
fun <T> Flow<T>.asLiveData(
    context: CoroutineContext = EmptyCoroutineContext,
    timeoutInMs: Long = 5000L
): LiveData<T> = liveData(context, timeoutInMs) {
    collect { emit(it) }  // liveData builder collects and emits to LiveData
}

// The liveData{} builder is backed by CoroutineLiveData which:
// 1. Starts coroutine when LiveData becomes ACTIVE (first observer)
// 2. Cancels after timeoutInMs when LiveData becomes INACTIVE (no observers)
// 3. Re-starts collection from scratch when ACTIVE again (cold Flow restarts)
// For StateFlow/SharedFlow (hot), re-start just re-subscribes — no duplicate work

// USE CASE 1: Gradual migration from LiveData codebase
class ProfileViewModel : ViewModel() {
    private val _user = MutableStateFlow<User?>(null)

    // For new Compose screens — collect directly
    val user: StateFlow<User?> = _user.asStateFlow()

    // For legacy XML screens — bridge to LiveData
    val userLive: LiveData<User?> = _user.asLiveData()
}

// USE CASE 2: XML data binding — only LiveData works with 2-way binding
// <TextView android:text="@{viewModel.userLive.name}" />  ← needs LiveData

// USE CASE 3: When you need Transformations.switchMap
val combined: LiveData<String> = Transformations.switchMap(userIdLive) { id ->
    userRepo.getUser(id).asLiveData()  // asLiveData() inside switchMap
}

// WHEN TO PREFER direct StateFlow collection in Compose:
val user by viewModel.user.collectAsStateWithLifecycle()
// ✅ No boxing through LiveData, no 5s timeout overhead
// ✅ Compose recomposition is already smart — no need for LiveData's observer model
// ✅ collectAsStateWithLifecycle handles lifecycle safety natively

Prefer asLiveData() when: you have XML layouts with Data Binding, you need Transformations.switchMap(), or you're in a mixed migration where some screens are still View-based. For pure Compose codebases, collect StateFlow directly with collectAsStateWithLifecycle() — it's more efficient and avoids the LiveData allocation.

Custom Views don't have a LifecycleOwner natively, but they can get one via findViewTreeLifecycleOwner() (API 28+ / AndroidX). This lets the View observe StateFlow safely without holding a permanent scope reference.

class LiveScoreView @JvmOverloads constructor(
    context: Context,
    attrs: AttributeSet? = null
) : View(context, attrs) {

    private var collectionJob: Job? = null
    private var scoreFlow: StateFlow<Int>? = null
    private var currentScore = 0

    fun bind(flow: StateFlow<Int>) {
        scoreFlow = flow
        // If already attached, start collecting immediately
        if (isAttachedToWindow()) startCollecting()
    }

    override fun onAttachedToWindow() {
        super.onAttachedToWindow()
        startCollecting()
    }

    override fun onDetachedFromWindow() {
        super.onDetachedFromWindow()
        collectionJob?.cancel()
        collectionJob = null
    }

    private fun startCollecting() {
        val flow = scoreFlow ?: return
        // Get lifecycle from the host Fragment/Activity
        val lifecycle = findViewTreeLifecycleOwner()?.lifecycle ?: return
        val scope = findViewTreeLifecycleOwner()?.lifecycleScope ?: return

        collectionJob?.cancel()
        collectionJob = scope.launch {
            lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
                flow.collect { score ->
                    currentScore = score
                    invalidate()  // trigger redraw
                }
            }
        }
    }

    override fun onDraw(canvas: Canvas) {
        super.onDraw(canvas)
        // Draw currentScore on canvas
        canvas.drawText(currentScore.toString(), 50f, 50f, paint)
    }
}

// Usage in Fragment:
override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
    binding.scoreView.bind(viewModel.score)
    // No need to manually cancel — View handles its own lifecycle
}

The lifecycle pairing is: onAttachedToWindow → start collection, onDetachedFromWindow → cancel job. The inner repeatOnLifecycle(STARTED) handles app backgrounding — collection pauses when the host goes to STOPPED (e.g., another Activity comes on top) and resumes when it returns to STARTED. This prevents wasted work off-screen and is leak-free.

Collaborative editing requires operational transforms or CRDTs. In Android, you model it with: local optimistic state (StateFlow), outgoing operations (SharedFlow to server), and incoming server patches (SharedFlow from server) that reconcile conflicts.

data class DocumentState(
    val content: String,
    val version: Int,
    val pendingOps: List<TextOp> = emptyList()
)

sealed class TextOp {
    data class Insert(val pos: Int, val text: String, val opId: String) : TextOp()
    data class Delete(val pos: Int, val length: Int, val opId: String) : TextOp()
}

class CollabDocViewModel @Inject constructor(
    private val docRepo: DocumentRepository
) : ViewModel() {

    private val _docState = MutableStateFlow(DocumentState("", 0))
    val docState: StateFlow<DocumentState> = _docState.asStateFlow()

    // Cursor positions of other collaborators
    private val _collaboratorCursors = MutableStateFlow<Map<String, Int>>(emptyMap())
    val collaboratorCursors: StateFlow<Map<String, Int>> = _collaboratorCursors.asStateFlow()

    init {
        // Observe server patches and reconcile
        viewModelScope.launch {
            docRepo.serverPatches()  // SharedFlow from WebSocket
                .collect { patch ->
                    when (patch) {
                        is ServerPatch.Ack -> {
                            // Server confirmed our op — remove from pending
                            _docState.update { state ->
                                state.copy(
                                    pendingOps = state.pendingOps.filter { it.opId != patch.opId },
                                    version = patch.newVersion
                                )
                            }
                        }
                        is ServerPatch.RemoteOp -> {
                            // Transform remote op against our pending ops (OT algorithm)
                            _docState.update { state ->
                                val transformed = transformOp(patch.op, state.pendingOps)
                                state.copy(
                                    content = applyOp(state.content, transformed),
                                    version = patch.newVersion
                                )
                            }
                        }
                        is ServerPatch.CursorUpdate -> {
                            _collaboratorCursors.update { cursors ->
                                cursors + (patch.userId to patch.position)
                            }
                        }
                    }
                }
        }
    }

    // Local edit — apply optimistically, queue for server
    fun insertText(position: Int, text: String) {
        val op = TextOp.Insert(position, text, generateOpId())
        _docState.update { state ->
            state.copy(
                content = applyOp(state.content, op),
                pendingOps = state.pendingOps + op
            )
        }
        viewModelScope.launch {
            docRepo.sendOp(op, _docState.value.version)
        }
    }
}

The key design principle: StateFlow holds the "ground truth" local state including pending (unacknowledged) operations. Server Acks remove from pending; server RemoteOps are transformed against pending before application. If the server rejects an operation, roll back by re-applying all remaining pending ops to the server-confirmed base state.

This is a comprehensive review question that tests multiple Flow concepts simultaneously. The buggy code contains 6 distinct issues.

// ❌ BUGGY CODE (submitted for review):
class SearchViewModel(private val repo: SearchRepository) : ViewModel() {
    val query = MutableStateFlow("")

    val results: LiveData<List<Result>> = query  // Bug 1
        .debounce(300)
        .flatMapLatest { q -> repo.search(q) }  // Bug 2
        .catch { /* swallowed */ }  // Bug 3
        .asLiveData()  // Bug 4

    fun onQueryChange(q: String) {
        viewModelScope.launch(Dispatchers.IO) {  // Bug 5
            query.emit(q)
        }
    }
}

// In Fragment:
viewModel.results.observe(this) { results ->  // Bug 6
    adapter.submitList(results)
}

// ✅ CORRECTED VERSION:
class SearchViewModel(private val repo: SearchRepository) : ViewModel() {

    // Fix 1: Keep MutableStateFlow private, expose immutable
    private val _query = MutableStateFlow("")
    val query: StateFlow<String> = _query.asStateFlow()

    // Fix 2 + 4: Use StateFlow instead of LiveData; handle empty query
    val results: StateFlow<SearchState> = _query
        .debounce(300)
        .distinctUntilChanged()  // skip if query unchanged after debounce
        .flatMapLatest { q ->
            if (q.isBlank()) flowOf(SearchState.Empty)
            else repo.search(q)
                .map { SearchState.Success(it) as SearchState }
                .onStart { emit(SearchState.Loading) }
        }
        // Fix 3: Emit error state instead of swallowing
        .catch { e -> emit(SearchState.Error(e.message ?: "Unknown error")) }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5000),
            initialValue = SearchState.Empty
        )

    // Fix 5: StateFlow.value is thread-safe — no coroutine needed
    fun onQueryChange(q: String) {
        _query.value = q  // atomic, thread-safe, no coroutine needed
    }
}

// Fix 6: In Fragment — use lifecycle-safe collection, not observe()
viewLifecycleOwner.lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.results.collect { state ->
            when (state) {
                is SearchState.Success -> adapter.submitList(state.items)
                is SearchState.Loading -> showShimmer()
                is SearchState.Error -> showError(state.message)
                SearchState.Empty -> showEmptyState()
            }
        }
    }
}

Summary of all 6 bugs: (1) MutableStateFlow exposed publicly allows external mutation. (2) flatMapLatest on blank query hits the network unnecessarily. (3) catch{} with no emit swallows errors silently — UI stays stuck on last result. (4) asLiveData() loses WhileSubscribed optimization — use stateIn instead. (5) StateFlow.value is already thread-safe — launching a coroutine on IO just to emit() adds unnecessary overhead. (6) LiveData.observe(this) in a Fragment leaks if this is the Fragment not viewLifecycleOwner, and doesn't handle the STARTED/STOPPED lifecycle correctly for StateFlow.