Flow в Kotlin

ОБЯЗАТЕЛЬНОДЛЯ НОВИЧКОВ

Разработчику Архитектору

---

Flow в Kotlin

Корутина выполняет одну задачу и возвращает один результат. Flow — последовательность значений во времени — пришла новая строка из БД, обновился счётчик, прилетела порция из сети.

Типичные места:

• Room — observeAll(): Flow<List<Note>>;
• Ktor — стриминг ответов;
• Compose — collectAsState из StateFlow.

База: корутины · синхронные коллекции: Коллекции и Sequence в Kotlin.

---

List, Sequence и Flow — сравнение

	List	Sequence	Flow
Когда данные	Всё сразу в памяти	Лениво, в одном потоке	Лениво + suspend
Сеть / БД	Нужно самому загрузить в List	То же	Может эмитить по мере готовности
Повторный запуск	Тот же объект	Новый проход итератора	Холодный Flow — заново при новом collect
Контекст	Обычный код	Обычный код	Корутины

---

Словарь терминов

Термин	Простыми словами
emit	"Выдать" следующее значение подписчикам Flow.
collect	Подписаться и обрабатывать каждое значение (suspend-цикл).
Холодный Flow	Начинает работу, когда появился первый подписчик.
Горячий поток	Живёт сам по себе; подписчики получают текущее состояние (StateFlow).
Upstream / downstream	Продюсер → операторы → collect.
Backpressure	Продюсер быстрее потребителя — буферы (buffer).

---

Холодный Flow

Код ITЗагрузка примера кода…

Разбор:

• Вход: определяется холодный Flow<Int> и его сбор в main.
• Ключевые API: flow, emit, delay, collect образуют минимальный конвейер producer-consumer.
• Поток выполнения: при collect запускается цикл 1..5, каждое число эмитится с паузой 200 мс.
• Результат: в консоль последовательно выводятся значения потока.
• Типичная ошибка: ожидать эмиссии без collect; холодный Flow без подписки не выполняется.

Таблица по ключевым строкам:

Часть	Смысл
flow { }	Билдер: внутри можно emit и вызывать suspend-функции.
delay(200)	Пауза между числами; поток не блокируется.
collect { }	Запускает выполнение; без collect холодный Flow ничего не делает.

Два подряд counter().collect { } выполнят цикл дважды — это норма для холодного потока.

---

Горячие потоки — StateFlow и SharedFlow

Экрану нужно одно актуальное состояние ("загрузка" / "список" / "ошибка"):

private val _uiState = MutableStateFlow(UiState.Loading)
val uiState: StateFlow<UiState> = _uiState.asStateFlow()

fun onDataLoaded(data: List<Item>) {
    _uiState.value = UiState.Success(data)
}

Разбор:

• MutableStateFlow(UiState.Loading) задаёт начальное состояние экрана.
• _uiState оставляют private, чтобы изменить состояние можно было только внутри класса.
• asStateFlow() публикует наружу read-only интерфейс StateFlow.
• _uiState.value = ... синхронно обновляет текущее состояние и уведомляет подписчиков.
• паттерн снижает риск случайных изменений из внешнего кода.

Тип	Когда
StateFlow	Состояние UI, настройки — всегда есть "последнее значение"
SharedFlow	События: "показать Snackbar", навигация один раз
channelFlow	Несколько продюсеров пишут в один Flow

В Compose: uiState.collectAsStateWithLifecycle() — подписка с учётом жизненного цикла.

---

Операторы (как у коллекций, но suspend)

repository.observeOrders()
    .map { orders -> orders.filter { it.isOpen } }
    .distinctUntilChanged()
    .catch { e -> emit(emptyList()); log(e) }
    .flowOn(Dispatchers.IO)
    .collect { openOrders -> render(openOrders) }

Разбор:

• observeOrders() даёт исходный upstream-поток данных.
• map { ... } преобразует каждую эмиссию в список только открытых заказов.
• distinctUntilChanged() убирает дубликаты подряд и снижает шум в UI.
• catch { ... } перехватывает ошибку upstream, логирует и эмитит fallback-значение.
• flowOn(Dispatchers.IO) переносит upstream-обработку на IO-диспетчер.
• collect { render(...) } выполняет терминальную обработку и рендерит результат.

Оператор	Роль
map	Преобразовать каждое значение
filter	Отфильтровать
distinctUntilChanged	Пропустить, если новое equals предыдущему (меньше перерисовок UI)
catch	Поймать ошибку upstream, можно emit запасное значение
flowOn(Dispatchers.IO)	Тяжёлую работу upstream выполнять на IO
debounce	Подождать паузу ввода (поиск по буквам)
flatMapLatest	При новом значении отменить предыдущую вложенную работу

flatMapLatest типичен для поиска: пользователь напечатал "к" → "ко" → "кот"; запрос за "к" отменяется, когда пришло "ко".

