Wait, why is my producer slowing down just because the collector is slow?

This is the first thing that trips people up with Flow. By default, Kotlin Flows are sequential. If you have a producer that emits a value and a collector that takes a long time to process it, the producer actually suspends. It literally stops and waits for the collector to finish before it even attempts to emit the next item.

I like to think of it as a relay race. The runner (producer) can't just throw the baton forward and keep running; they have to wait until the next person actually grabs it. In a real app, if you're fetching data from a sensor every 10ms but saving it to a database that takes 50ms, your sensor loop is going to lag. It's not "dropping" data; it's just being held back by the consumer.

val sensorFlow = flow {
    for (i in 1..3) {
        println("Emitting $i")
        emit(i)
        println("Producer is free after $i")
    }
}

sensorFlow.collect { value ->
    delay(100) // Simulate slow processing
    println("Collected $value")
}
// Output:
// Emitting 1
// Collected 1
// Producer is free after 1
// Emitting 2... (Notice the producer waits for the collector!)

Can I just skip the old data and only care about the most recent update?

In many cases, you don't actually need every single emission. If you're building a UI that shows the current price of Bitcoin, the user doesn't care about the price from 2 seconds ago if a new price just arrived. This is where conflate() comes in.

When you call conflate(), you're telling the Flow: "If the collector is too slow, just drop the intermediate values and give it the most recent one whenever it's ready." It essentially turns your flow into a buffer of size one that always overwrites the old value.

val priceFlow = flow {
    for (i in 1..5) {
        emit(i) 
        delay(10) // Fast producer
    }
}

priceFlow
    .conflate() 
    .collect { value ->
        delay(100) // Slow collector
        println("Processed price: $value")
    }
// You'll notice it skips 2, 3, and 4, jumping straight to 5.

What if I want to queue up a few items without blocking the producer?

If you can't afford to lose data but you don't want the producer to sit idle, you need a buffer(). This runs the producer in a separate coroutine, allowing it to keep pumping out values into a channel while the collector works through them at its own pace.

I usually reach for buffer() when I'm doing something like logging or analytics. I want the main logic of my app to keep moving, and I'm okay with the analytics being processed a few milliseconds behind. You can even specify the capacity of the buffer if you're worried about memory usage.

val logFlow = flow {
    for (i in 1..3) {
        println("Logging event $i")
        emit(i)
    }
}

logFlow
    .buffer(capacity = 2) 
    .collect { value ->
        delay(100)
        println("Saved event $value to disk")
    }
// Here, "Logging event 1, 2, 3" will likely print almost instantly,
// and then the collector will slowly process them.

What's the difference between conflate and collectLatest?

This is a subtle but critical distinction. conflate() lets the collector finish what it's doing, but skips the values that arrived in the meantime. collectLatest() is more aggressive: the moment a new value is emitted, it cancels the collector's current block of code and starts over with the new value.

Imagine a search bar. If the user types "K", you start a network request. If they immediately type "Ko", you don't want to wait for the "K" request to finish and then show the results—that's a waste of resources. You want to kill the "K" request immediately and start the "Ko" request. That's exactly what collectLatest() does.

val searchFlow = flow {
    emit("K")
    delay(50)
    emit("Ko")
    delay(50)
    emit("Kot")
}

searchFlow.collectLatest { query ->
    println("Searching for $query...")
    delay(100) // Simulate network lag
    println("Results for $query arrived!")
}
// Only "Results for Kot arrived!" will actually print, 
// because the previous collections were cancelled.

📋 Practical Task

Exercise: Building a High-Frequency Stock Ticker Buffer

You are building a stock market dashboard. The backend sends price updates every 50ms, but your UI rendering engine is heavy and takes 200ms to update the chart. If you don't handle backpressure, the UI will lag behind the real market by several seconds.

Your Task:

• Create a flow that emits integers from 1 to 20, with a delay(50) between each emission.
• Implement a collector that simulates a heavy UI render using delay(200).
• First, run the code without any backpressure handling and observe how long it takes to complete.
• Apply the conflate() operator to ensure the UI always shows the most recent price and doesn't lag behind.
• Verify that the output skips intermediate numbers and finishes significantly faster.