flowOn and Buffering

flowOn Changes the Upstream Thread

flowOn(dispatcher) changes the dispatcher of everything above it. The collector keeps running in its own context. This keeps the UI thread free while the producer does I/O.

flow { read DB }      ┐
  .map { parse }      ├─ runs on Dispatchers.IO
  .flowOn(IO)         ┘
  .collect { show }   ── runs on the collector's dispatcher (Main)

Example

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun where() = Thread.currentThread().name.substringBefore("-worker")

fun main() = runBlocking {
    val emittedOn = mutableSetOf<String>()
    val collectedOn = mutableSetOf<String>()
    flow {
        for (i in 1..3) {
            emittedOn += where()
            emit(i)
        }
    }
        .map { it * 10 }
        .flowOn(Dispatchers.IO)
        .collect { collectedOn += where() }
    println("Emitted on:   $emittedOn")
    println("Collected on: $collectedOn")
}

Output

Emitted on:   [DefaultDispatcher]
Collected on: [main]

buffer, conflate, collectLatest

By default the producer waits while the collector works. Three operators change that.

Operator	Producer	Collector gets
`buffer()`	keeps producing into a buffer	every value
`conflate()`	keeps producing, overwrites old	only the latest when it is ready
`collectLatest { }`	keeps producing	cancels the slow block when a new value comes

Example

Producer needs 200 ms per item, collector 300 ms.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.measureTimeMillis

fun producer() = flow { for (i in 1..3) { delay(200); emit(i) } }

fun main() = runBlocking {
    val plain = measureTimeMillis { producer().collect { delay(300) } }
    val buffered = measureTimeMillis { producer().buffer().collect { delay(300) } }
    println("No buffer  ~1500 ms: ${plain >= 1500}")
    println("With buffer ~1100 ms: ${buffered < 1400}")

    val seen = mutableListOf<Int>()
    flow { for (i in 1..5) { delay(200); emit(i) } }
        .conflate()
        .collect { seen += it; delay(500) }
    println("conflate saw: $seen")
}

Output

No buffer  ~1500 ms: true
With buffer ~1100 ms: true
conflate saw: [1, 3, 5]