Kotlin Flow Cheat Sheet

import kotlinx.coroutines.flow.*import kotlinx.coroutines.*fun numbers(): Flow<Int> = flow {    for (i in 1..5) {        delay(100)          // suspend inside the builder is fine        emit(i)               // push a value downstream    }}suspend fun main() {    numbers().collect { value ->        println(value)          // 1, 2, 3, 4, 5 -- printed as they're emitted    }}// flowOf and asFlow for simple sourcesflowOf(1, 2, 3)listOf(1, 2, 3).asFlow()

numbers()    .map { it * it }    .filter { it % 2 == 0 }    .take(3)    .onEach { println("saw $it") }    .catch { e -> println("caught: $e") }     // catches upstream exceptions    .collect { println("result: $it") }// flatMapConcat / flatMapMerge / flatMapLatest for flows-of-flowssearchQuery    .debounce(300)    .flatMapLatest { query -> searchApi(query) }   // cancels previous search on new input    .collect { results -> render(results) }

fun heavyWork(): Flow<Int> = flow {    for (i in 1..5) {        Thread.sleep(50)      // pretend CPU-bound work        emit(i)    }}.flowOn(Dispatchers.Default)   // upstream (the flow{} builder) runs on Defaultsuspend fun run() {    withContext(Dispatchers.Main) {        heavyWork().collect { value ->            updateUI(value)      // collect lambda runs on Main, as expected        }    }}

class CounterViewModel : ViewModel() {    private val _count = MutableStateFlow(0)      // requires an initial value    val count: StateFlow<Int> = _count.asStateFlow()    fun increment() {        _count.update { it + 1 }        // atomic update    }}class EventBus {    private val _events = MutableSharedFlow<String>(replay = 0, extraBufferCapacity = 1)    val events: SharedFlow<String> = _events.asSharedFlow()    suspend fun publish(event: String) = _events.emit(event)}// Collecting StateFlow in Composeval count by viewModel.count.collectAsStateWithLifecycle()

val name: Flow<String> = flowOf("Ada", "Grace")val age: Flow<Int> = flowOf(30, 40)// combine: recomputes with the LATEST value from each flow on every emissionname.combine(age) { n, a -> "$n is $a" }    .collect { println(it) }// zip: pairs by index, completes when the shorter flow completesname.zip(age) { n, a -> "$n/$a" }    .collect { println(it) }// merge: interleaves emissions from multiple flows of the same typeval fast = flow { emit(1); delay(10); emit(2) }val slow = flow { delay(5); emit(100) }merge(fast, slow).collect { println(it) } // order depends on timing

flow {    for (i in 1..100) {        delay(10)        emit(i)          // fast producer    }}.buffer(capacity = 50)                       // decouples producer/consumer with a channel.collect { slowConsume(it) }                   // producer no longer waits on consumer// conflate: drops intermediate values, keeps only the latest when consumer is behindsensorReadings.conflate().collect { render(it) }// collectLatest: cancels the in-progress collector block when a new value arrivessearchQuery.collectLatest { query ->    val results = searchApi(query) // canceled mid-flight if `query` changes again    render(results)}

fun locationUpdates(client: LocationClient): Flow<Location> = callbackFlow {    val callback = object : LocationCallback() {        override fun onLocationChanged(loc: Location) {            trySend(loc)                 // non-suspending, safe from any thread                .onFailure { /* channel closed/full */ }        }    }    client.requestUpdates(callback)    awaitClose {        client.removeUpdates(callback)  // guaranteed to run when the collector cancels    }}// Contrast with the simpler `flow { }` builder: callbackFlow backs onto a// Channel, so trySend() can be called from a non-suspending callback thread.

fun fetchData(): Flow<Data> = flow {    emit(api.getData())     // throws on network failure}.retryWhen { cause, attempt ->    if (cause is IOException && attempt < 3) {        delay(1000L * (attempt + 1)) // simple linear backoff        true                            // true = retry, false = rethrow    } else {        false    }}.catch { e -> emit(Data.Empty) }   // final fallback after retries are exhausted// Exception transparency rule: never try/catch around emit() inside a flow{}// builder -- it breaks cancellation semantics. Let exceptions propagate and// handle them with .catch{} downstream instead.