Kotlin Flow Operators — Detailed Guide

Flow operators help you transform, combine, control, and manage data streams. Operators are divided into:

1. Terminal Operators

2. Non-Terminal Operators

Common Flow Operators (Detailed with Examples)

1. onStart

Runs BEFORE the flow starts emitting values. Useful for loading states, logs, or counters.

Example 1

flow {
    emit(1)
    emit(2)
}.onStart {
    println("Flow started")
}.collect { println(it) }
Output: 
Flow started
1
2

Example 2

fun fetchData() = flow {
    delay(300)
    emit("Data Loaded")
}

fetchData()
    .onStart { println("Loading...") }
    .collect { println(it) }
Output: 
Loading...
Data Loaded

2. onCompletion

Runs AFTER the flow completes — even if an exception occurs.

Example 1

flow {
    emit(10)
    emit(20)
}
.onCompletion { println("Finished") }
.collect { println(it) }
Output: 
10
20
Finished

Example 2 (exception)

flow {
    emit(1)
    throw RuntimeException("Boom!")
}
.onCompletion { println("Completed (error or success)") }
.catch { println("Caught: ${'$'}it") }
.collect { println(it) }
Output: 
1
Completed (error or success)
Caught: java.lang.RuntimeException: Boom!

3. catch

Catches upstream exceptions. Does NOT catch exceptions inside collect.

Example 1

flow {
    emit(1)
    throw Exception("Error")
}.catch { emit(0) }
.collect { println(it) }
Output: 
1
0

Example 2

flow {
    throw IllegalStateException("Invalid")
}
.catch { println("Caught: ${'$'}it") }
.collect()
Output: 
Caught: java.lang.IllegalStateException: Invalid

4. collectLatest

Cancels previous collector block when a new value arrives. Useful for search boxes, typing events.

Example 1

(1..3).asFlow()
    .onEach { delay(200) }
    .collectLatest { 
        println("Collecting ${'$'}it")
        delay(300)
    }
Output: 
Collecting 1   (cancelled by 2)
Collecting 2   (cancelled by 3)
Collecting 3

Example 2 (typing simulation)

flowOf("h", "he", "hel", "hell", "hello")
    .onEach { delay(100) }
    .collectLatest { println("Searching for: ${'$'}it") }
Output: 
Searching for: h   (cancelled)
Searching for: he  (cancelled)
Searching for: hel (cancelled)
Searching for: hell(cancelled)
Searching for: hello

5. zip

Combines two flows by pairing items by index. Stops when shorter flow ends.

Example 1

val f1 = flowOf(1, 2, 3)
val f2 = flowOf("A", "B")

f1.zip(f2) { n, c -> "${'$'}n -> ${'$'}c" }
.collect { println(it) }
Output: 
1 -> A
2 -> B

Example 2

(1..3).asFlow().zip((10..30 step 10).asFlow()) { a, b ->
    a + b
}.collect { println(it) }
Output: 
11
22
33

6. combine

Emits whenever ANY flow emits, using the latest values from both. Unlike zip, combine does NOT require equal lengths.

Example 1

val f1 = flowOf(1, 2)
val f2 = flowOf("A", "B", "C")

f1.combine(f2) { a, b -> "${'$'}a + ${'$'}b" }
.collect { println(it) }
Output: 
1 + A
1 + B
1 + C
2 + C

Example 2

val numbers = (1..3).asFlow().onEach { delay(100) }
val letters = flowOf("X", "Y").onEach { delay(200) }

numbers.combine(letters) { a, b -> "${'$'}a & ${'$'}b" }
.collect { println(it) }
Output: 
1 & X
2 & X
3 & X
3 & Y

7. flattenMerge

Merges inner flows concurrently, emitting as soon as values are available.

Example 1

flowOf(
    flow { delay(200); emit("A") },
    flow { delay(100); emit("B") }
).flattenMerge()
.collect { println(it) }
Output: 
B
A

Example 2

listOf(1, 2, 3).map {
    flow {
        delay(300 - it * 50)
        emit("Flow ${'$'}it")
    }
}.asFlow().flattenMerge()
.collect { println(it) }
Output: 
Flow 3
Flow 2
Flow 1

8. flattenConcat

Collects flows sequentially. Unlike merge, it waits for one to finish before starting the next.

Example 1

flowOf(
    flow { emit("A1"); emit("A2") },
    flow { emit("B1") }
).flattenConcat()
.collect { println(it) }
Output: 
A1
A2
B1

Example 2

listOf(
    flow { delay(100); emit(1) },
    flow { delay(50); emit(2) }
).asFlow().flattenConcat()
.collect { println(it) }
Output: 
1
2

9. map

Transforms each emitted value.

Example 1

(1..3).asFlow()
    .map { it * 2 }
    .collect { println(it) }
Output: 
2
4
6

Example 2

flowOf("a", "bb", "ccc")
    .map { it.length }
    .collect { println(it) }
Output: 
1
2
3

10. filter

Filters out unwanted values.

Example 1

(1..5).asFlow()
    .filter { it % 2 == 0 }
    .collect { println(it) }
Output: 
2
4

Example 2

flowOf("Dog", "Cat", "Apple")
    .filter { it.startsWith("A") }
    .collect { println(it) }
Output: 
Apple

11. take

Stops after emitting N values.

Example 1

(1..5).asFlow().take(3)
.collect { println(it) }
Output: 
1
2
3

Example 2

flow {
    emit("A")
    emit("B")
    emit("C")
}.take(2)
.collect { println(it) }
Output: 
A
B

12. drop

Skips first N values.

Example 1

(1..5).asFlow().drop(2)
.collect { println(it) }
Output: 
3
4
5

Example 2

flowOf("A", "B", "C", "D")
    .drop(1)
    .collect { println(it) }
Output: 
B
C
D

13. reduce (Terminal Operator)

Reduces the flow into a single value.

Example 1

(1..4).asFlow()
    .reduce { acc, value -> acc + value }
Output: 
10

Example 2

flowOf("A", "B", "C")
    .reduce { acc, value -> acc + value }
Output: 
ABC

14. fold (Terminal Operator)

Same as reduce but starts with an initial value.

Example 1

(1..3).asFlow()
    .fold(10) { acc, value -> acc + value }
Output: 
16

Example 2

flowOf("A", "B")
    .fold("Start: ") { acc, value -> acc + value }
Output: 
Start: AB

15. asFlow

Converts collections, ranges, or sequences into flows.

Example 1

listOf(1, 2, 3).asFlow()
    .collect { println(it) }
Output: 
1
2
3

Example 2

(10..15).asFlow()
    .filter { it > 12 }
    .collect { println(it) }
Output: 
13
14
15

Summary

This article covered: