Reactive streams let you model asynchronous sequences of values — a real-time search box, a stream of sensor readings, a feed of database changes. Kotlin Flow is the modern, coroutine-native answer. Understanding when a producer is faster than its consumer — backpressure — is what separates robust reactive systems from ones that silently drop data or crash under load.
Cold vs Hot Streams
This distinction is fundamental:
Cold stream — the producer doesn't start until there's a subscriber. Each subscriber gets its own independent execution of the entire sequence.
val coldFlow = flow {
println("Starting work") // runs once PER collector
emit(fetchFromNetwork())
}
coldFlow.collect { } // "Starting work" prints
coldFlow.collect { } // "Starting work" prints again — new execution
Hot stream — the producer runs regardless of subscribers. Subscribers join an ongoing stream; they see only values emitted after they subscribe.
val hotFlow = MutableStateFlow(0) // starts immediately, holds latest value
hotFlow.value = 1
hotFlow.collect { } // new collector sees 1, then future values
Flow: The Kotlin-Native Reactive Stream
Flow<T> is a cold, sequential, suspending stream. It runs on the coroutine that calls collect.
fun userUpdates(userId: String): Flow<User> = flow {
while (true) {
emit(api.getUser(userId))
delay(30_000) // check every 30 seconds
}
}
viewModelScope.launch {
userUpdates(userId)
.catch { e -> emit(User.empty()) } // handle errors inline
.collect { user -> _uiState.value = user }
}
Key Flow Operators
userFlow
.filter { it.isActive } // skip inactive users
.map { it.toDisplayModel() } // transform
.distinctUntilChanged() // suppress duplicate emissions
.debounce(300) // wait 300ms of silence (search box)
.flatMapLatest { searchForUser(it) } // cancel previous inner flow on new emission
.flowOn(Dispatchers.IO) // upstream runs on IO; collect still on caller
flowOn vs withContext
flowOn changes the dispatcher of the upstream portion of the pipeline. withContext inside a flow {} builder is wrong — use flowOn instead:
// ✅ Correct
flow { emit(loadFromDisk()) }
.flowOn(Dispatchers.IO) // loadFromDisk runs on IO
.collect { display(it) } // collect still on Main
// ❌ Wrong — throws IllegalStateException
flow {
withContext(Dispatchers.IO) { emit(loadFromDisk()) }
}
StateFlow and SharedFlow: Hot Flows
StateFlow — Single Latest Value
StateFlow is a hot flow that holds exactly one value. New collectors immediately receive the current value. It's a direct replacement for LiveData.
class FeedViewModel : ViewModel() {
private val _uiState = MutableStateFlow<FeedState>(FeedState.Loading)
val uiState: StateFlow<FeedState> = _uiState.asStateFlow()
init {
viewModelScope.launch {
try {
val posts = withContext(Dispatchers.IO) { repository.getPosts() }
_uiState.value = FeedState.Success(posts)
} catch (e: Exception) {
_uiState.value = FeedState.Error(e.message)
}
}
}
}
// In Fragment:
viewLifecycleOwner.lifecycleScope.launch {
repeatOnLifecycle(Lifecycle.State.STARTED) {
viewModel.uiState.collect { state -> render(state) }
}
}
Important: StateFlow uses equals() to deduplicate — if you emit the same value twice, collectors are notified only once. Use SharedFlow if you need all events delivered.
SharedFlow — Configurable Hot Flow
SharedFlow is a general-purpose hot flow without the single-value constraint:
private val _events = MutableSharedFlow<UiEvent>(
replay = 0, // don't replay past events to new subscribers
extraBufferCapacity = 1, // buffer 1 event if there's no active collector
onBufferOverflow = BufferOverflow.DROP_OLDEST
)
val events: SharedFlow<UiEvent> = _events.asSharedFlow()
// Fire-and-forget events (navigation, toasts)
viewModelScope.launch {
_events.emit(UiEvent.ShowToast("Saved!"))
}
Backpressure
Backpressure occurs when the producer emits faster than the consumer can process. In a Flow, the producer is automatically suspended until the consumer is ready — this is built-in, because Flow is sequential and suspending.
flow {
repeat(1_000_000) { i ->
emit(i) // suspends here if collector is slow
}
}
.collect { value ->
delay(100) // slow consumer — producer suspends between emissions
process(value)
}
This is implicit backpressure via coroutine suspension — the safest default. No values are dropped, no buffer overflow.
Buffering: Decouple Producer and Consumer
When you want the producer and consumer to run concurrently (producer shouldn't wait for consumer), add a buffer:
flow { emit(readLargeBatch()) } // IO-bound producer
.buffer(capacity = 64) // producer runs ahead, filling the buffer
.map { processItem(it) } // CPU-bound consumer
.collect { saveResult(it) }
Conflation: Keep Only the Latest
For UI state, you almost never care about intermediate values — just the latest:
locationFlow // emits GPS coordinates very frequently
.conflate() // if collector is busy, drop intermediate values
.collect { updateMap(it) } // only processes latest coordinate
conflate() is equivalent to .buffer(Channel.CONFLATED).
collectLatest: Cancel-and-Restart
Useful when each emission triggers a new long-running operation and you only care about the result for the latest input:
searchQueryFlow
.debounce(300)
.collectLatest { query -> // cancels previous collection block if new query arrives
val results = api.search(q) // previous in-flight search is cancelled
_results.value = results
}
Comparing Flow to RxJava
If you're migrating from RxJava:
| RxJava | Flow equivalent |
|---|---|
Observable | Flow (cold) |
Subject | MutableSharedFlow |
BehaviorSubject | MutableStateFlow |
Single | suspend fun T |
Maybe | suspend fun T? |
Completable | suspend fun Unit |
.subscribeOn(Schedulers.io()) | .flowOn(Dispatchers.IO) |
.observeOn(AndroidSchedulers.mainThread()) | collect on Dispatchers.Main |
onErrorReturn | .catch { emit(default) } |
flatMap | flatMapMerge |
switchMap | flatMapLatest |
concatMap | flatMapConcat |
zip | zip |
combineLatest | combine |
The biggest conceptual shift: in RxJava, backpressure is a problem you configure per-operator (Flowable vs Observable). In Kotlin Flow, suspension handles it automatically — you opt in to buffering only when you need concurrency.
Practical: Search Box with Flow
class SearchViewModel(private val repo: SearchRepository) : ViewModel() {
private val _query = MutableStateFlow("")
val results: StateFlow<List<Result>> = _query
.debounce(300)
.filter { it.length >= 2 }
.distinctUntilChanged()
.flatMapLatest { query ->
flow { emit(repo.search(query)) }
.catch { emit(emptyList()) }
}
.stateIn(
scope = viewModelScope,
started = SharingStarted.WhileSubscribed(5_000),
initialValue = emptyList()
)
fun onQueryChanged(query: String) {
_query.value = query
}
}
stateIn converts a cold Flow into a StateFlow — sharing a single upstream subscription across all collectors, and automatically starting/stopping based on subscribers.
Key Takeaways
| Concept | When to Use |
|---|---|
Flow (cold) | One-shot or on-demand streams; network calls, DB queries |
StateFlow | UI state — single current value, deduplication |
SharedFlow | Events (toast, navigation) — no current-value semantics |
flowOn | Change dispatcher for upstream work |
buffer() | Decouple producer/consumer speed |
conflate() | Drop intermediate values; keep only latest |
collectLatest | Cancel-and-restart on each new emission |
debounce | Rate-limit fast inputs (search, scrolling) |
flatMapLatest | Switch to new inner flow, cancel previous |