Flow — qu'est-ce que c'est, flux cold et hot dans les coroutines Kotlin

Qu'est-ce que Flow dans Kotlin ?

Flow est un type du package kotlinx.coroutines.flow, représentant un flux de données asynchrone cold. À la base, Flow est une séquence de coroutine qui émet des valeurs via la fonction emit() et se termine soit avec succès, soit avec une exception. La collecte du flux s'effectue via l'opérateur terminal collect(), qui est une fonction suspend.

Sémantique cold

Flux cold signifie que le code à l'intérieur du builder flow s'exécute à nouveau pour chaque abonné. Observable.fromIterable dans RxJava se comporte de manière similaire : un nouvel abonné reçoit toutes les valeurs depuis le début. Dans Flow, ceci est implémenté via la fonction suspend collect, qui bloque la coroutine pendant toute la durée de la collecte de données.

Constructeurs Flow

Kotlin fournit plusieurs façons de créer Flow : flow { } — la construction de base avec emit(), flowOf(vararg values) — pour un ensemble fixe de valeurs, .asFlow() — une extension pour les collections et Sequence. Tous les constructeurs sont cold — les données sont générées uniquement lors de l'appel de l'opérateur terminal.

Flux cold et hot

La distinction entre flux cold et hot est un concept clé de la programmation réactive. Flux cold (Flow, Observable) démarre la génération de données lors de l'abonnement. Flux hot (Channel, SharedFlow) émet des données indépendamment — l'abonné reçoit uniquement ce qui se produit après l'abonnement, sans le début de la séquence.

SharedFlow est un Flow hot qui peut avoir plusieurs abonnés et rejouer les valeurs récentes lorsque replay est configuré. SharedFlow convient aux événements (notifications uniques). StateFlow est sa variante avec une valeur d'état fixe, qui met en cache la dernière valeur pour les nouveaux abonnés.

ChannelFlow utilise Channel en interne, combinant les propriétés de Flow et Channel. Il prend en charge la mise en mémoire tampon et le backpressure via capacity. ChannelFlow est utile lors de la conversion d'API callback en flux réactif, où les valeurs sont émises depuis différentes coroutines.

Conversion entre cold et hot

Pour convertir un Flow cold en SharedFlow hot, on utilise l'opérateur shareIn(scope, started, replay). Le paramètre started contrôle le moment du démarrage : SharingStarted.WhileSubscribed() — actif tant qu'il y a des abonnés, Lazily — démarrage au premier abonné, Eagerly — démarrage immédiat. La conversion inverse — hot vers cold : StateFlow.asFlow() retourne un Flow cold qui émet la valeur actuelle du StateFlow lors de collect. Ceci est pratique pour les tests.

Opérateurs Flow

Flow fournit un ensemble riche d'opérateurs qui fonctionnent comme des fonctions suspend à l'intérieur d'une coroutine. Les opérateurs sont sans état et retournent un nouveau Flow — le flux d'origine reste inchangé. Cela permet de construire des chaînes de transformation sûres sans effets secondaires.

L'opérateur map transforme chaque valeur du flux via une transformation asynchrone ou synchrone. filter ne laisse passer que les valeurs qui satisfont la condition. catch intercepte les exceptions avant l'opérateur terminal et permet de récupérer le flux. flatMapLatest annule l'émission précédente lorsqu'une nouvelle valeur arrive — similaire à switchMap dans Rx.

L'opérateur debounce dans Flow retarde la publication de la valeur d'un délai spécifié. Si une nouvelle valeur arrive pendant ce temps, le chronomètre se réinitialise. Sous Android, debounce est utilisé pour la recherche : la requête n'est envoyée qu'après une pause de 300 à 400 ms, ce qui réduit les appels API de 3 à 5 fois.

Opérateurs terminaux

En plus de collect(), Flow prend en charge d'autres opérateurs terminaux : toList() collecte toutes les valeurs dans une liste — utile pour les tests, first() retourne le premier élément et annule le flux, single() attend exactement un élément. fold(initial) accumule les valeurs via une fonction passée. Tous les opérateurs terminaux sont des fonctions suspend et doivent être appelés à l'intérieur d'une coroutine ou d'une autre fonction suspend.

Exemples de code Flow

Le premier exemple — un Flow de base générant des nombres avec transformation via l'opérateur map :

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

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

Le deuxième exemple — transformation de flux avec filtrage et gestion d'erreurs via catch :

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

Le troisième exemple — utilisation de StateFlow dans ViewModel pour une UI réactive dans 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 et SharedFlow

StateFlow est un Flow hot avec une seule valeur actuelle. Il met en cache la dernière valeur et la transmet immédiatement à un nouvel abonné. StateFlow est un conteneur observable pour l'état, prend en charge la comparaison equals — si la nouvelle valeur correspond à la valeur actuelle, aucune émission n'a lieu. Jetpack Compose utilise StateFlow via collectAsState().

