Flow operators: map, filter, onEach, catch, and backpressure

1. Why Flow operators are needed

When you see Flow for the first time, your hand reaches for this: “well, I know for and if — so I’ll just write everything inside collect { ... }”. It works, but it usually turns into one huge handler where filtering, transformation, debug printing, and error handling are all mixed together. A week later this code looks like: “I don’t remember what I meant by this if”.

Flow operators help keep your head clear: each pipeline step is responsible for one simple idea. It’s almost like a factory: first parts arrive, then go through a quality filter, then get labeled, then get put into a box, and only at the end do we get the finished product.

The diagram we’ll aim for today looks like this:

flowchart LR
    A["Source: flow { emit(...) }"] --> B["filter {...}"]
    B --> C["map {...}"]
    C --> D["onEach {...}"]
    D --> E["catch {...}"]
    E --> F["collect {...}"]

2. Example: a stream of monitoring events

To keep the examples connected, let’s build a small “console monitoring system”: we have events (for example, "INFO", "WARN", "ERROR"), we want to pass through only important ones, transform them into readable text, sometimes print debug info, and avoid crashing the whole application because of one parsing error.

Let’s start with the event model:

data class MonitorEvent(
    val level: String,
    val message: String
)

And let’s also agree on the “raw string” format, as if it came from a log: "INFO|Service started" or "ERROR|Disk is full".

3. Basic Flow: source and collect

Before operators, it helps to feel the “bare pipe”: there is a source that emits values, and there is a collector. Flow runs only on collect (that is, the stream is essentially lazy / cold).

Here is a minimal “raw text” stream:

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    val lines = flow {
        emit("INFO|Service started")
        emit("WARN|Cache miss")
    }

    lines.collect { line ->
        println("raw: $line") // raw: INFO|Service started ...
    }
}

So far this is not an “analyzer”, it is just delivering values. Now we will start turning the stream into a pipeline.

4. Transformations: map, filter, onEach

map: transform and change the type

map in Flow is very similar in meaning to map on collections: take each element and turn it into something new. With Flow the idea is the same, only elements arrive “over time”, and the transformation can be suspend-friendly.

Let’s write a parser from a string to a MonitorEvent. For simplicity, without Regex, just a regular split.

fun parseLine(line: String): MonitorEvent {
    val parts = line.split("|", limit = 2)
    val level = parts[0]
    val message = parts.getOrElse(1) { "" }
    return MonitorEvent(level = level, message = message)
}

Now let’s apply map:

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    flow {
        emit("INFO|Service started")
        emit("ERROR|Disk is full")
    }
        .map { parseLine(it) }
        .collect { e ->
            println("${e.level}: ${e.message}")
            // INFO: Service started
            // ERROR: Disk is full
        }
}

Notice an important thing: the stream type changed. It was Flow<String>, and became Flow<MonitorEvent>. That’s the superpower of map: it can change the type, and that makes the pipeline much more expressive.

A small reading tip: if you get lost and wonder “what type is the stream right now?”, remember that map changes the type, while filter usually does not.

filter: keep only what you need

Now imagine there are many events, but we only care about "WARN" and "ERROR". That’s where filter comes in: selecting elements by a predicate (a lambda that returns true/false). For Flow the meaning is exactly the same: a value either passes further down the pipeline or is discarded.

Let’s filter after parsing:

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    flow {
        emit("INFO|Service started")
        emit("WARN|Cache miss")
        emit("ERROR|Disk is full")
    }
        .map { parseLine(it) }
        .filter { e -> e.level != "INFO" }
        .collect { e ->
            println("IMPORTANT: ${e.level} ${e.message}")
            // IMPORTANT: WARN Cache miss
            // IMPORTANT: ERROR Disk is full
        }
}

Here filter does not change the type: the stream was MonitorEvent and it stays that way. It only “narrows” the set of values.

onEach: side effects without changing the data

