Flow — co to je, cold a hot streamy v korutinách Kotlin

Co je Flow v Kotlin?

Flow — je typ z balíčku kotlinx.coroutines.flow, představující studený asynchronní datový tok. V podstatě je Flow korutinová sekvence, která emituje hodnoty přes funkci emit() a končí buď úspěšně, nebo výjimkou. Sběr toku se provádí přes terminální operátor collect(), což je suspend-funkce.

Studená sémantika

Cold stream znamená, že kód uvnitř flow-builderu se provádí znovu pro každého odběratele. Observable.fromIterable v RxJava se chová podobně: nový odběratel obdrží všechny hodnoty od začátku. Ve Flow je to implementováno přes suspend-funkci collect, která blokuje korutinu po celou dobu sběru dat.

Flow builders

Kotlin poskytuje několik způsobů vytvoření Flow: flow { } — základní konstrukce s emit(), flowOf(vararg values) — pro pevnou sadu hodnot, .asFlow() — rozšíření pro kolekce a Sequence. Všechny buildery jsou studené — data se generují pouze při volání terminálního operátoru.

Cold a Hot streamy

Rozdělení na cold a hot streamy je klíčový koncept reaktivního programování. Cold stream (Flow, Observable) spouští generování dat při odběru. Hot stream (Channel, SharedFlow) emituje data nezávisle — odběratel obdrží pouze to, co se děje po odběru, bez začátku sekvence.

SharedFlow — je horký Flow, který může mít více odběratelů a může přehrát poslední hodnoty při nastavení replay. SharedFlow je vhodný pro události (jednorázová oznámení). StateFlow — jeho varianta s pevnou hodnotou stavu, která ukládá poslední hodnotu pro nové odběratele.

ChannelFlow používá Channel pod kapotou, kombinuje vlastnosti Flow a Channel. Podporuje bufferování a backpressure přes kapacitu (capacity). ChannelFlow je užitečný při konverzi callback-API na reaktivní tok, když jsou hodnoty emitovány z různých korutin.

Konverze mezi cold a hot

Pro konverzi cold Flow na hot SharedFlow se používá operátor shareIn(scope, started, replay). Parametr started řídí okamžik spuštění: SharingStarted.WhileSubscribed() — aktivní dokud existují odběratelé, Lazily — spuštění při prvním odběrateli, Eagerly — okamžité spuštění. Opačná konverze — hot na cold: StateFlow.asFlow() vrací studený Flow, který při collect emituje aktuální hodnotu StateFlow. To je pohodlné pro testování.

Operátory Flow

Flow poskytuje bohatou sadu operátorů pracujících jako suspend-funkce uvnitř korutiny. Operátory nemají stav a vracejí nový Flow — původní tok zůstává nezměněn. To umožňuje stavět bezpečné transformační řetězce bez vedlejších účinků.

Operátor map transformuje každou hodnotu toku přes asynchronní nebo synchronní transformaci. filter propouští pouze hodnoty splňující podmínku. catch zachycuje výjimky před terminálním operátorem a umožňuje obnovení toku. flatMapLatest ruší předchozí emisi při příchodu nové hodnoty — analogicky ke switchMap v Rx.

Operátor debounce ve Flow zpožďuje publikaci hodnoty o zadaný timeout. Pokud během této doby přijde nová hodnota — timer se resetuje. V Androidu se debounce používá pro vyhledávání: požadavek je odeslán až po pauze 300-400 ms, což snižuje počet volání API 3-5krát.

Terminální operátory

Kromě collect() Flow podporuje další terminální operátory: toList() shromažďuje všechny hodnoty do seznamu — užitečné pro testy, first() vrací první prvek a ruší tok, single() očekává přesně jeden prvek. fold(initial) akumuluje hodnoty přes předanou funkci. Všechny terminální operátory jsou suspend-funkce a musí být volány uvnitř korutiny nebo jiné suspend-funkce.

Příklady kódu Flow

První příklad — základní Flow s generováním čísel a transformací přes operátor map:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "Číslo: $it" }
        .collect { value ->
            println(value)
        }
}

Druhý příklad — transformace toku s filtrováním a zpracováním chyb přes catch:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

Třetí příklad — použití StateFlow ve ViewModel pro reaktivní UI v Jetpack Compose:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlow a SharedFlow

StateFlow — je horký Flow s jedinou aktuální hodnotou. Ukládá poslední hodnotu a okamžitě ji předává novému odběrateli. StateFlow je Observable kontejner pro stav, podporuje porovnání equals — pokud se nová hodnota shoduje s aktuální, emise nenastane. Jetpack Compose používá StateFlow přes collectAsState().

