Irbisa · Cheatsheet

Kotlin Flow

Cold and hot streams, operators, StateFlow and SharedFlow, Channels, flowOn, backpressure, stateIn and shareIn, error handling

Middle Developer · 20 questions

  1. 01

    What is Flow, and what is the difference between a cold and a hot stream?

    Flow is an asynchronous stream of values built on suspend functions.

    val numbers = flow {  println("producer started")  emit(1)  emit(2)}…
  2. 02

    What is the difference between Flow and Sequence, and why can't you call delay inside a sequence { } builder?

    Both produce values lazily, one at a time, but a Sequence is synchronous: the consumer pulls each value with next() and its thread runs the producer until the next yield.

    // Sequence: synchronous, the consumer pulls each valueval fib = sequence {  var a = 0  var b = 1  while (true) {    yield(a)                 // pauses until the next value is pulled…
  3. 03

    Which Flow operators are terminal, why does a chain that ends in onEach { } do nothing, and what does launchIn(scope) return?

    Terminal operators start the collection: collect, first, toList, fold and the like are suspend functions, and launchIn collects in a new coroutine.

    val prices = flowOf(1, 2, 3)  .onEach { println("got $it") }  .map { it * 10 }                        // nothing runs yet: no terminal operator suspend fun demo(scope: CoroutineScope) {  println(prices.first())                 // got 1, 10: 2 and 3 are never produced…
  4. 04

    What is the difference between StateFlow and SharedFlow, and when do you use each?

    Use StateFlow for screen state: it always holds one current value, needs an initial value and skips equal values.

    class CartVM : ViewModel() {  private val _state = MutableStateFlow(CartState())  val state: StateFlow<CartState> = _state.asStateFlow()   private val _messages = MutableSharedFlow<String>(extraBufferCapacity = 1)  val messages: SharedFlow<String> = _messages.asSharedFlow()…
  5. 05

    Find the bug: tapping "Add" puts the item into the cart, but the list on screen never updates.

    class CartViewModel : ViewModel() {  private val items = mutableListOf<Item>()  private val _cart = MutableStateFlow<List<Item>>(items)  val cart: StateFlow<List<Item>> = _cart.asStateFlow()   fun add(item: Item) {    items.add(item)    _cart.value = items  }}

    StateFlow drops a new value that equals the current one, and _cart.value = items assigns the very list it already holds.

  6. 06

    Find the bug: the screen shows the user's name, but the unread count collected right below it never appears.

    class InboxFragment : Fragment(R.layout.fragment_inbox) {  private val vm: InboxViewModel by viewModels()   override fun onViewCreated(view: View, savedInstanceState: Bundle?) {    val binding = FragmentInboxBinding.bind(view)    viewLifecycleOwner.lifecycleScope.launch {      viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {        vm.userName.collect { binding.name.text = it }          // StateFlow<String>        vm.unreadCount.collect { binding.badge.text = "$it" }   // StateFlow<Int>      }    }  }}

    collect on a StateFlow never returns, because a hot flow never completes, so the coroutine stays on the first line and the second collect is never reached.

  7. 07

    What is a Channel, how does it differ from a Flow, and what do the RENDEZVOUS, BUFFERED, CONFLATED and UNLIMITED capacities change for send?

    A Channel is a hot queue between coroutines: send puts an element in, receive takes it out, and each element reaches exactly one receiver.

    fun main() = runBlocking {  // A work queue: each job goes to exactly one worker  val jobs = Channel<Int>(Channel.BUFFERED)  repeat(2) { worker ->    launch {      for (job in jobs) {                 // ends when the channel is closed and drained…
  8. 08

    On which thread does each numbered step of this flow run, and how does flowOn work?

    // launched from viewModelScope (Dispatchers.Main)repo.rows()                          // flow { emit(db.readAll()) }  .map { it.toUi() }                 // (1)  .flowOn(Dispatchers.IO)  .filter { it.isNotEmpty() }        // (2)  .flowOn(Dispatchers.Default)  .collect { render(it) }            // (3)

    flowOn changes the context only for the operators above it, up to the previous flowOn.

  9. 09

    Find the bug: this flow loads two pages in parallel and crashes with IllegalStateException: Flow invariant is violated.

    fun feed(api: FeedApi): Flow<List<Post>> = flow {  coroutineScope {    launch { emit(api.page(1)) }   // each page in its own coroutine    launch { emit(api.page(2)) }  }}

    emit is called from coroutines started with launch, not from the coroutine that collects the flow, and flow { } forbids that: "Emission from another coroutine is detected".

  10. 10

    What does this code print with zip, and what with combine? When would you use merge instead?

    val numbers = flow { for (n in 1..3) { delay(100); emit(n) } }        // at 100, 200, 300 msval letters = flow { delay(150); emit("A"); delay(100); emit("B") }   // at 150, 250 ms fun main() = runBlocking {  numbers.zip(letters) { n, l -> "$n$l" }.collect { print("$it ") }  println()  numbers.combine(letters) { n, l -> "$n$l" }.collect { print("$it ") }}

    zip prints 1A 2B: it pairs values by position and stops as soon as one flow ends, so 3 is never paired.

  11. 11

    flatMapConcat, flatMapMerge or flatMapLatest: which do you use for search-as-you-type, and what else does that pipeline need?

    flatMapLatest: when a new query arrives it cancels the request for the old one, so stale results cannot overwrite fresh ones.

    @OptIn(FlowPreview::class, ExperimentalCoroutinesApi::class)class SearchVM(private val repo: SearchRepo) : ViewModel() {  private val query = MutableStateFlow("")  fun onQueryChange(text: String) { query.value = text }   val results: StateFlow<SearchState> = query…
  12. 12

    What is the difference between debounce, sample and throttle-first, and which one fits a search box, a sensor stream and a Pay button?

    debounce(300) emits a value only after 300 ms pass with no newer one, sample(300) emits the newest value once every 300 ms, and throttle-first emits the first value and ignores the rest for a window.

    // kotlinx.coroutines has no throttleFirst: a few lines with flow { }fun <T> Flow<T>.throttleFirst(window: Duration): Flow<T> = flow {  var opened: TimeMark? = null              // per collector: state lives inside flow { }  collect { value ->    val mark = opened    if (mark == null || mark.elapsedNow() >= window) {…
  13. 13

    A Flow emits faster than the collector can process. What happens by default, and how do buffer, conflate and collectLatest change it?

    By default nothing is lost: emit suspends until the collector finishes the previous value, so the producer slows down to the collector's pace.

    val source = flow {  repeat(3) { delay(100); emit(it) }   // 100 ms to produce each} suspend fun compare() {  // default: about 900 ms, producer waits for the collector…
  14. 14

    Why doesn't catch see the exception thrown in your collect block, and where do retry and catch go in a flow chain?

    catch only handles exceptions from upstream (the operators above it), so an exception in collect { } is below it and goes straight to the collecting coroutine.

    class FeedVM(private val repo: FeedRepo) : ViewModel() {  val feed: StateFlow<FeedState> = repo.feedFlow()    .retryWhen { cause, attempt ->      if (cause !is IOException || attempt >= 3) return@retryWhen false      delay(500L * (1 shl attempt.toInt()))          // 0.5 s, 1 s, 2 s      true                                           // re-collect the upstream…
  15. 15

    How do onStart, onCompletion and onEmpty work, and where do you clear a loading flag so it also clears on error and cancellation?

    onStart runs before the upstream starts and can emit a first value, onEmpty runs when the upstream completes without emitting anything, and onCompletion runs after the flow ends for any reason: success, an exception or cancellation.

    class SyncController(private val repo: SyncRepository, private val scope: CoroutineScope) {  val state = MutableStateFlow(SyncUiState())   fun start(): Job = repo.sync()                    // Flow<SyncProgress>, completes when done    .onStart { state.update { it.copy(isSyncing = true, error = null) } }    .onEach { p -> state.update { it.copy(percent = p.percent) } }…
  16. 16

    When should a repository function return Flow<T>, and when a suspend fun that returns T?

    Return Flow<T> when the data can change while someone is looking at it and the caller should see each change: a Room query, a DataStore preference, a socket.

  17. 17

    What do stateIn and shareIn do, and what does SharingStarted.WhileSubscribed(5_000) actually control?

    stateIn and shareIn collect a cold upstream once and re-broadcast it to many collectors; SharingStarted decides when that collection starts and stops.

    class NotesVM(private val repo: NoteRepo) : ViewModel() {  // right: built once, one upstream collection for the ViewModel's life  val notes: StateFlow<List<Note>> = repo.notesFlow()    .stateIn(      scope = viewModelScope,      started = SharingStarted.WhileSubscribed(5_000),…
  18. 18

    Where should a screen's first load start: in the ViewModel's init { }, in a LaunchedEffect, or in a flow exposed with stateIn? What does each cost?

    Usually inside a flow exposed with stateIn(viewModelScope, WhileSubscribed(5_000), initial): the load starts when the screen first collects, survives rotation and stops when nobody watches.

    class ProfileViewModel(private val repo: ProfileRepository) : ViewModel() {  private val retries = MutableStateFlow(0)   @OptIn(ExperimentalCoroutinesApi::class)  val state: StateFlow<ProfileUiState> = retries    .flatMapLatest {…
  19. 19

    One slow subscriber of a MutableSharedFlow event bus stalls the other subscribers and the emitter. Why, and what do extraBufferCapacity and onBufferOverflow change?

    A SharedFlow has one buffer for all subscribers, and a slot is freed only once the slowest subscriber has read it.

    fun main() = runBlocking {  val bus = MutableSharedFlow<Int>()             // replay 0, extraBufferCapacity 0  launch { bus.collect { delay(300); println("slow got $it") } }  launch { bus.collect { println("fast got $it") } }  delay(50)                                      // let both subscribe  repeat(4) { bus.emit(it) }                     // returns only after slow takes 3…
  20. 20

    How is Flow implemented under the hood (Flow, FlowCollector), and how would you write your own operator?

    Flow is an interface with one function, suspend fun collect(collector: FlowCollector<T>), and FlowCollector has one too, suspend fun emit(value: T).

    // The whole contract, simplified from kotlinx.coroutines:// interface Flow<out T> { suspend fun collect(collector: FlowCollector<T>) }// fun interface FlowCollector<in T> { suspend fun emit(value: T) } // A custom operator: each value paired with the one before itfun <T : Any> Flow<T>.pairwise(): Flow<Pair<T, T>> = flow {…