A flow connects a producer with a consumer. When production is faster than consumption, the application needs a policy: slow the producer, store some values, skip outdated values, or cancel obsolete consumer work. Each choice changes observable behavior and must match the meaning of the data.
The flows lesson introduces cold flows and state holders. This guide develops buffering and sharing lifetime without promising a timing-specific sequence of dropped values. Examples target Kotlin/JVM with kotlinx-coroutines-core 1.10.2. RunBlocking is a console entry point, not a recommended way to block a user interface.
Default flow processing provides a pace
An ordinary sequential flow pipeline runs producer and downstream work in one coroutine. Emission waits for downstream processing before the producer proceeds. This naturally applies backpressure: a slow consumer limits the pace of production without an ever-growing queue.
Buffer introduces a channel boundary so upstream and downstream can progress in separate coroutines. The default overflow policy suspends when the configured buffer is full, preserving values while keeping capacity bounded. Buffering can overlap different work, but it does not guarantee a speedup or automatically move work to a different CPU thread.
Choose capacity from the workload's needs rather than selecting unlimited storage to silence a slow-consumer problem. A queue consumes memory and increases the age of values waiting to be processed. If every payment event matters, preserving all events is essential; if values are cursor positions, processing old positions may be unnecessary.
Conflation and latest work solve different problems
Conflate allows a slow collector to skip intermediate pending values and receive a more recent value. It does not cancel the collector's current action. Use it when the latest state matters more than observing every transition. A progress display can often skip old percentages; a ledger of payments usually cannot.
CollectLatest cancels the previous collector action when a new value arrives. The action must cooperate with cancellation, and any already completed external effect remains completed. This fits replaceable work such as rendering a preview, but is dangerous if you assume cancelling an action undoes a submitted order.
Timing-based demos often claim one exact list of surviving values. That list can change with scheduling and machine load. Test the semantic contract instead: preserved values for buffer, possible skipped intermediates for conflate, and cancellation of obsolete work for collectLatest. Use controlled signals when an exact interleaving is genuinely part of the test.
Sharing adds an owner and a starting policy
A cold flow normally runs its upstream separately for each collector. ShareIn can turn it into a shared hot flow; stateIn produces a StateFlow holding the latest value. The supplied scope owns the sharing coroutine, so choose a scope whose lifetime matches the component using the data.
SharingStarted.Eagerly starts upstream without waiting for a subscriber. Lazily waits for the first subscriber and then follows its documented lifetime. WhileSubscribed can stop upstream after subscribers disappear and supports explicit timeout and replay-expiration choices. These policies affect resource usage and later subscribers' experience; they are not interchangeable decorative options.
StateFlow represents current state and conflates updates using equality. A slow subscriber may miss intermediate states, and normal upstream completion does not turn the StateFlow into a completed finite stream. Handle expected upstream errors before sharing if subscribers need an error state. Otherwise an upstream failure affects the sharing job rather than being delivered as ordinary data automatically.
Scope cancellation and duplicate upstream work
Passing an application-wide scope for a temporary screen can keep work alive after the screen leaves. Conversely, a scope cancelled too early can stop data that another owner still needs. Identify the owner, close it deliberately, and avoid creating anonymous detached scopes with no disposal path.
Sharing can reduce duplicate subscriptions to an expensive source, but replay has a memory and staleness policy. Decide which old values a new subscriber should receive. A state holder's initial value is real observable state, so choose something such as Loading when zero would misleadingly look like a completed count.
Remember that flowOn controls upstream execution context, while buffering and sharing control other aspects of the pipeline. Do not treat every flow operator as a thread switch. Keep ownership, data-loss policy and execution context separate in your design so each can be reviewed and tested directly.
Official references
Read the buffer API, conflate API, collectLatest API and stateIn contract. Write down value preservation, current-work cancellation, replay and lifetime policies before tuning capacity.
Kotlin Flow Buffering and Sharing Lifetime Examples
Each example shows the code, the output it produces and what each part does.
Preserve values through a bounded buffer
The example prints only from the collector and uses the suspending overflow policy, so all values stay in order. It does not assert a precise interleaving of upstream log messages or a runtime duration. Those details depend on scheduling and are unnecessary to prove the preservation contract.
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.runBlocking
fun main() = runBlocking {
flowOf(2, 4, 6)
.map { it * 3 }
.buffer(capacity = 1)
.collect { value ->
delay(1)
println(value)
}
}6
12
18Kotlin Flow Buffering and Sharing Lifetime in Real Projects
Worked exercise: Own a shared stock-count state
Create a short-lived scope, turn a finite source into current state, wait for the expected final count, and cancel the owning scope in finally. The first predicate waits for a meaningful state instead of sleeping and hoping the sharing job has run.
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.runBlocking
fun main() = runBlocking {
val owner = CoroutineScope(SupervisorJob() + Dispatchers.Default)
try {
val stock = flowOf(1, 2, 3).stateIn(owner, SharingStarted.Eagerly, -1)
val finalCount = stock.first { it == 3 }
println("stock:$finalCount")
println("current:${stock.value}")
} finally {
owner.cancel()
}
}stock:3
current:3The scope is explicit and disposed. The consumer does not require observing one and two, because StateFlow models the latest state. In production, do not wait forever for a value that the source might never produce; define timeout, failure and lifecycle policies appropriate to the application.
Kotlin Flow Buffering and Sharing Lifetime FAQ
Does buffer move work to another thread?
It creates a coroutine boundary and channel buffering, but a separate coroutine is not necessarily a separate thread. Execution still depends on the coroutine context and dispatcher. Buffer can overlap suspending upstream and downstream work while both use the same thread at different times. Use an appropriate context policy for blocking or CPU-intensive work and measure completed processing, rather than assuming that adding buffer automatically creates parallel CPU execution.
Are conflate and collectLatest equivalent?
No. Conflate can replace queued intermediate values while the collector finishes its current action. CollectLatest cancels the current action when a newer value arrives. Both can be useful for replaceable state, but their cleanup and side-effect implications differ. If an action submits a payment before cancellation, the submission remains an external effect. Select the operator from the domain's tolerance for missed values and cancelled work, not from a similar-looking output demo.
Will StateFlow deliver every intermediate value?
No. StateFlow represents the latest state and uses equality-based conflation. A slow collector can miss intermediate updates, and an equal update may not produce a new observed state. This is appropriate for current screen state but unsuitable as the sole durable event ledger. If every event matters, use a design with explicit delivery, buffering and persistence guarantees. Separate an event history from a latest-state view when the application needs both.
Why does a sharing scope need a lifecycle owner?
Sharing launches upstream collection under the supplied scope. Without a deliberate owner, subscriptions and resources can outlive the component that requested them, or stop before consumers are finished. Choose a component, screen or application lifetime based on who truly needs the data, then define cancellation and starting policy. A short-lived console scope belongs in a finally cleanup as shown here; framework lifecycle scopes should follow their own supported disposal rules.
Kotlin Flow Buffering and Sharing Lifetime Exercises
Answer each one and check it instantly. If you get stuck, open the hint before the answer.
A temporary component shares an expensive source. Which design addresses its lifetime?
Kotlin Flow Buffering and Sharing Lifetime Quiz
2 questions to check what you learned.
Which policy preserves buffered values by waiting when capacity is full?
- Always discard every new value.
- Buffer's suspending overflow policy.
- CollectLatest cancels every producer permanently.
What does collectLatest do when another value arrives?
- Automatically reverses every external effect of the old action.
- Guarantees every action runs to completion.
- Cancels the previous collector action and starts work for the newer value.