SharedFlow — je flexibilnější horký Flow bez povinné počáteční hodnoty. SharedFlow se konfiguruje přes replay (počet hodnot pro nové odběratele), extraBufferCapacity (buffer mimo replay) a onBufferOverflow (strategie při přetečení). SharedFlow je ideální pro jednorázové události: navigace, Snackbar, analytika.

Flow je v architektuře Android doporučen Googlem jako hlavní zdroj dat (Vrstva: Repository → UseCase → ViewModel). LiveData je méně flexibilní: Flow podporuje korutiny, operátory, backpressure a funguje mimo UI vrstvu. Migrace z LiveData na Flow je standardní praxí v moderních Android projektech.

Při použití Flow ve ViewModel je důležitý správný výběr typu. StateFlow je ideální pro UI stav, který musí přežít rotaci obrazovky. SharedFlow je vhodný pro události, kde je opětovné zpracování nepřípustné — například navigace. Flow s collect() v lifecycleScope dává maximální kontrolu nad kontextem provádění, ale vyžaduje ruční zrušení při opuštění obrazovky.

Testování Flow se provádí přes kotlinx-coroutines-test. Knihovna poskytuje TestDispatcher — virtuální čas, který umožňuje zrychlovat zpoždění (delay) a řídit pořadí provádění korutin. TestScope.runTest { } vytváří izolované prostředí pro testování Flow. Operátor toList() se často používá v testech ke shromáždění všech hodnot flow s timeoutem, aby se ověřilo, že tok emitoval správnou sekvenci dat.

Flow se dobře integruje s Room (knihovna Android pro databáze): DAO metody mohou vracet Flow<List<Entity>>. Room automaticky emituje novou hodnotu při každé změně tabulky — UI se aktualizuje bez ručního spouštěče. To je implementováno přes InvalidationTracker, který pod kapotou používá Flow s callbackFlow. Takový přístup eliminuje potřebu LiveData a činí datovou vrstvu plně korutinově orientovanou. Jetpack Compose přes collectAsState() odebírá StateFlow a překresluje pouze ty komponenty, jejichž data se změnila — to poskytuje výkon nedosažitelný s LiveData orientovanými architekturami. DataStore (náhrada SharedPreferences) také vrací Flow<Preferences>, zajišťující reaktivní čtení nastavení aplikace bez ručních spouštěčů aktualizace.

Flow podporuje meziprocesovou komunikaci přes kotlinx-coroutines-core na JVM bez dalších knihoven. Například v serverových aplikacích na Ktor může Flow představovat tok příchozích WebSocket zpráv. Každá zpráva je emitována do toku, prochází filtrováním a agregací přes operátory a výsledek je odeslán klientovi. Takový přístup nahrazuje reaktivní knihovny jako Reactor nebo RxJava v Kotlin projektech.

Kompatibilitu Flow se stávajícím RxJava kódem zajišťuje modul kotlinx-coroutines-rx3. Rozšiřující funkce Flow.asObservable() převádí Flow na Observable z RxJava 3. Opačná konverze — CompletableSource.asFlow(), Observable.asFlow(). To zjednodušuje migraci z RxJava na korutiny: projekt lze přepisovat po etapách, přičemž část vrstev zůstává na RxJava. Při konverzi je třeba vzít v úvahu rozdíl v cold/hot sémantice: Observable může být cold i hot, Flow je vždy cold pro běžný Flow a hot pro SharedFlow.

Zpracování chyb a testování Flow

Pro zpracování chyb ve Flow existuje zvláštnost: pokud výjimka vznikne uvnitř flow-builderu před terminálním operátorem, je předána do catch. Pokud výjimka vznikne v operátoru po builderu, catch po tomto operátoru ji zachytí. retryWhen umožňuje opakovat odběr s podmínkou: opakuj při chybě sítě až 3krát, ale neopakuj při CancellationException. Flow eliminuje chyby závislé na stavu, protože neukládá stav — to zjednodušuje ladění ve srovnání s Observable, kde Subject ukládá vnitřní stav.

Testování Flow s kotlinx-coroutines-test používá TestDispatcher pro simulaci zpoždění. Turbine — populární knihovna z komunity pro testování Flow: test { } spouští Flow, awaitItem() očekává další hodnotu, awaitComplete() očekává dokončení. Turbine přidává výchozí timeout, který zabraňuje zaseknutí testů. Pro testování StateFlow použijte .testIn(scope) s ověřením hodnot v chronologickém pořadí.