Flow — cos'è, stream cold e hot nelle coroutine Kotlin

Cos'è Flow in Kotlin?

Flow è un tipo del pacchetto kotlinx.coroutines.flow che rappresenta un flusso di dati asincrono cold. In sostanza, Flow è una sequenza di coroutine che emette valori tramite la funzione emit() e termina con successo o con un'eccezione. La raccolta del flusso viene eseguita tramite l'operatore terminale collect(), che è una funzione suspend.

Semantica cold

Stream cold significa che il codice all'interno del builder flow viene eseguito di nuovo per ogni abbonato. Observable.fromIterable in RxJava si comporta in modo simile: un nuovo abbonato riceve tutti i valori dall'inizio. In Flow, ciò è implementato tramite la funzione suspend collect, che blocca la coroutine per tutta la durata della raccolta dati.

Costruttori Flow

Kotlin fornisce diversi modi per creare Flow: flow { } — la costruzione di base con emit(), flowOf(vararg values) — per un insieme fisso di valori, .asFlow() — un'estensione per collezioni e Sequence. Tutti i costruttori sono cold — i dati vengono generati solo quando viene chiamato l'operatore terminale.

Stream cold e hot

La distinzione tra stream cold e hot è un concetto chiave della programmazione reattiva. Stream cold (Flow, Observable) avvia la generazione dei dati al momento dell'abbonamento. Stream hot (Channel, SharedFlow) emette dati indipendentemente — l'abbonato riceve solo ciò che accade dopo l'abbonamento, senza l'inizio della sequenza.

SharedFlow è un Flow hot che può avere più abbonati e riprodurre valori recenti quando replay è configurato. SharedFlow è adatto per eventi (notifiche una tantum). StateFlow è la sua variante con un valore di stato fisso, che memorizza nella cache l'ultimo valore per i nuovi abbonati.

ChannelFlow utilizza Channel internamente, combinando proprietà di Flow e Channel. Supporta il buffering e il backpressure tramite capacity. ChannelFlow è utile quando si convertono API callback in un flusso reattivo, dove i valori vengono emessi da diverse coroutine.

Conversione tra cold e hot

Per convertire un Flow cold in SharedFlow hot si usa l'operatore shareIn(scope, started, replay). Il parametro started controlla il momento di avvio: SharingStarted.WhileSubscribed() — attivo finché ci sono abbonati, Lazily — avvio al primo abbonato, Eagerly — avvio immediato. La conversione inversa — hot in cold: StateFlow.asFlow() restituisce un Flow cold che alla collect emette il valore corrente di StateFlow. Ciò è utile per i test.

Operatori Flow

Flow fornisce un ricco insieme di operatori che funzionano come funzioni suspend all'interno di una coroutine. Gli operatori sono senza stato e restituiscono un nuovo Flow — il flusso originale rimane invariato. Ciò consente di costruire catene di trasformazione sicure senza effetti collaterali.

L'operatore map trasforma ogni valore del flusso tramite una trasformazione asincrona o sincrona. filter fa passare solo i valori che soddisfano la condizione. catch intercetta le eccezioni prima dell'operatore terminale e consente il recupero del flusso. flatMapLatest annulla l'emissione precedente quando arriva un nuovo valore — simile a switchMap in Rx.

L'operatore debounce in Flow ritarda la pubblicazione del valore di un timeout specificato. Se durante questo tempo arriva un nuovo valore, il timer si azzera. In Android, debounce viene utilizzato per la ricerca: la richiesta viene inviata solo dopo una pausa di 300-400 ms, riducendo le chiamate API di 3-5 volte.

Operatori terminali

Oltre a collect(), Flow supporta altri operatori terminali: toList() raccoglie tutti i valori in una lista — utile per i test, first() restituisce il primo elemento e annulla il flusso, single() si aspetta esattamente un elemento. fold(initial) accumula valori tramite una funzione passata. Tutti gli operatori terminali sono funzioni suspend e devono essere chiamati all'interno di una coroutine o di un'altra funzione suspend.

Esempi di codice Flow

Il primo esempio — un Flow base che genera numeri con trasformazione tramite l'operatore map:

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

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

Il secondo esempio — trasformazione del flusso con filtraggio e gestione errori tramite catch:

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

Il terzo esempio — utilizzo di StateFlow nel ViewModel per UI reattiva 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 e SharedFlow

StateFlow è un Flow hot con un singolo valore corrente. Memorizza nella cache l'ultimo valore e lo passa immediatamente a un nuovo abbonato. StateFlow è un contenitore osservabile per lo stato, supporta il confronto equals — se il nuovo valore corrisponde a quello corrente, non si verifica alcuna emissione. Jetpack Compose utilizza StateFlow tramite collectAsState().

