vnidrop/.codex/skills/compose-skill/references/coroutines-flow-advanced.md

Coroutines & Flow — Advanced Patterns

Backpressure strategies, bridging callback APIs to Flow, concurrency primitives, and testing with Turbine. For core coroutine and Flow patterns (StateFlow/SharedFlow/Channel, operators, dispatchers, scopes, exception handling, stateIn/shareIn), see coroutines-flow.md.

Backpressure

When producer emits faster than consumer processes:

Strategy	Behavior	Use when
Default (no buffer)	Producer suspends until consumer processes	Simple sequential work
buffer(capacity)	Queue between producer and consumer	Smooth speed spikes, process every item
conflate()	Drop old values, keep only latest	UI updates, progress bars — stale data unnecessary
collectLatest { }	Cancel previous processing when new value arrives	Search — only final result matters

// Search with collectLatest: only the last query completes
queryFlow
    .debounce(300)
    .distinctUntilChanged()
    .collectLatest { query ->
        val results = repository.search(query) // cancelled if new query arrives
        _state.update { it.copy(results = results) }
    }

flowOn

flowOn changes the dispatcher for upstream operators and automatically buffers at the context switch:

repository.observeProducts()        // runs on IO
    .map { it.toDomain() }           // runs on IO
    .flowOn(Dispatchers.IO)           // everything above runs on IO
    .collect { updateUi(it) }         // runs on caller's dispatcher (Main)

callbackFlow and channelFlow

callbackFlow — bridge listener APIs to Flow

Use callbackFlow to convert callback-based platform APIs into a Flow. In CMP, place these wrappers in expect/actual declarations or platform source sets.

// Android example — LocationManager (place in androidMain for CMP)
fun LocationManager.locationUpdates(): Flow<Location> = callbackFlow {
    val listener = LocationListener { location ->
        trySend(location) // non-blocking, thread-safe
    }
    requestLocationUpdates(GPS_PROVIDER, 1000L, 0f, listener)
    awaitClose { removeUpdates(listener) } // mandatory cleanup
}

Rules:

• Use trySend() (non-blocking) not send() (suspending) from callbacks
• awaitClose { } is mandatory — omitting it throws IllegalStateException
• The cleanup block in awaitClose unregisters the listener

channelFlow — concurrent production

fun loadDashboard(): Flow<DashboardSection> = channelFlow {
    launch { send(DashboardSection.Profile(fetchProfile())) }
    launch { send(DashboardSection.Stats(fetchStats())) }
    launch { send(DashboardSection.Feed(fetchFeed())) }
}

Use channelFlow when producing values from multiple concurrent coroutines. Use callbackFlow specifically for wrapping external callback APIs.

Concurrency Primitives

Mutex — mutual exclusion

private val mutex = Mutex()
private var tokenCache: String? = null

suspend fun getToken(): String = mutex.withLock {
    tokenCache ?: refreshToken().also { tokenCache = it }
}

Use Mutex for: token refresh synchronization, shared mutable state protection, sequential access to resources.

Semaphore — limited concurrency

private val semaphore = Semaphore(permits = 3)

suspend fun downloadFile(url: String): ByteArray = semaphore.withPermit {
    httpClient.get(url).body()
}

Use Semaphore for: rate-limiting concurrent network calls, limiting parallel file operations.

Why not synchronized?

synchronized blocks the thread. Coroutines suspend — blocking a thread holding a coroutine defeats the purpose. Use Mutex.withLock instead of synchronized in coroutine code.

Testing with Turbine

Turbine API quick reference

Function	Purpose
flow.test { }	Start collecting and asserting
awaitItem()	Wait for next emission, fail if timeout
awaitComplete()	Assert flow completes
awaitError()	Assert flow throws
expectNoEvents()	Assert no emissions pending
cancelAndIgnoreRemainingEvents()	Clean up after assertions
cancelAndConsumeRemainingEvents()	Cancel and return remaining events

runTest from kotlinx-coroutines-test provides deterministic coroutine execution — delays are skipped automatically. Use advanceUntilIdle() to process all pending coroutines.

For full ViewModel event→state→effect testing patterns with Turbine, see testing.md.