Оператори Flow: map, filter, onEach, catch і backpressure

1. Навіщо потрібні Flow-оператори

Коли ви бачите Flow уперше, рука так і тягнеться зробити найпростіше: «ну я ж умію for і if — отже, напишу все всередині collect { ... }». Це працює. Але зазвичай дуже швидко перетворюється на один величезний обробник, у якому змішані фільтрація, перетворення, налагоджувальний друк і обробка помилок. А за тиждень цей код виглядає вже так: «я не памʼятаю, що хотів сказати цим if».

Оператори Flow допомагають тримати думки в порядку: кожен крок конвеєра відповідає за одну просту ідею. Виходить майже як на виробництві: спочатку деталі надходять, потім проходять фільтр якості, потім їх маркують, далі складають у коробку — і лише наприкінці ми отримуємо готовий продукт.

Схема, до якої ми сьогодні прагнутимемо, виглядає так:

flowchart LR
    A["Джерело flow { emit(...) }"] --> B["filter {...}"]
    B --> C["map {...}"]
    C --> D["onEach {...}"]
    D --> E["catch {...}"]
    E --> F["collect {...}"]

2. Приклад: потік подій моніторингу

Щоб приклади були повʼязані між собою, зробімо маленьку «консольну систему моніторингу». У нас є події (наприклад, "INFO", "WARN", "ERROR"). Ми хочемо пропускати лише важливі, перетворювати їх на зрозумілий текст, інколи друкувати діагностику й не «класти» весь застосунок через одну помилку парсингу.

Почнемо з моделі події:

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

І ще домовимося про формат «сирого рядка» — ніби він прийшов із лога: "INFO|Сервіс запущено" або "ERROR|Диск заповнено".

3. Базовий Flow: джерело і collect

Перш ніж говорити про оператори, корисно відчути «голу трубу»: є джерело, яке емітує значення, і є збирач. Flow запускається лише під час collect (тобто потік, по суті, лінивий / cold).

Ось мінімальний потік «сирого тексту»:

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

fun main() = runBlocking {
    val lines = flow {
        emit("INFO|Сервіс запущено")
        emit("WARN|Промах кешу")
    }

    lines.collect { line ->
        println("сирий: $line") // сирий: INFO|Сервіс запущено ...
    }
}

Поки що це не «аналізатор», а просто доставка значень. Далі ми почнемо перетворювати потік на повноцінний конвеєр.

4. Перетворення: map, filter, onEach

map: перетворення і зміна типу

map у Flow за змістом дуже схожий на map у колекціях: берете кожен елемент і перетворюєте його на щось нове. З Flow ідея та сама, лише елементи приходять «у часі», а перетворення може бути suspend-дружнім.

Зробімо парсер рядка в MonitorEvent. Для простоти — без Regex: звичайний 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)
}

Тепер застосуємо 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|Сервіс запущено")
        emit("ERROR|Диск заповнено")
    }
        .map { parseLine(it) }
        .collect { e ->
            println("${e.level}: ${e.message}")
            // INFO: Сервіс запущено
            // ERROR: Диск заповнено
        }
}

Зверніть увагу на важливу річ: тип потоку змінився. Було Flow<String>, стало Flow<MonitorEvent>. Це суперсила map: він уміє змінювати тип, а завдяки цьому конвеєр стає значно виразнішим.

Невеликий лайфхак для читання коду: якщо ви загубилися й не розумієте, «якого типу потік зараз», згадайте, що map змінює тип, а filter — зазвичай ні.

filter: залишаємо лише потрібне

Тепер уявіть, що подій багато, а вас цікавлять лише "WARN" і "ERROR". Тут і стане в пригоді filter: він відбирає елементи за предикатом (лямбдою, яка повертає true/false). Для Flow зміст той самий: значення або проходить далі конвеєром, або відкидається.

Зробімо фільтрацію після парсингу:

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|Сервіс запущено")
        emit("WARN|Промах кешу")
        emit("ERROR|Диск заповнено")
    }
        .map { parseLine(it) }
        .filter { e -> e.level != "INFO" }
        .collect { e ->
            println("ВАЖЛИВО: ${e.level} ${e.message}")
            // ВАЖЛИВО: WARN Промах кешу
            // ВАЖЛИВО: ERROR Диск заповнено
        }
}