SharedFlow è un Flow hot più flessibile senza valore iniziale obbligatorio. SharedFlow è configurato tramite replay (numero di valori per i nuovi abbonati), extraBufferCapacity (buffer oltre al replay) e onBufferOverflow (strategia in caso di overflow). SharedFlow è ideale per eventi una tantum: navigazione, Snackbar, analisi.

Flow nell'architettura Android è raccomandato da Google come fonte dati primaria (Livello: Repository → UseCase → ViewModel). LiveData è inferiore a Flow in flessibilità: Flow supporta coroutine, operatori, backpressure e funziona al di fuori del livello UI. La migrazione da LiveData a Flow è una pratica standard nei progetti Android moderni.

Quando si utilizza Flow in ViewModel, è importante scegliere il tipo giusto. StateFlow è ideale per lo stato UI che deve sopravvivere alla rotazione dello schermo. SharedFlow è adatto per eventi in cui la rielaborazione è inaccettabile — ad esempio, la navigazione. Flow con collect() in lifecycleScope dà il massimo controllo sul contesto di esecuzione ma richiede l'annullamento manuale all'uscita dallo schermo.

Il testing di Flow viene eseguito tramite kotlinx-coroutines-test. La libreria fornisce TestDispatcher — tempo virtuale che consente di accelerare i ritardi (delay) e controllare l'ordine di esecuzione delle coroutine. TestScope.runTest { } crea un ambiente isolato per testare Flow. L'operatore toList() è spesso utilizzato nei test per raccogliere tutti i valori del flow con un timeout, per verificare che il flusso abbia emesso la sequenza di dati corretta.

Flow si integra bene con Room (libreria Android per DB): i metodi DAO possono restituire Flow<List<Entity>>. Room emette automaticamente un nuovo valore a ogni modifica della tabella — l'UI si aggiorna senza attivatore manuale. Ciò è implementato tramite InvalidationTracker, che sotto il cofano utilizza Flow con callbackFlow. Questo approccio elimina la necessità di LiveData e rende il livello dati completamente orientato alle coroutine. Jetpack Compose tramite collectAsState() si abbona a StateFlow e ridisegna solo i componenti i cui dati sono cambiati — ciò offre prestazioni irraggiungibili con architetture orientate a LiveData. DataStore (sostituto di SharedPreferences) restituisce anch'esso Flow<Preferences>, fornendo lettura reattiva delle impostazioni dell'applicazione senza attivatori di aggiornamento manuali.

Flow supporta la comunicazione interprocesso tramite kotlinx-coroutines-core su JVM senza librerie aggiuntive. Ad esempio, nelle applicazioni server con Ktor, Flow può rappresentare un flusso di messaggi WebSocket in entrata. Ogni messaggio viene emesso nel flusso, passa attraverso filtraggio e aggregazione tramite operatori, e il risultato viene inviato al client. Questo approccio sostituisce librerie reattive come Reactor o RxJava nei progetti Kotlin.

La compatibilità di Flow con il codice RxJava esistente è fornita dal modulo kotlinx-coroutines-rx3. La funzione di estensione Flow.asObservable() converte Flow in Observable di RxJava 3. La conversione inversa — CompletableSource.asFlow(), Observable.asFlow(). Ciò semplifica la migrazione da RxJava alle coroutine: si può riscrivere il progetto per fasi, lasciando alcuni livelli su RxJava. Durante la conversione, bisogna considerare la differenza nella semantica cold/hot: Observable può essere sia cold che hot, Flow è sempre cold per Flow normale e hot per SharedFlow.

Gestione errori e testing di Flow

La gestione degli errori in Flow ha una particolarità: se un'eccezione si verifica all'interno del builder flow prima dell'operatore terminale, viene propagata a catch. Se un'eccezione si verifica in un operatore dopo il builder, il catch dopo quell'operatore la intercetta. retryWhen consente di ritentare l'abbonamento con una condizione: ritentare in caso di errore di rete fino a 3 volte, ma non ritentare in caso di CancellationException. Flow elimina gli errori dipendenti dallo stato perché non memorizza lo stato — ciò semplifica il debug rispetto a Observable, dove Subject memorizza lo stato interno.

Il testing di Flow con kotlinx-coroutines-test utilizza TestDispatcher per simulare i ritardi. Turbine è una libreria comunitaria popolare per testare Flow: test { } avvia Flow, awaitItem() attende il valore successivo, awaitComplete() attende il completamento. Turbine aggiunge un timeout predefinito, prevenendo il blocco dei test. Per testare StateFlow, utilizzare .testIn(scope) con verifica dei valori in ordine cronologico.