When you debug a stream, you want to “peek” at what is happening at each step. The first thought is to put println into map. But that’s poor style: map should transform data by meaning, and printing is a side effect.

onEach is exactly for that: perform an action for each element and pass the element onward unchanged. It’s like forEach, but in the middle of a flow pipeline.

We’ll add diagnostics: show what exactly passed the filter.

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    flow {
        emit("INFO|Service started")
        emit("WARN|Cache miss")
        emit("ERROR|Disk is full")
    }
        .map { parseLine(it) }
        .filter { it.level != "INFO" }
        .onEach { println("debug: passed filter -> $it") }
        .collect { e ->
            println("ALERT: ${e.message}")
        }
}

The nice thing about onEach is that it’s easy to remove when debugging is no longer needed, and it doesn’t “break” the meaning of map.

By the way, if you ever see println inside map, it’s not a mortal sin, but it is almost always a signal: “it’s time to clean this up a bit”.

5. Errors: catch and the boundaries of its effect

catch: error handling in the pipeline

Error handling is what separates a “textbook example” from a “program that lives longer than 15 seconds”. In Kotlin we already know try/catch and understand that an exception interrupts the normal execution flow. In Flow errors happen too, but it’s often more convenient to catch them right in the pipeline using the catch operator.

Let’s make the parser stricter: if the line does not contain "|", we treat it as a format error.

fun parseLineStrict(line: String): MonitorEvent {
    val parts = line.split("|", limit = 2)
    if (parts.size < 2) error("Bad format: '$line'")
    return MonitorEvent(level = parts[0], message = parts[1])
}

Now let’s use catch and “replace” the error with a special event:

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    flow {
        emit("INFO|Service started")
        emit("BROKEN LINE")
        emit("ERROR|Disk is full")
    }
        .map { parseLineStrict(it) }
        .catch { e ->
            println("handled error: ${e.message}") // handled error: Bad format: 'BROKEN LINE'
            emit(MonitorEvent("ERROR", "Parser failed, but app continues"))
        }
        .collect { e ->
            println("event: ${e.level} ${e.message}")
        }
}

Here it’s important to understand the mechanics: catch intercepts errors that occurred before it in the pipeline (upstream). That is, if an error is thrown in map { parseLineStrict(it) }, our catch will see it. But if an error happens later (for example, inside collect), then catch might not help.

Why catch does not catch an error inside collect

This is a common trap: you add catch, relax, and think the stream is now “unkillable”. But catch is not a magic umbrella against all trouble; it only catches what happens “up the pipe”.

Let’s create an artificial error inside collect:

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    flow {
        emit(1)
    }
        .catch { println("catch: ${it.message}") }
        .collect { _ ->
            error("boom in collect")
        }
}

This code will crash (unless you wrap collect in try/catch), because the error happened in the consumer. That is logical: collect is a terminal operation, and if the consumer itself fails, then it must be saved by an outer try/catch, just like in ordinary life.

Practical rule: catch is for errors in the source and operators before collect. Errors in the handler inside collect should be caught outside with a regular try/catch.

6. Backpressure in Flow

The word backpressure sounds like the name of a rock band (“Backpressure and their new album ‘Suspend Me Gently’”), but the meaning is very down-to-earth. If the consumer processes values slowly, you want the source not to keep producing values infinitely fast, otherwise we will either choke or start accumulating values in memory.

In Flow (in the basic model) there is a natural mechanism: emit and the whole pipeline are suspend-friendly, so if the consumer is slow, upstream will often pause and wait.

Let’s look at an example: the source emits numbers quickly, and collect processes them slowly via delay.

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    flow {
        repeat(3) { i ->
            emit(i)
            println("emitted $i") // printing goes in sync with processing
        }
    }.collect { v ->
        delay(100)
        println("collected $v")
        // emitted 0
        // collected 0
        // emitted 1
        // collected 1
        // emitted 2
        // collected 2
    }
}

