The Last Frame Matters: Finishing Android Flows with transformWhile
A field guide to ending a Kotlin Flow on your terms, without losing the one value users actually care about.
Every Android codebase eventually grows a screen that watches something finish: an upload, a payment, a sync, a background job. The data source is a stream, the stream has an ending, and the ending is the whole point.
Yet the first operator most of us reach for to stop that stream, takeWhile, quietly throws the ending away. This post is about the operator that fixes it, transformWhile from kotlinx.coroutines.flow, and about the production failure modes that show up once you start relying on it.
The behavior described here follows the official transformWhile API reference (documented against kotlinx.coroutines 1.11.0 at the time of writing). Everything else comes from patterns I would want a reviewer to catch.
Version baseline
transformWhile was introduced as an experimental operator in kotlinx.coroutines 1.4.0 and promoted to stable in 1.6.0. Target 1.6.0 or newer, or you will hit @ExperimentalCoroutinesApi opt-in errors on older enterprise codebases.
The samples also use data object (Kotlin 1.9+) and trailing commas (Kotlin 1.4+).
1. The bug that ships: the finish line gets swallowed
Here is the model we will use throughout. It is deliberately small.
sealed interface UploadState {
data object Queued : UploadState
data class Progress(val percent: Int) : UploadState
data class Success(val remoteUrl: String) : UploadState
data class Failed(val reason: Throwable) : UploadState
}
val UploadState.isTerminal: Boolean
get() = this is UploadState.Success || this is UploadState.Failed
A developer wants the collector to stop once the upload settles, so they write the obvious thing:
// Looks right. Is wrong.
fun Flow<UploadState>.untilSettled(): Flow<UploadState> = takeWhile { !it.isTerminal }
takeWhile stops before the first value that fails the predicate. The Success or Failed value is exactly the one that fails it, so the UI sees a progress bar frozen at 70% and a flow that ends politely, with no result.
In a debug build with a fast fake, you may never notice. In production, the last value is the one carrying the URL or the error.
The fix is to emit first and decide afterwards:
fun Flow<UploadState>.untilSettled(): Flow<UploadState> = transformWhile { state ->
emit(state) // deliver the value, including the terminal one
!state.isTerminal // keep going only while the upload is unsettled
}
2. The contract in one minute
transformWhile takes a suspending lambda whose receiver is a FlowCollector<R>. The lambda runs for each upstream value and returns a Boolean: true means "keep collecting upstream", false means "stop now".
That is the whole protocol, and three consequences follow from it:
- You control emission. Inside the lambda you may emit nothing, one value, or many, and the output type
Rcan differ from the input typeT. - Emission and termination are decoupled. The value you emit and the value you return are independent decisions, which is why the boundary element can be delivered.
takeWhileis a special case. The reference describestransformWhileas a generalization oftakeWhileand a handy building block for other operators.
Use this to choose between the options:
And here is how it compares with the other common way to wait for a terminal value: first { } is a good fit when you only need the single settled result as a suspend call. Reach for transformWhile when downstream still needs the journey (progress, intermediate states) and the destination.
3. A production-shaped example: watching a WorkManager upload
Real uploads run in WorkManager so they survive process death. getWorkInfoByIdFlow gives you a stream of WorkInfo, and it has a natural ending: a succeeded, failed or cancelled state. Map it to the domain model and finish on the terminal state.
import androidx.work.WorkInfo
import androidx.work.WorkManager
import java.util.UUID
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.map
class UploadCancelled : Exception("Upload was cancelled")
private fun WorkInfo.toUploadState(): UploadState = when (state) {
WorkInfo.State.ENQUEUED, WorkInfo.State.BLOCKED -> UploadState.Queued
WorkInfo.State.RUNNING -> UploadState.Progress(progress.getInt("percent", 0))
WorkInfo.State.SUCCEEDED -> UploadState.Success(outputData.getString("url").orEmpty())
WorkInfo.State.FAILED -> UploadState.Failed(IllegalStateException(outputData.getString("error") ?: "Upload failed"))
WorkInfo.State.CANCELLED -> UploadState.Failed(UploadCancelled())
}
fun WorkManager.observeUpload(id: UUID): Flow<UploadState> =
getWorkInfoByIdFlow(id)
.map { it.toUploadState() }
.untilSettled()
In current work-runtime releases, getWorkInfoByIdFlow returns a non-null Flow<WorkInfo>, so no null filtering is needed and adding filterNotNull() would only be a redundant call that strict lint or -Werror setups may flag.
If you are pinned to an older release whose element type is nullable, add .filterNotNull() before the map; check the signature in your version.
4. More than a stopper: skip, expand, retype
Because you own the collector inside the lambda, one operator can do the jobs of three.
sealed interface UploadUi {
data object Spinner : UploadUi
data class Bar(val percent: Int) : UploadUi
data class Done(val url: String) : UploadUi
data class Error(val message: String) : UploadUi
}
fun Flow<UploadState>.asUiUntilSettled(): Flow<UploadUi> = transformWhile { state ->
when (state) {
UploadState.Queued -> emit(UploadUi.Spinner)
is UploadState.Progress -> emit(UploadUi.Bar(state.percent))
is UploadState.Success -> {
emit(UploadUi.Bar(100)) // one input, two outputs
emit(UploadUi.Done(state.remoteUrl))
}
is UploadState.Failed -> emit(UploadUi.Error(state.reason.message ?: "Upload failed"))
}
!state.isTerminal
}
Success turns into two UI events, the type changes from UploadState to UploadUi, and the stream still ends on the terminal value. Skipping is just as easy: return true without calling emit.
5. Failure modes that survive code review
5.1 Returning false before you emit
transformWhile { state ->
if (state.isTerminal) return@transformWhile false // value vanishes silently
emit(state)
true
}
Nothing crashes. The flow completes and the final value never existed. Make it a habit: the emit call comes first, the Boolean is the last expression of the lambda, and an early return is a code smell.
5.2 Swallowing the stop signal around emit
transformWhile { state ->
try {
emit(state)
} catch (e: Exception) { // too broad
log.warn("render failed", e)
}
!state.isTerminal
}
emit can throw cancellation-type exceptions, including when something downstream decides to stop early or the collecting coroutine is cancelled. A blanket catch (e: Exception) eats those and lets the lambda keep running as if nothing happened, which breaks structured concurrency.
If you must handle errors, catch narrow types, rethrow CancellationException, or move the error handling downstream with catch { } where it belongs.
5.3 Mutable state captured outside the flow
Say you want to throttle progress to steps of five percent. A counter feels natural:
// Broken: `lastShown` outlives the first collection.
fun Flow<UploadState>.coarseBroken(step: Int = 5): Flow<UploadState> {
var lastShown = -step
return transformWhile { state ->
val skip = state is UploadState.Progress && state.percent - lastShown < step
if (!skip) {
if (state is UploadState.Progress) lastShown = state.percent
emit(state)
}
!state.isTerminal
}
}
Flows are cold, so the returned flow can be collected more than once: a retry, a second screen, a test. The second collection starts from the first one's leftovers and silently drops most of its progress.
Scope the state to each collection by creating it inside a flow { } builder:
fun Flow<UploadState>.coarseUntilSettled(step: Int = 5): Flow<UploadState> {
val upstream = this
return flow {
var lastShown = -step // fresh for every collector
emitAll(
upstream.transformWhile { state ->
val skip = state is UploadState.Progress && state.percent - lastShown < step
if (!skip) {
if (state is UploadState.Progress) lastShown = state.percent
emit(state)
}
!state.isTerminal
}
)
}
}
5.4 The replayed terminal state
This one bites on real devices, not in unit tests. If the source is a StateFlow, it keeps its latest value for as long as it lives. A terminal Success stays in the flow after the first collector has handled it, so any new collector receives it immediately:
// Fragment: runs again after rotation, and again every time we return to STARTED
viewLifecycleOwner.lifecycleScope.launch {
viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {
viewModel.state.untilSettled().collect { state ->
render(state)
if (state is UploadState.Success) navigator.openReceipt(state.remoteUrl) // fires again
}
}
}
untilSettled() finished the first collection correctly. The problem is that repeatOnLifecycle starts a new collection from scratch each time the lifecycle reaches STARTED, and the new one is handed the old terminal value.
The fix is to stop treating navigation as an event and model it as state that the UI acknowledges. Rendering stays idempotent, and the pending action lives in the state until the UI reports it has been handled, so a replay can never fire it twice:
import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import java.util.UUID
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch
data class UploadScreenState(
val upload: UploadState = UploadState.Queued,
val receiptToOpen: String? = null, // navigation modeled as state
)
class UploadViewModel(private val uploads: UploadRepository) : ViewModel() {
private val _screen = MutableStateFlow(UploadScreenState())
val screen: StateFlow<UploadScreenState> = _screen.asStateFlow()
private var watchedId: UUID? = null
private var watchJob: Job? = null
fun start(id: UUID) {
if (watchedId == id) return // same upload: safe on rotation, even after it settled
watchJob?.cancel() // a different upload: stop watching the old one
watchedId = id
_screen.value = UploadScreenState() // fresh state for the new upload
watchJob = viewModelScope.launch {
uploads.observe(id) // cold source with a real ending
.untilSettled()
.collect { state ->
_screen.update { current ->
current.copy(
upload = state,
receiptToOpen = if (state is UploadState.Success) state.remoteUrl else current.receiptToOpen,
)
}
}
}
}
fun onReceiptOpened() = _screen.update { it.copy(receiptToOpen = null) }
}
// Fragment: a plain collect of state; no terminal operator needed on the UI side
viewLifecycleOwner.lifecycleScope.launch {
viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {
viewModel.screen.collect { ui ->
render(ui.upload)
ui.receiptToOpen?.let { url ->
navigator.openReceipt(url)
viewModel.onReceiptOpened() // acknowledge: replay is now harmless
}
}
}
}
Notice that start is guarded by the upload id, not by whether the job is still active. An isActive check looks tempting, but once the stream settles the job completes, and the next rotation would relaunch the watcher and raise receiptToOpen again after the UI had already acknowledged it. Comparing ids keeps rotation idempotent while still allowing a new upload to take over; a retry should enqueue new work with a new id.
transformWhile still earns its place, but on the cold, finite source inside the ViewModel, not on a long-lived StateFlow that the UI re-subscribes to.
A note on the alternative many teams reach for: a Channel exposed with receiveAsFlow() delivers each event to one consumer, but it has no notion of lifecycle. An event can be consumed by a collector that is being torn down during a configuration change and never reach the new UI. Android's architecture guidance leans toward representing UI events as state for this reason. If you do choose a channel, treat that lost-delivery risk as a known trade-off and test rotation explicitly.
5.5 A terminal state that never arrives
transformWhile can only end the flow when your predicate says so. If the process dies, the job is evicted, or a bug means the terminal value is never produced, the collector suspends forever. Put a deadline on any wait that blocks a user-facing path:
import kotlin.time.Duration
import kotlinx.coroutines.flow.lastOrNull
import kotlinx.coroutines.withTimeoutOrNull
suspend fun Flow<UploadState>.awaitSettled(timeout: Duration): UploadState? =
withTimeoutOrNull(timeout) {
untilSettled().lastOrNull()
}
The Duration here is kotlin.time.Duration, which withTimeoutOrNull accepts directly on recent coroutines versions. A null means "no verdict in time, or the upstream ended empty". Treat that as its own UI case rather than as success or failure.
5.6 Upstream errors arrive before your terminal state
If the upstream throws, transformWhile does not get to see a terminal value. Decide deliberately whether an exception should become a domain value:
fun Flow<UploadState>.safeUntilSettled(): Flow<UploadState> =
untilSettled().catch { emit(UploadState.Failed(it)) }
Place catch after the terminating operator so that it only handles upstream failures, and never rethrow-swallow cancellation inside it.
6. Test the ending, not just the middle
Terminal behavior is easy to assert an
Comments
No comments yet. Start the discussion.