Kotlin Flows (Android / KMP)

For cancellation safety, parallel execution, synchronization, scopes, and blocking code patterns, see kotlin-coroutines.

Flow Type Selection

StateFlow for held state

private val _state = MutableStateFlow(ProfileState())
val state = _state.asStateFlow()

// Always update with .update { } — never replace the flow
_state.update { it.copy(isLoading = true) }

SharedFlow for multi-subscriber events

private val _bluetoothMessages = MutableSharedFlow<BluetoothMessage>(
    replay = 0,
    extraBufferCapacity = 64,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)
val bluetoothMessages = _bluetoothMessages.asSharedFlow()

• replay = 0 — new subscribers don't get old events
• extraBufferCapacity — buffer emissions when collectors are slow
• onBufferOverflow — what to do when buffer is full

callbackFlow for callback conversion

fun observeLocationUpdates(
    locationClient: FusedLocationProviderClient,
    request: LocationRequest
): Flow<Location> = callbackFlow {
    val callback = object : LocationCallback() {
        override fun onLocationResult(result: LocationResult) {
            result.lastLocation?.let { trySend(it) }
        }
    }
    locationClient.requestLocationUpdates(request, callback, Looper.getMainLooper())

    awaitClose {                              // ← MANDATORY
        locationClient.removeLocationUpdates(callback)
    }
}

awaitClose is mandatory in every callbackFlow. Without it, the flow completes immediately.

Channel for single-subscriber events

private val _events = Channel<ProfileEvent>()
val events = _events.receiveAsFlow()

Anti-pattern: one-shot fetch in a flow builder

// WRONG — this is just a suspend function pretending to be a flow
fun getNotes(): Flow<List<Note>> = flow {
    val notes = repository.fetchNotes()
    emit(notes)
}

// CORRECT — just make it a suspend function
suspend fun getNotes(): List<Note> = repository.fetchNotes()

Offline-first pattern

Return Room's reactive Flow as the single source of truth. Trigger updates via insert() elsewhere — never wrap fetch-and-insert in a flow { } builder:

// Repository
fun observeNotes(): Flow<List<Note>> = noteDao.observeAll()  // Room's reactive Flow

suspend fun refreshNotes() {
    val remote = api.fetchNotes()
    noteDao.upsertAll(remote.map { it.toEntity() })
    // Room's Flow automatically emits the new data
}

stateIn and shareIn

ViewModel layer — always WhileSubscribed

val state: StateFlow<ProfileState> = combine(
    userRepository.observeUser(userId),
    settingsRepository.observeSettings()
) { user, settings ->
    ProfileState(user = user.toUi(), settings = settings.toUi())
}
    .stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5_000L),
        initialValue = ProfileState()
    )

WhileSubscribed(5_000L) stops the upstream flow 5 seconds after the last subscriber disappears. This survives configuration changes (screen rotation) without restarting the flow, but cleans up when the screen is truly gone.

Outside ViewModel — prefer Lazily

class BluetoothRepository(
    private val applicationScope: CoroutineScope,
    private val bluetoothAdapter: BluetoothAdapter
) {
    val connectionState: StateFlow<ConnectionState> = observeConnectionState()
        .stateIn(
            scope = applicationScope,
            started = SharingStarted.Lazily,      // starts on first collector, stays hot
            initialValue = ConnectionState.Disconnected
        )
}

Use Lazily outside ViewModels — starts on first collection, stays hot for the scope's lifetime. Only use Eagerly if the flow must start immediately regardless of collectors (rare).

Single subscriber? Don't use shareIn

If you know there will be only one subscriber, use Channel + receiveAsFlow() instead of shareIn:

// WRONG — shareIn with a single subscriber wastes resources
val events = eventFlow.shareIn(scope, SharingStarted.Lazily)

// CORRECT — Channel for single subscriber
private val _events = Channel<Event>()
val events = _events.receiveAsFlow()

SharedFlow configuration