If you expected to see "emitted 0, emitted 1, emitted 2" immediately, and then "collected …", that is exactly the key backpressure insight in Flow: the stream does not have to “fire” everything instantly; it can be naturally synchronized with the consumer via suspensions.

7. Assemble the pipeline into a neat flow

Now let’s assemble a mini version of our “monitor” so that the code reads like plain text: “take lines → parse → keep important → log → catch errors → print the result”.

First, let’s make a source function. In a real project, events come from somewhere external, but today a simulation is enough:

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow

fun rawLogLines(): Flow<String> = flow {
    emit("INFO|Service started")
    emit("WARN|Cache miss")
    emit("BROKEN LINE")
    emit("ERROR|Disk is full")
}

Then we’ll describe the pipeline as a function that returns Flow<MonitorEvent>. This is a good style: we separate “what to do” from “how to output”.

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.catch

fun importantEvents(): Flow<MonitorEvent> =
    rawLogLines()
        .map { parseLineStrict(it) }
        .filter { it.level != "INFO" }
        .onEach { println("trace: $it") }
        .catch { e ->
            emit(MonitorEvent("ERROR", "stream failed: ${e.message}"))
        }

And the final main, where everything comes together:

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    importantEvents().collect { e ->
        println("ALERT: ${e.level} -> ${e.message}")
    }
}

Notice how convenient it is to read this kind of code: in importantEvents() there is no terminal operation, which means it can be reused, tested, and plugged into different places. And main stays simple: “collect and print”.

8. Operator cheat sheet

When there are more operators, it helps to keep a “map” in your head. Let’s make a small table:

map { ... }

filter { ... }

onEach { ... }

catch { ... }

If you see a long chain, you can literally read it out loud: rawLogLines, map, filter, onEach, catch, collect. Usually that is already a good sign: the code looks like an explanation, not a riddle.

9. Common mistakes

Mistake No. 1: doing side effects in map, and then wondering why “the logic feels kinda murky”.
When map both transforms data and prints and calculates statistics, it stops being a “transformation” and becomes “a chunk of everything”. As a result, it is hard to understand what exactly changes and why. It’s much calmer to keep map as a pure transformation, and move printing and debugging into onEach.

Mistake No. 2: adding catch and thinking it catches absolutely everything.
catch catches errors that happened before it in the chain, i.e., upstream. If an error happened inside collect { ... }, that is downstream, and catch may not help. In such situations it’s better to wrap collect in a regular try/catch, as we used to do when learning exceptions.

Mistake No. 3: “swallowing” an exception in catch and continuing as if nothing happened, with no signal.
An empty catch {} is like taping over the “Check Engine” light: it does not make the car better, but the driver stops understanding what is going on. Even if you produce a fallback value via emit, it’s useful to at least leave an error message or explicitly form an “error event” so the behavior is predictable.

Mistake No. 4: building the whole pipeline inside collect and ending up with an 80-line “combine harvester”.
Such code is hard to test and hard to reuse: the logic is tied to the output location. Most often it’s better to extract the pipeline into a function that returns Flow<...>, and leave only collect in main. This makes the code simpler and structurally very similar to ordinary collection-processing functions, just over time.

Mistake No. 5: expecting Flow to “spray” values faster than the consumer can process them, and not understanding why everything is slow.
This is not a bug: the basic idea of backpressure is that Flow can naturally adapt to processing speed via suspend. If there is a delay in collect, then upstream will often “wait” — and that is exactly what helps avoid memory blow-ups and event avalanches.

1

Task

Kotlin SELF, level 54, lesson 3

Locked

Energy Squares

Energy Squares

Open

1

Task

Kotlin SELF, level 54, lesson 3

Locked

Sensor filter

Sensor filter

Open

1

Task

Kotlin SELF, level 54, lesson 3

Locked

Command debugging

Command debugging

Open

1

Task

Kotlin SELF, level 54, lesson 3

Locked

Log parser

Log parser

Open