---

Объединение потоков

val merged = merge(flowA, flowB)

val combined = combine(tempFlow, humidityFlow) { t, h ->
    Weather(t, h)
}

Разбор:

• merge(flowA, flowB) объединяет эмиссии источников в один поток без ожидания пары.

• combine(...) { t, h -> ... } собирает новое значение из последних данных каждого входного потока.

• лямбда Weather(t, h) формирует итоговую доменную модель для downstream.

• merge удобен для событий, combine — для вычисления состояния из нескольких сигналов.

• merge — пришло значение из любого источника.

• combine — ждёт обновления каждого участника, строит результат из последних пар.

---

Отмена

collect — suspend-функция. При отмене viewModelScope (уход с экрана) цепочка останавливается, upstream получает CancellationException.

Внутри flow { } блокирующий код без withContext(Dispatchers.IO) может занять поток диспетчера — переносите тяжёлую работу на IO.

---

Тестирование

@Test
fun emitsFirstThree() = runTest {
    val items = counter().take(3).toList()
    assertEquals(listOf(1, 2, 3), items)
}

Разбор:

• runTest запускает тест в контролируемом coroutine test scope.
• take(3) ограничивает поток первыми тремя эмиссиями.
• toList() терминально собирает эмиссии в список для проверки.
• assertEquals(...) валидирует порядок и значения полученных элементов.

Для пошаговой проверки событий по мере прихода удобна библиотека Turbine — см. экосистему.

---

Частые ошибки

Симптом	Причина
Flow "молчит"	Нет collect у холодного Flow
Двойной запрос к API	Два collect без shareIn / кэша
UI дёргается	Нет distinctUntilChanged на частых обновлениях
Зависание	Блокирующий код в flow { } на Main

---

Дальше

Корутины · Room + Flow · Compose · тесты

---

Практические сценарии применения Flow

Ниже примеры, где Flow даёт заметный выигрыш по сравнению с "разовым" suspend-запросом:

Сценарий	Что эмитим	Почему Flow удобен
Поле поиска	строку ввода	debounce и flatMapLatest автоматически отменяют старые запросы
Лента из БД	List<Item>	UI получает обновления сразу после insert/delete
Мониторинг сети	состояние Online/Offline	несколько экранов подписываются на один источник
Прогресс загрузки	проценты 0..100	можно показывать промежуточные значения, а не только финал

---

Поиск по мере ввода — полный шаблон

Код ITЗагрузка примера кода…

Разбор:

• queryFlow представляет поток строк от поля ввода.
• map { it.trim() } нормализует текст до бизнес-обработки.
• debounce(300) ждёт паузу набора и уменьшает число запросов.
• distinctUntilChanged() отбрасывает повторяющиеся подряд запросы.
• flatMapLatest отменяет предыдущий запрос и запускает новый для актуального текста.
• ветка flowOf(emptyList()) задаёт корректный результат для пустого запроса.

Что важно в этом шаблоне:

• debounce(300) снижает нагрузку на API при быстром наборе.
• distinctUntilChanged() пропускает повтор одного и того же текста.
• flatMapLatest отменяет старый запрос, когда пользователь продолжил ввод.
• flowOf(emptyList()) задаёт предсказуемое состояние для пустого поиска.

Сетевую часть можно реализовать через Ktor Client, а хранение кэша — через Room.

---

Жизненный цикл и безопасность подписок

Для Android-экрана удобно держать подписку в ViewModel, а в Compose читать только готовый StateFlow:

class FeedViewModel(...) : ViewModel() {
    val uiState = repository.observeFeed()
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5_000),
            initialValue = emptyList()
        )
}

Разбор:

• stateIn(...) превращает холодный поток репозитория в StateFlow для UI.
• scope = viewModelScope привязывает жизненный цикл подписки к ViewModel.
• SharingStarted.WhileSubscribed(5_000) удерживает upstream активным короткое время без подписчиков.
• initialValue = emptyList() гарантирует предсказуемое начальное состояние.
• uiState можно безопасно собирать в Compose через lifecycle-aware API.

WhileSubscribed(5_000) сохраняет поток "тёплым" короткое время после ухода с экрана и уменьшает число повторных запросов при быстрой навигации.

Дополнительный пример — retry с ограничением попыток

Код ITЗагрузка примера кода…

Разбор:

• Вход: поток оборачивает сетевой вызов, который может временно падать по IOException.
• Ключевые API: flow, retryWhen, delay.
• Поток выполнения: при ошибке IOException выполняется до трёх повторов с нарастающей задержкой, затем ошибка пробрасывается дальше.
• Результат: временные сетевые сбои переживаются автоматически без ручной логики повторов в каждом месте.
• Типичная ошибка: делать бесконечный retry без лимита; это перегружает API и маскирует реальные проблемы.

---