Тут filter не змінює тип: як був потік MonitorEvent, так і лишився. Він лише «звужує» набір значень.

onEach: побічні ефекти без зміни даних

Коли ви налагоджуєте потік, дуже хочеться «підглянути», що відбувається на кожному кроці. Перша думка — засунути println у map. Але це поганий стиль: map за змістом має перетворювати дані, а друк — це побічний ефект.

onEach якраз для цього: виконати дію для кожного елемента й пропустити елемент далі без змін. Це схоже на forEach, тільки посеред Flow-конвеєра.

Додамо діагностику: покажемо, що саме пройшло фільтр.

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|Сервіс запущено")
        emit("WARN|Промах кешу")
        emit("ERROR|Диск заповнено")
    }
        .map { parseLine(it) }
        .filter { it.level != "INFO" }
        .onEach { println("debug: пройшло фільтр -> $it") }
        .collect { e ->
            println("ALERT: ${e.message}")
        }
}

Плюс onEach у тому, що його легко прибрати, коли налагодження не потрібне, і він не «ламає» зміст map.

До речі, якщо ви колись побачите println усередині map, це не смертний гріх. Але майже завжди це сигнал: «код час трохи привести до тями».

5. Помилки: catch і межі його дії

catch: обробка помилок у конвеєрі

Обробка помилок — це те, що відрізняє «приклад із підручника» від «програми, яка живе довше 15 секунд». У Kotlin ви вже вмієте try/catch і розумієте, що виняток перериває звичайний потік виконання. У Flow помилки теж трапляються, але часто зручніше перехоплювати їх прямо в конвеєрі оператором catch.

Зробімо парсер суворішим: якщо рядок не містить "|", вважатимемо це помилкою формату.

fun parseLineStrict(line: String): MonitorEvent {
    val parts = line.split("|", limit = 2)
    if (parts.size < 2) error("Поганий формат: '$line'")
    return MonitorEvent(level = parts[0], message = parts[1])
}

Тепер застосуємо catch і «підмінимо» помилку спеціальною подією:

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|Сервіс запущено")
        emit("ЗЛАМАНИЙ РЯДОК")
        emit("ERROR|Диск заповнено")
    }
        .map { parseLineStrict(it) }
        .catch { e ->
            println("помилку оброблено: ${e.message}") // помилку оброблено: Поганий формат: 'ЗЛАМАНИЙ РЯДОК'
            emit(MonitorEvent("ERROR", "Парсер зламався, але застосунок продовжує роботу"))
        }
        .collect { e ->
            println("подія: ${e.level} ${e.message}")
        }
}

Тут важливо зрозуміти механіку: catch перехоплює помилки, які сталися до нього в конвеєрі (upstream). Тобто якщо помилка вилетіла в map { parseLineStrict(it) }, наш catch її побачить. А якщо помилка станеться пізніше (наприклад, усередині collect), то catch може не допомогти.

Чому catch не ловить помилку всередині collect

Це типова пастка: ви ставите catch, заспокоюєтеся й думаєте, що потік тепер «невбивний». Але catch — не чарівна парасоля від усіх бід. Він ловить лише те, що сталося «вище за течією».

Зробімо штучну помилку всередині 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 у collect")
        }
}

Цей код упаде (якщо не обгорнути collect у try/catch), бо помилка сталася вже в споживачі. Це логічно: collect — термінальна операція. Якщо падає саме споживач, то «рятувати» його має зовнішній try/catch, як і в звичайному житті.

Практичне правило: catch — для помилок джерела та операторів до collect. Помилки обробника в collect — ловіть зовні звичайним try/catch.

6. Backpressure у Flow

Слово backpressure звучить як назва рок-гурту («Backpressure і їхній новий альбом „Suspend Me Gently“»), але зміст у нього дуже приземлений. Якщо споживач обробляє значення повільно, хочеться, щоб джерело не продовжувало видавати їх нескінченно швидко. Інакше ми або «захлинемося», або почнемо накопичувати значення в памʼяті.

У Flow (у базовій моделі) є природний механізм: emit і весь конвеєр — suspend-дружні. Тому якщо споживач повільний, то upstream часто призупиняється і чекає.