SharedFlow est un Flow hot plus flexible sans valeur initiale obligatoire. SharedFlow est configuré via replay (nombre de valeurs pour les nouveaux abonnés), extraBufferCapacity (tampon au-delà de replay) et onBufferOverflow (stratégie en cas de débordement). SharedFlow est idéal pour les événements uniques : navigation, Snackbar, analyses.

Flow dans l'architecture Android est recommandé par Google comme source de données principale (Couche : Repository → UseCase → ViewModel). LiveData est inférieur à Flow en flexibilité : Flow prend en charge les coroutines, les opérateurs, le backpressure et fonctionne en dehors de la couche UI. La migration de LiveData vers Flow est une pratique standard dans les projets Android modernes.

Lors de l'utilisation de Flow dans ViewModel, il est important de choisir le bon type. StateFlow est idéal pour l'état UI qui doit survivre à la rotation de l'écran. SharedFlow convient aux événements où le retraitement est inacceptable — par exemple, la navigation. Flow avec collect() dans lifecycleScope donne un contrôle maximal sur le contexte d'exécution mais nécessite une annulation manuelle lors de la sortie de l'écran.

Le test de Flow s'effectue via kotlinx-coroutines-test. La bibliothèque fournit TestDispatcher — un temps virtuel qui permet d'accélérer les délais (delay) et de contrôler l'ordre d'exécution des coroutines. TestScope.runTest { } crée un environnement isolé pour tester Flow. L'opérateur toList() est souvent utilisé dans les tests pour collecter toutes les valeurs du flow avec un délai d'attente, afin de vérifier que le flux a émis la séquence de données correcte.

Flow s'intègre bien avec Room (bibliothèque Android pour BD) : les méthodes DAO peuvent retourner Flow<List<Entity>>. Room émet automatiquement une nouvelle valeur à tout changement de table — l'UI se met à jour sans déclencheur manuel. Ceci est implémenté via InvalidationTracker, qui utilise en interne Flow avec callbackFlow. Cette approche élimine le besoin de LiveData et rend la couche de données entièrement orientée coroutines. Jetpack Compose via collectAsState() s'abonne à StateFlow et redessine uniquement les composants dont les données ont changé — cela offre des performances inaccessibles avec les architectures orientées LiveData. DataStore (remplacement de SharedPreferences) retourne également Flow<Preferences>, fournissant une lecture réactive des paramètres de l'application sans déclencheurs de mise à jour manuels.

Flow prend en charge la communication interprocessus via kotlinx-coroutines-core sur JVM sans bibliothèques supplémentaires. Par exemple, dans les applications serveur avec Ktor, Flow peut représenter un flux de messages WebSocket entrants. Chaque message est émis dans le flux, passe par un filtrage et une agrégation via des opérateurs, et le résultat est envoyé au client. Cette approche remplace les bibliothèques réactives comme Reactor ou RxJava dans les projets Kotlin.

La compatibilité de Flow avec le code RxJava existant est fournie par le module kotlinx-coroutines-rx3. La fonction d'extension Flow.asObservable() convertit Flow en Observable de RxJava 3. La conversion inverse — CompletableSource.asFlow(), Observable.asFlow(). Cela simplifie la migration de RxJava vers les coroutines : on peut réécrire le projet par étapes, en laissant certaines couches sur RxJava. Lors de la conversion, il faut tenir compte de la différence de sémantique cold/hot : Observable peut être à la fois cold et hot, Flow est toujours cold pour un Flow normal et hot pour SharedFlow.

Gestion des erreurs et test de Flow

La gestion des erreurs dans Flow a une particularité : si une exception se produit à l'intérieur du builder flow avant l'opérateur terminal, elle est propagée à catch. Si une exception se produit dans un opérateur après le builder, le catch après cet opérateur l'intercepte. retryWhen permet de réessayer l'abonnement avec une condition : réessayer en cas d'erreur réseau jusqu'à 3 fois, mais ne pas réessayer en cas de CancellationException. Flow élimine les erreurs dépendantes de l'état car il ne stocke pas d'état — cela simplifie le débogage par rapport à Observable, où Subject stocke un état interne.

Le test de Flow avec kotlinx-coroutines-test utilise TestDispatcher pour simuler des délais. Turbine est une bibliothèque communautaire populaire pour tester Flow : test { } lance Flow, awaitItem() attend la valeur suivante, awaitComplete() attend la fin. Turbine ajoute un délai d'attente par défaut, empêchant les tests de se bloquer. Pour tester StateFlow, utilisez .testIn(scope) avec vérification des valeurs dans l'ordre chronologique.