Kotlin Coroutines Flows

1---2name: kotlin-coroutines-flows3description: Kotlin Coroutines and Flow patterns for Android and KMP — structured concurrency, Flow operators, StateFlow, error handling, and testing.4origin: ECC5---6 7# Kotlin Coroutines & Flows8 9Patterns for structured concurrency, Flow-based reactive streams, and coroutine testing in Android and Kotlin Multiplatform projects.10 11## When to Activate12 13- Writing async code with Kotlin coroutines14- Using Flow, StateFlow, or SharedFlow for reactive data15- Handling concurrent operations (parallel loading, debounce, retry)16- Testing coroutines and Flows17- Managing coroutine scopes and cancellation18 19## Structured Concurrency20 21### Scope Hierarchy22 23```24Application25  └── viewModelScope (ViewModel)26        └── coroutineScope { } (structured child)27              ├── async { } (concurrent task)28              └── async { } (concurrent task)29```30 31Always use structured concurrency — never `GlobalScope`:32 33```kotlin34// BAD35GlobalScope.launch { fetchData() }36 37// GOOD — scoped to ViewModel lifecycle38viewModelScope.launch { fetchData() }39 40// GOOD — scoped to composable lifecycle41LaunchedEffect(key) { fetchData() }42```43 44### Parallel Decomposition45 46Use `coroutineScope` + `async` for parallel work:47 48```kotlin49suspend fun loadDashboard(): Dashboard = coroutineScope {50    val items = async { itemRepository.getRecent() }51    val stats = async { statsRepository.getToday() }52    val profile = async { userRepository.getCurrent() }53    Dashboard(54        items = items.await(),55        stats = stats.await(),56        profile = profile.await()57    )58}59```60 61### SupervisorScope62 63Use `supervisorScope` when child failures should not cancel siblings:64 65```kotlin66suspend fun syncAll() = supervisorScope {67    launch { syncItems() }       // failure here won't cancel syncStats68    launch { syncStats() }69    launch { syncSettings() }70}71```72 73## Flow Patterns74 75### Cold Flow — One-Shot to Stream Conversion76 77```kotlin78fun observeItems(): Flow<List<Item>> = flow {79    // Re-emits whenever the database changes80    itemDao.observeAll()81        .map { entities -> entities.map { it.toDomain() } }82        .collect { emit(it) }83}84```85 86### StateFlow for UI State87 88```kotlin89class DashboardViewModel(90    observeProgress: ObserveUserProgressUseCase91) : ViewModel() {92    val progress: StateFlow<UserProgress> = observeProgress()93        .stateIn(94            scope = viewModelScope,95            started = SharingStarted.WhileSubscribed(5_000),96            initialValue = UserProgress.EMPTY97        )98}99```100 101`WhileSubscribed(5_000)` keeps the upstream active for 5 seconds after the last subscriber leaves — survives configuration changes without restarting.102 103### Combining Multiple Flows104 105```kotlin106val uiState: StateFlow<HomeState> = combine(107    itemRepository.observeItems(),108    settingsRepository.observeTheme(),109    userRepository.observeProfile()110) { items, theme, profile ->111    HomeState(items = items, theme = theme, profile = profile)112}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), HomeState())113```114 115### Flow Operators116 117```kotlin118// Debounce search input119searchQuery120    .debounce(300)121    .distinctUntilChanged()122    .flatMapLatest { query -> repository.search(query) }123    .catch { emit(emptyList()) }124    .collect { results -> _state.update { it.copy(results = results) } }125 126// Retry with exponential backoff127fun fetchWithRetry(): Flow<Data> = flow { emit(api.fetch()) }128    .retryWhen { cause, attempt ->129        if (cause is IOException && attempt < 3) {130            delay(1000L * (1 shl attempt.toInt()))131            true132        } else {133            false134        }135    }136```137 138### SharedFlow for One-Time Events139 140```kotlin141class ItemListViewModel : ViewModel() {142    private val _effects = MutableSharedFlow<Effect>()143    val effects: SharedFlow<Effect> = _effects.asSharedFlow()144 145    sealed interface Effect {146        data class ShowSnackbar(val message: String) : Effect147        data class NavigateTo(val route: String) : Effect148    }149 150    private fun deleteItem(id: String) {151        viewModelScope.launch {152            repository.delete(id)153            _effects.emit(Effect.ShowSnackbar("Item deleted"))154        }155    }156}157 158// Collect in Composable159LaunchedEffect(Unit) {160    viewModel.effects.collect { effect ->161        when (effect) {162            is Effect.ShowSnackbar -> snackbarHostState.showSnackbar(effect.message)163            is Effect.NavigateTo -> navController.navigate(effect.route)164        }165    }166}167```168 169## Dispatchers170 171```kotlin172// CPU-intensive work173withContext(Dispatchers.Default) { parseJson(largePayload) }174 175// IO-bound work176withContext(Dispatchers.IO) { database.query() }177 178// Main thread (UI) — default in viewModelScope179withContext(Dispatchers.Main) { updateUi() }180```181 182In KMP, use `Dispatchers.Default` and `Dispatchers.Main` (available on all platforms). `Dispatchers.IO` is JVM/Android only — use `Dispatchers.Default` on other platforms or provide via DI.183 184## Cancellation185 186### Cooperative Cancellation187 188Long-running loops must check for cancellation:189 190```kotlin191suspend fun processItems(items: List<Item>) = coroutineScope {192    for (item in items) {193        ensureActive()  // throws CancellationException if cancelled194        process(item)195    }196}197```198 199### Cleanup with try/finally200 201```kotlin202viewModelScope.launch {203    try {204        _state.update { it.copy(isLoading = true) }205        val data = repository.fetch()206        _state.update { it.copy(data = data) }207    } finally {208        _state.update { it.copy(isLoading = false) }  // always runs, even on cancellation209    }210}211```212 213## Testing214 215### Testing StateFlow with Turbine216 217```kotlin218@Test219fun `search updates item list`() = runTest {220    val fakeRepository = FakeItemRepository().apply { emit(testItems) }221    val viewModel = ItemListViewModel(GetItemsUseCase(fakeRepository))222 223    viewModel.state.test {224        assertEquals(ItemListState(), awaitItem())  // initial225 226        viewModel.onSearch("query")227        val loading = awaitItem()228        assertTrue(loading.isLoading)229 230        val loaded = awaitItem()231        assertFalse(loaded.isLoading)232        assertEquals(1, loaded.items.size)233    }234}235```236 237### Testing with TestDispatcher238 239```kotlin240@Test241fun `parallel load completes correctly`() = runTest {242    val viewModel = DashboardViewModel(243        itemRepo = FakeItemRepo(),244        statsRepo = FakeStatsRepo()245    )246 247    viewModel.load()248    advanceUntilIdle()249 250    val state = viewModel.state.value251    assertNotNull(state.items)252    assertNotNull(state.stats)253}254```255 256### Faking Flows257 258```kotlin259class FakeItemRepository : ItemRepository {260    private val _items = MutableStateFlow<List<Item>>(emptyList())261 262    override fun observeItems(): Flow<List<Item>> = _items263 264    fun emit(items: List<Item>) { _items.value = items }265 266    override suspend fun getItemsByCategory(category: String): Result<List<Item>> {267        return Result.success(_items.value.filter { it.category == category })268    }269}270```271 272## Anti-Patterns to Avoid273 274- Using `GlobalScope` — leaks coroutines, no structured cancellation275- Collecting Flows in `init {}` without a scope — use `viewModelScope.launch`276- Using `MutableStateFlow` with mutable collections — always use immutable copies: `_state.update { it.copy(list = it.list + newItem) }`277- Catching `CancellationException` — let it propagate for proper cancellation278- Using `flowOn(Dispatchers.Main)` to collect — collection dispatcher is the caller's dispatcher279- Creating `Flow` in `@Composable` without `remember` — recreates the flow every recomposition280 281## References282 283See skill: `compose-multiplatform-patterns` for UI consumption of Flows.284See skill: `android-clean-architecture` for where coroutines fit in layers.