val messages: SharedFlow<Message> = messageFlow.shareIn(
    scope = applicationScope,
    started = SharingStarted.Lazily,
    replay = 1                                   // new subscribers get the last emission
)

• replay = 0 — no replay, only future emissions (default)
• replay = 1 — new subscribers immediately get the most recent value
• Think carefully about whether replay > 0 makes sense for your use case

Flow Operators

Search pipeline example

val searchResults: StateFlow<List<Result>> = searchQuery
    .debounce(300L)                               // wait for typing to pause
    .distinctUntilChanged()                        // skip if query didn't change
    .flatMapLatest { query ->                      // cancel previous search on new query
        if (query.isBlank()) flowOf(emptyList())
        else searchRepository.search(query)
    }
    .stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5_000L),
        initialValue = emptyList()
    )

combine — merge latest from multiple flows

val dashboardState: StateFlow<DashboardState> = combine(
    userRepository.observeUser(),
    notificationRepository.observeUnreadCount(),
    settingsRepository.observeTheme()
) { user, unreadCount, theme ->
    DashboardState(user = user, unreadCount = unreadCount, theme = theme)
}
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000L), DashboardState())

onStart / onCompletion — for side effects

repository.observeNotes()
    .onStart { _state.update { it.copy(isLoading = true) } }
    .onCompletion { _state.update { it.copy(isLoading = false) } }
    .collect { notes -> _state.update { it.copy(notes = notes) } }

flowOn Placement

flowOn affects UPSTREAM operators only. Place it immediately after the operators that need the dispatcher change.

// CORRECT — map runs on IO, collect runs on the caller's dispatcher
flow { emit(readFromDisk()) }
    .map { parse(it) }
    .flowOn(Dispatchers.IO)       // ← affects flow{} and map{} above
    .collect { updateUi(it) }     // runs on caller's dispatcher (Main)

// WRONG — flowOn at the end does NOT affect collect
flow { emit(readFromDisk()) }
    .collect { updateUi(it) }
    .flowOn(Dispatchers.IO)       // ← does nothing useful here

Backpressure

// Search — cancel stale processing
searchResults.collectLatest { results ->
    // If a new emission arrives while this is running, this block is cancelled
    renderResults(results)
}

// Sensor — only care about latest reading
sensorFlow
    .conflate()
    .collect { reading -> updateDisplay(reading) }

// Producer-consumer — don't slow down the producer
dataStream
    .buffer(capacity = 64)
    .collect { item -> slowProcess(item) }

Lifecycle-Aware Collection

Activity / Fragment — repeatOnLifecycle

lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.state.collect { state ->
            // Only collects when lifecycle is at least STARTED.
            // Automatically cancels collection in onStop, restarts in onStart.
            updateUi(state)
        }
    }
}

Compose — collectAsStateWithLifecycle

@Composable
fun ProfileScreen(viewModel: ProfileViewModel = koinViewModel()) {
    val state by viewModel.state.collectAsStateWithLifecycle()
    // Automatically lifecycle-aware — stops collecting when the composable leaves composition
    ProfileContent(state = state)
}

Testing

See android-testing skill for full Flow test patterns including Turbine (.test { awaitItem() }), runTest, and UnconfinedTestDispatcher.

Common Mistakes

Checklist: Writing Flow Code

• Flow type matches use case (StateFlow for state, Channel for single-sub events, callbackFlow for callbacks)
• awaitClose present in every callbackFlow
• stateIn uses WhileSubscribed(5_000L) in ViewModels
• stateIn / shareIn outside ViewModel uses Lazily unless there's a strong reason for Eagerly
• flowOn placed upstream of the operators it should affect
• Single-subscriber flows use Channel + receiveAsFlow(), not shareIn
• No one-shot flow { emit(singleCall()) } — use a suspend fun instead
• Backpressure strategy chosen when producer is faster than consumer
• Lifecycle-aware collection used in UI (repeatOnLifecycle or collectAsStateWithLifecycle)
• SharedFlow replay and extraBufferCapacity configured intentionally