Intermediate Operators

Example

Values flow through the chain one at a time. take(3) stops the upstream after the third value.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    (1..10).asFlow()
        .filter { it % 2 == 0 }
        .map { it * it }
        .onEach { println("onEach: $it") }
        .take(3)
        .collect { println("collect: $it") }
}

Output

onEach: 4
collect: 4
onEach: 16
collect: 16
onEach: 36
collect: 36

Example

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    // transform can emit 0, 1 or many values per input
    flowOf("order-1", "order-2")
        .transform { id ->
            emit("$id: placed")
            emit("$id: packed")
        }
        .collect { println(it) }

    // ignore repeated values in a row
    val battery = flowOf(80, 80, 79, 79, 79, 78, 80)
    println(battery.distinctUntilChanged().toList())

    // running total, emits every step
    println(flowOf(100, 250, 50).scan(0) { acc, x -> acc + x }.toList())

    // stop when a condition is false
    println(flowOf(5, 3, 0, 7).takeWhile { it > 0 }.toList())
}

Output

order-1: placed
order-1: packed
order-2: placed
order-2: packed
[80, 79, 78, 80]
[0, 100, 350, 400]
[5, 3]

Example

debounce(300) emits a value only when no new value came for 300 ms. The user types fast, so only "sho" and the final "shoes" reach the API.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

@OptIn(FlowPreview::class)
fun main() = runBlocking {
    val typed = flow {
        emit("s"); delay(50)
        emit("sh"); delay(50)
        emit("sho"); delay(500)        // user pauses
        emit("shoe"); delay(50)
        emit("shoes")
    }
    typed.debounce(300).collect { println("Search API called for: $it") }
}

Output

Search API called for: sho
Search API called for: shoes