kotlin-coroutines-flows

Coroutines et Flows Kotlin

Modèles pour la concurrence structurée, les flux réactifs basés sur Flow, et les tests de coroutines dans les projets Android et Kotlin Multiplatform.

Quand activer

• Écrire du code asynchrone avec les coroutines Kotlin
• Utiliser Flow, StateFlow ou SharedFlow pour des données réactives
• Gérer des opérations concurrentes (chargement parallèle, debounce, retry)
• Tester les coroutines et les Flows
• Gérer les portées de coroutines et l'annulation

Concurrence structurée

Hiérarchie des portées

Application
  └── viewModelScope (ViewModel)
        └── coroutineScope { } (enfant structuré)
              ├── async { } (tâche concurrente)
              └── async { } (tâche concurrente)

Utilisez toujours la concurrence structurée — jamais GlobalScope :

// MAUVAIS
GlobalScope.launch { fetchData() }

// BON — limité au cycle de vie du ViewModel
viewModelScope.launch { fetchData() }

// BON — limité au cycle de vie du composable
LaunchedEffect(key) { fetchData() }

Décomposition parallèle

Utilisez coroutineScope + async pour le travail parallèle :

suspend fun loadDashboard(): Dashboard = coroutineScope {
    val items = async { itemRepository.getRecent() }
    val stats = async { statsRepository.getToday() }
    val profile = async { userRepository.getCurrent() }
    Dashboard(
        items = items.await(),
        stats = stats.await(),
        profile = profile.await()
    )
}

SupervisorScope

Utilisez supervisorScope lorsque les échecs d'un enfant ne doivent pas annuler ses frères :

suspend fun syncAll() = supervisorScope {
    launch { syncItems() }       // un échec ici n'annulera pas syncStats
    launch { syncStats() }
    launch { syncSettings() }
}

Modèles Flow

Cold Flow — Conversion one-shot en flux

fun observeItems(): Flow<List<Item>> = flow {
    // Ré-émet à chaque changement de la base de données
    itemDao.observeAll()
        .map { entities -> entities.map { it.toDomain() } }
        .collect { emit(it) }
}

StateFlow pour l'état de l'UI

class DashboardViewModel(
    observeProgress: ObserveUserProgressUseCase
) : ViewModel() {
    val progress: StateFlow<UserProgress> = observeProgress()
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5_000),
            initialValue = UserProgress.EMPTY
        )
}

WhileSubscribed(5_000) maintient le flux amont actif pendant 5 secondes après le départ du dernier abonné — survit aux changements de configuration sans redémarrer.

Combinaison de plusieurs Flows

val uiState: StateFlow<HomeState> = combine(
    itemRepository.observeItems(),
    settingsRepository.observeTheme(),
    userRepository.observeProfile()
) { items, theme, profile ->
    HomeState(items = items, theme = theme, profile = profile)
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), HomeState())

Opérateurs Flow

// Debounce de la saisie de recherche
searchQuery
    .debounce(300)
    .distinctUntilChanged()
    .flatMapLatest { query -> repository.search(query) }
    .catch { emit(emptyList()) }
    .collect { results -> _state.update { it.copy(results = results) } }

// Réessai avec backoff exponentiel
fun fetchWithRetry(): Flow<Data> = flow { emit(api.fetch()) }
    .retryWhen { cause, attempt ->
        if (cause is IOException && attempt < 3) {
            delay(1000L * (1 shl attempt.toInt()))
            true
        } else {
            false
        }
    }

SharedFlow pour les événements uniques

class ItemListViewModel : ViewModel() {
    private val _effects = MutableSharedFlow<Effect>()
    val effects: SharedFlow<Effect> = _effects.asSharedFlow()

    sealed interface Effect {
        data class ShowSnackbar(val message: String) : Effect
        data class NavigateTo(val route: String) : Effect
    }

    private fun deleteItem(id: String) {
        viewModelScope.launch {
            repository.delete(id)
            _effects.emit(Effect.ShowSnackbar("Élément supprimé"))
        }
    }
}

// Collecte dans un Composable
LaunchedEffect(Unit) {
    viewModel.effects.collect { effect ->
        when (effect) {
            is Effect.ShowSnackbar -> snackbarHostState.showSnackbar(effect.message)
            is Effect.NavigateTo -> navController.navigate(effect.route)
        }
    }
}

Dispatchers

// Travail intensif en CPU
withContext(Dispatchers.Default) { parseJson(largePayload) }

// Travail lié aux E/S
withContext(Dispatchers.IO) { database.query() }

// Thread principal (UI) — par défaut dans viewModelScope
withContext(Dispatchers.Main) { updateUi() }

En KMP, utilisez Dispatchers.Default et Dispatchers.Main (disponibles sur toutes les plateformes). Dispatchers.IO est réservé à JVM/Android — utilisez Dispatchers.Default sur les autres plateformes ou fournissez-le via l'injection de dépendances.

Annulation

Annulation coopérative

Les boucles longues doivent vérifier l'annulation :

suspend fun processItems(items: List<Item>) = coroutineScope {
    for (item in items) {
        ensureActive()  // lance CancellationException si annulé
        process(item)
    }
}

Nettoyage avec try/finally

viewModelScope.launch {
    try {
        _state.update { it.copy(isLoading = true) }
        val data = repository.fetch()
        _state.update { it.copy(data = data) }
    } finally {
        _state.update { it.copy(isLoading = false) }  // s'exécute toujours, même en cas d'annulation
    }
}

Tests

Test de StateFlow avec Turbine

@Test
fun `la recherche met à jour la liste d'articles`() = runTest {
    val fakeRepository = FakeItemRepository().apply { emit(testItems) }
    val viewModel = ItemListViewModel(GetItemsUseCase(fakeRepository))

    viewModel.state.test {
        assertEquals(ItemListState(), awaitItem())  // initial

        viewModel.onSearch("query")
        val loading = awaitItem()
        assertTrue(loading.isLoading)

        val loaded = awaitItem()
        assertFalse(loaded.isLoading)
        assertEquals(1, loaded.items.size)
    }
}

Test avec TestDispatcher

@Test
fun `le chargement parallèle se termine correctement`() = runTest {
    val viewModel = DashboardViewModel(
        itemRepo = FakeItemRepo(),
        statsRepo = FakeStatsRepo()
    )

    viewModel.load()
    advanceUntilIdle()

    val state = viewModel.state.value
    assertNotNull(state.items)
    assertNotNull(state.stats)
}

Simulation de Flows

class FakeItemRepository : ItemRepository {
    private val _items = MutableStateFlow<List<Item>>(emptyList())

    override fun observeItems(): Flow<List<Item>> = _items

    fun emit(items: List<Item>) { _items.value = items }

    override suspend fun getItemsByCategory(category: String): Result<List<Item>> {
        return Result.success(_items.value.filter { it.category == category })
    }
}

Anti-modèles à éviter

• Utiliser GlobalScope — fuit des coroutines, pas d'annulation structurée
• Collecter des Flows dans init {} sans portée — utilisez viewModelScope.launch
• Utiliser MutableStateFlow avec des collections mutables — utilisez toujours des copies immuables : _state.update { it.copy(list = it.list + newItem) }
• Attraper CancellationException — laissez-le se propager pour une annulation correcte
• Utiliser flowOn(Dispatchers.Main) pour collecter — le dispatcher de collecte est celui de l'appelant
• Créer un Flow dans @Composable sans remember — recrée le flow à chaque recomposition

Références

Voir la compétence : compose-multiplatform-patterns pour la consommation UI des Flows. Voir la compétence : android-clean-architecture pour le placement des coroutines dans les couches.