Подивімося на приклад: джерело швидко емітує числа, а collect обробляє повільно через 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("емічено $i") // друк іде "в такт" обробці
        }
    }.collect { v ->
        delay(100)
        println("зібрано $v")
        // емічено 0
        // зібрано 0
        // емічено 1
        // зібрано 1
        // емічено 2
        // зібрано 2
    }
}

Якщо ви очікували побачити "емічено 0, емічено 1, емічено 2" одразу, а потім "зібрано …", то це якраз і є головний інсайт backpressure у Flow: потік не зобовʼязаний «вистрілювати» все миттєво. Він може бути природно синхронізований зі споживачем через призупинення.

7. Збираємо конвеєр в акуратний пайплайн

Тепер зберемо мініверсію нашого «монітора» так, щоб код читався як текст: «беремо рядки → парсимо → залишаємо важливе → логуємо → ловимо помилки → друкуємо підсумок».

Спочатку зробимо функцію-джерело. У реальному проєкті події приходять звідкись ззовні, але сьогодні нам вистачить симуляції:

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

fun rawLogLines(): Flow<String> = flow {
    emit("INFO|Сервіс запущено")
    emit("WARN|Промах кешу")
    emit("ЗЛАМАНИЙ РЯДОК")
    emit("ERROR|Диск заповнено")
}

Потім опишемо конвеєр як функцію, яка повертає Flow<MonitorEvent>. Це хороший стиль: ми відокремлюємо «що робити» від «як виводити».

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", "потік зламався: ${e.message}"))
        }

І фінальний main, де все збирається:

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

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

Зверніть увагу, як зручно читати такий код: у importantEvents() немає термінальної операції, отже його можна перевикористовувати, тестувати й підʼєднувати в різні місця. А main лишається простим: «збери й виведи».

8. Памʼятка щодо операторів

Коли операторів стає більше, корисно мати в голові «карту». Нехай буде маленька табличка:

map { ... }

filter { ... }

onEach { ... }

catch { ... }

Якщо ви бачите довгий ланцюжок, його можна буквально прочитати вголос: rawLogLines, map, filter, onEach, catch, collect. Зазвичай це вже добрий знак: код схожий на пояснення, а не на загадку.

9. Типові помилки

Помилка № 1: робити побічні ефекти в map, а потім дивуватися, що «логіка якась каламутна».
Коли map одночасно і перетворює дані, і друкує, і рахує статистику, він перестає бути «перетворенням» і стає «шматком усього». У підсумку складно зрозуміти, що саме і навіщо змінюється. Набагато спокійніше тримати map як чисту трансформацію, а друк і налагодження виносити в onEach.

Помилка № 2: поставити catch і вважати, що він ловить узагалі все.
catch ловить помилки, які сталися до нього в ланцюжку, тобто upstream. Якщо помилка виникла всередині collect { ... }, це вже downstream — і catch може не допомогти. У таких ситуаціях краще обгортати collect у звичайний try/catch, як ви робили раніше під час вивчення винятків.

Помилка № 3: «проковтнути» виняток у catch і продовжити, ніби нічого не сталося, без сигналу.
Порожній catch {} — це як заклеїти лампочку «Check Engine» ізоляційною стрічкою: машині від цього легше не стає, зате водій перестає розуміти, що відбувається. Навіть якщо ви видаєте fallback-значення через emit, корисно хоча б залишити повідомлення про помилку або явно сформувати «помилкову подію», щоб поведінка була передбачуваною.

Помилка № 4: зробити весь конвеєр усередині collect і отримати «комбайн» на 80 рядків.
Такий код складно тестувати й важко перевикористовувати: логіка привʼязана до місця виводу. Найчастіше краще винести конвеєр у функцію, яка повертає Flow<...>, а в main лишити лише collect. Це робить код простішим і за структурою дуже схожим на звичайні функції обробки колекцій — тільки «в часі».

Помилка № 5: очікувати, що Flow «настріляє» значення швидше, ніж споживач їх обробить, і не розуміти, чому все йде повільно.
Це не баг: базова ідея backpressure в тому, що Flow уміє природно підлаштовуватися під швидкість обробки завдяки suspend. Якщо в collect стоїть delay, то upstream часто буде «чекати» — і це якраз допомагає не роздувати памʼять та не влаштовувати лавину подій.