Flow — wat is het, cold en hot streams in Kotlin coroutines

Wat is Flow in Kotlin?

Flow — is een type uit het pakket kotlinx.coroutines.flow, dat een koude asynchrone gegevensstroom vertegenwoordigt. In essentie is Flow een coroutine-sequentie die waarden emitteert via de functie emit() en eindigt met succes of met een uitzondering. Het verzamelen van de stroom gebeurt via de terminale operator collect(), die een suspend-functie is.

Cold-semantiek

Cold stream betekent dat de code in de flow-builder opnieuw wordt uitgevoerd voor elke abonnee. Observable.fromIterable in RxJava gedraagt zich vergelijkbaar: een nieuwe abonnee ontvangt alle waarden vanaf het begin. In Flow is dit geïmplementeerd via de suspend-functie collect, die de coroutine blokkeert gedurende de hele gegevensverzameling.

Flow builders

Kotlin biedt verschillende manieren om Flow te maken: flow { } — basisconstructie met emit(), flowOf(vararg values) — voor een vaste set waarden, .asFlow() — extensie voor collecties en Sequence. Alle builders zijn koud — gegevens worden alleen gegenereerd bij aanroep van de terminale operator.

Cold en Hot streams

Verdeling in cold en hot streams is een kernconcept van reactief programmeren. Cold stream (Flow, Observable) start gegevensgeneratie bij abonnement. Hot stream (Channel, SharedFlow) emitteert gegevens onafhankelijk — de abonnee ontvangt alleen wat er na abonnement gebeurt, zonder het begin van de reeks.

SharedFlow — is een hete Flow die meerdere abonnees kan hebben en bij het instellen van replay laatste waarden kan herhalen. SharedFlow is geschikt voor gebeurtenissen (eenmalige meldingen). StateFlow — zijn variant met een vaste toestandswaarde, die de laatste waarde cached voor nieuwe abonnees.

ChannelFlow gebruikt Channel onder de motorkap en combineert eigenschappen van Flow en Channel. Het ondersteunt buffering en backpressure via capaciteit (capacity). ChannelFlow is nuttig bij het converteren van callback-API naar een reactieve stroom, wanneer waarden uit verschillende coroutines worden geëmitteerd.

Conversie tussen cold en hot

Voor conversie van cold Flow naar hot SharedFlow wordt de operator shareIn(scope, started, replay) gebruikt. Parameter started regelt het startmoment: SharingStarted.WhileSubscribed() — actief zolang er abonnees zijn, Lazily — start bij eerste abonnee, Eagerly — onmiddellijke start. Omgekeerde conversie — hot naar cold: StateFlow.asFlow() retourneert een koude Flow die bij collect de huidige waarde van StateFlow emitteert. Dit is handig voor testen.

Flow operatoren

Flow biedt een rijke set operatoren die als suspend-functies binnen een coroutine werken. Operatoren hebben geen toestand en retourneren een nieuwe Flow — de originele stroom blijft ongewijzigd. Dit maakt het mogelijk veilige transformatieketens te bouwen zonder bijwerkingen.

Operator map transformeert elke waarde van de stroom via een asynchrone of synchrone transformatie. filter laat alleen waarden door die aan de voorwaarde voldoen. catch vangt uitzonderingen op vóór de terminale operator en maakt herstel van de stroom mogelijk. flatMapLatest annuleert de vorige emissie bij een nieuwe waarde — analoog aan switchMap in Rx.

Operator debounce in Flow vertraagt de publicatie van een waarde met een opgegeven timeout. Als er binnen deze tijd een nieuwe waarde komt — wordt de timer gereset. In Android wordt debounce gebruikt voor zoeken: het verzoek wordt pas na een pauze van 300-400 ms verzonden, wat het aantal API-aanroepen 3-5 keer vermindert.

Terminale operatoren

Naast collect() ondersteunt Flow andere terminale operatoren: toList() verzamelt alle waarden in een lijst — handig voor tests, first() retourneert het eerste element en annuleert de stroom, single() verwacht exact één element. fold(initial) accumuleert waarden via de doorgegeven functie. Alle terminale operatoren zijn suspend-functies en moeten worden aangeroepen binnen een coroutine of een andere suspend-functie.

Flow code voorbeelden

Eerste voorbeeld — basis Flow met getallengeneratie en transformatie via de operator map:

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

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

Tweede voorbeeld — stroomtransformatie met filtering en foutafhandeling via catch:

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

Derde voorbeeld — gebruik van StateFlow in ViewModel voor reactieve UI in 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 en SharedFlow

StateFlow — is een hete Flow met een enkele huidige waarde. Het cached de laatste waarde en geeft deze onmiddellijk door aan een nieuwe abonnee. StateFlow is een Observable-container voor toestand, ondersteunt equals-vergelijking — als de nieuwe waarde overeenkomt met de huidige, vindt geen emissie plaats. Jetpack Compose gebruikt StateFlow via collectAsState().

