In Kotlin Coroutines, Flow is an API that represents a stream of asynchronously produced values. Instead of producing a single result (like a suspend function), a Flow can emit multiple values over time (0 … n).
Flows are especially useful when you need to represent ongoing data sequences, such as updates from a database, sensor readings, UI events, or any kind of asynchronous pipeline.
map, filter, flatMap, etc. :contentReference[oaicite:1]{index=1}flow { … }, and only runs when collected.
val cold = flow {
emit(1)
delay(500)
emit(2)
}
StateFlow: holds a “state” (latest value) and replays it to new subscribers. :contentReference[oaicite:3]{index=3}SharedFlow: more general broadcast stream; supports replay, buffer, and multiple subscribers. :contentReference[oaicite:4]{index=4}One of your notes is very important: suspend functions only return one object. For example, a suspend function may return a User or a List. But once it's returned, that's it.
suspend fun fetchUsers(): List {
// Simulates a network or DB call
return userApi.getUsers()
}
fun userUpdates(): Flow = flow {
emit(User("Alice"))
delay(1000)
emit(User("Bob"))
delay(1000)
emit(User("Charlie"))
}
With the flow, you get each user update over time. It’s not just one batch — it's a stream.
By default, a Flow created with the flow { … } builder is **cold**: every time you call collect, the producer block inside flow { … } is re-executed for that collector. :contentReference[oaicite:6]{index=6}
Implications:
Hot flows emit values regardless of whether there are active collectors. Their lifecycle is decoupled from the collector. Examples include StateFlow and SharedFlow. :contentReference[oaicite:7]{index=7}
Some characteristics:
Trade-offs when using hot flows:
shareIn operator. :contentReference[oaicite:11]{index=11}Here’s a richer producer-consumer example using a cold flow, showing various operators like buffer, conflate, and collectLatest to handle mismatched speeds between producer and consumer.
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun numbersFlow(): Flow = flow {
for (i in 1..10) {
delay(200) // simulate a producer that emits an item every 200ms
println("Producing $i")
emit(i)
}
}
fun main() = runBlocking {
// Consumer 1: processes slowly
launch {
numbersFlow()
.buffer() // buffer items if producer is faster
.collect { value ->
delay(500) // slow processing
println("Collector A got: $value")
}
}
// Consumer 2: uses collectLatest
launch {
numbersFlow()
.collectLatest { value ->
// If a new value comes, cancel the block for the previous one
println("Collector B collecting latest: $value")
delay(400) // "processing" the value
println("Collector B done with: $value")
}
}
// Wait for collectors
delay(5000)
}
Here’s what’s happening:
.buffer(), so the producer doesn’t wait for Collector A. The items are queued up and processed when A is ready. :contentReference[oaicite:12]{index=12}.collectLatest: if a new value arrives while B is still processing the old, the old block is canceled and the new one starts. :contentReference[oaicite:13]{index=13}Backpressure happens when the producer is faster than the consumer — or when the consumer is slower than the speed of emissions. Flow gives you several operators to manage this:
buffer(): allows the producer to continue emitting into a buffer so that the consumer can catch up. :contentReference[oaicite:14]{index=14}conflate(): drop intermediate values and only keep the latest if the consumer is slow. :contentReference[oaicite:15]{index=15}collectLatest { … }: cancel the previous collection block when a new value arrives; useful for tasks where only the latest result matters (e.g., search queries). :contentReference[oaicite:16]{index=16}debounce(), sample(), etc. (see operator libraries) :contentReference[oaicite:17]{index=17}Since cold flows start fresh for each collector, you can have multiple independent consumers. Each collector triggers its own execution of the flow’s code.
fun simpleFlow(): Flow = flow {
for (i in 1..3) {
delay(300)
emit("Value $i")
println("Emitted Value $i")
}
}
runBlocking {
launch {
simpleFlow().collect { println("Collector 1 received: $it") }
}
delay(500)
launch {
simpleFlow().collect { println("Collector 2 received: $it") }
}
delay(1500)
}
Output will show that the flow body (“Emitted Value …”) runs twice — once per collector, because it's cold.
If you want a **single producer** that shares emissions to multiple collectors, you can convert a cold flow into a hot flow. Some common ways:
shareIn(scope, started, replay): gives you a SharedFlow from a cold flow. :contentReference[oaicite:18]{index=18}stateIn(scope, started, initialValue): gives you a StateFlow from a cold flow. :contentReference[oaicite:19]{index=19}val shared: SharedFlow = numbersFlow()
.shareIn(
scope = CoroutineScope(Dispatchers.Default),
started = SharingStarted.WhileSubscribed(5000),
replay = 1
)
val state: StateFlow = numbersFlow()
.stateIn(
scope = CoroutineScope(Dispatchers.Default),
started = SharingStarted.Eagerly,
initialValue = 0
)
In this setup:
shared will replay the last value (because we set `replay = 1`) to new collectors. :contentReference[oaicite:20]{index=20}state always has the most recent value. If a new collector arrives, it immediately receives the current value. :contentReference[oaicite:21]{index=21}SharingStarted.WhileSubscribed ensures that the upstream producer only runs while there are subscribers, preventing unnecessary work. :contentReference[oaicite:22]{index=22}If you like, I can also include a diagram (ASCII or SVG) to visually show the differences between cold and hot flows, producers and consumers. Do you want me to do that?