SharedFlow — is een flexibelere hete Flow zonder verplichte beginwaarde. SharedFlow wordt geconfigureerd via replay (aantal waarden voor nieuwe abonnees), extraBufferCapacity (buffer buiten replay) en onBufferOverflow (strategie bij overloop). SharedFlow is ideaal voor eenmalige gebeurtenissen: navigatie, Snackbar, analytiek.

Flow wordt in Android-architectuur door Google aanbevolen als primaire gegevensbron (Laag: Repository → UseCase → ViewModel). LiveData is inferieur aan Flow in flexibiliteit: Flow ondersteunt coroutines, operatoren, backpressure en werkt buiten de UI-laag. Migratie van LiveData naar Flow is standaardpraktijk in moderne Android-projecten.

Bij gebruik van Flow in ViewModel is de juiste typekeuze belangrijk. StateFlow is ideaal voor UI-toestand die schermrotatie moet overleven. SharedFlow is geschikt voor gebeurtenissen waar herverwerking onaanvaardbaar is — bijvoorbeeld navigatie. Flow met collect() in lifecycleScope geeft maximale controle over de uitvoeringscontext, maar vereist handmatige annulering bij het verlaten van het scherm.

Testen van Flow gebeurt via kotlinx-coroutines-test. De bibliotheek biedt TestDispatcher — virtuele tijd die het mogelijk maakt vertragingen (delay) te versnellen en de uitvoeringsvolgorde van coroutines te regelen. TestScope.runTest { } creëert een geïsoleerde omgeving voor het testen van Flow. Operator toList() wordt vaak gebruikt in tests om alle flow-waarden met timeout te verzamelen, om te controleren of de stroom de juiste gegevensreeks heeft geëmitteerd.

Flow integreert goed met Room (Android-bibliotheek voor databases): DAO-methoden kunnen Flow<List<Entity>> retourneren. Room emitteert automatisch een nieuwe waarde bij elke tabelwijziging — UI wordt bijgewerkt zonder handmatige trigger. Dit is geïmplementeerd via InvalidationTracker, die onder de motorkap Flow met callbackFlow gebruikt. Een dergelijke benadering elimineert de noodzaak voor LiveData en maakt de gegevenslaag volledig coroutine-georiënteerd. Jetpack Compose abonneert zich via collectAsState() op StateFlow en hertekent alleen die componenten waarvan de gegevens zijn gewijzigd — dit geeft prestaties die onbereikbaar zijn met LiveData-georiënteerde architecturen. DataStore (vervanger van SharedPreferences) retourneert ook Flow<Preferences>, wat reactief lezen van app-instellingen zonder handmatige updatetriggers mogelijk maakt.

Flow ondersteunt interprocescommunicatie via kotlinx-coroutines-core op JVM zonder extra bibliotheken. In serverapplicaties op Ktor kan Flow bijvoorbeeld een stroom van inkomende WebSocket-berichten vertegenwoordigen. Elk bericht wordt in de stroom geëmitteerd, doorloopt filtering en aggregatie via operatoren, en het resultaat wordt naar de client verzonden. Een dergelijke benadering vervangt reactieve bibliotheken zoals Reactor of RxJava in Kotlin-projecten.

Compatibiliteit van Flow met bestaande RxJava-code wordt verzorgd door de module kotlinx-coroutines-rx3. De extensiefunctie Flow.asObservable() converteert Flow naar Observable uit RxJava 3. Omgekeerde conversie — CompletableSource.asFlow(), Observable.asFlow(). Dit vereenvoudigt migratie van RxJava naar coroutines: het project kan stapsgewijs worden herschreven, waarbij een deel van de lagen op RxJava blijft. Bij conversie moet rekening worden gehouden met het verschil in cold/hot-semantiek: Observable kan zowel cold als hot zijn, Flow is altijd cold voor gewone Flow en hot voor SharedFlow.

Foutafhandeling en testen van Flow

Voor foutafhandeling in Flow is er een bijzonderheid: als een uitzondering optreedt in de flow-builder vóór de terminale operator, wordt deze doorgegeven aan catch. Als een uitzondering optreedt in een operator na de builder, vangt catch na deze operator deze op. retryWhen maakt herhaalde abonnering met een voorwaarde mogelijk: herhaal bij netwerkfout tot 3 keer, maar herhaal niet bij CancellationException. Flow elimineert toestandsafhankelijke fouten omdat het geen toestand opslaat — dit vereenvoudigt debuggen in vergelijking met Observable, waar Subject interne toestand opslaat.

Testen van Flow met kotlinx-coroutines-test gebruikt TestDispatcher voor simulatie van vertragingen. Turbine — populaire bibliotheek uit de community voor het testen van Flow: test { } start Flow, awaitItem() wacht op de volgende waarde, awaitComplete() wacht op voltooiing. Turbine voegt een standaard timeout toe, wat vastlopen van tests voorkomt. Voor het testen van StateFlow gebruikt u .testIn(scope) met controle van waarden in chronologische volgorde.