"Flowing Data" — Flow, Channels and Virtual Threads

// Fragment of story/ReactorDashboard.kt
// Kabir's first instinct, written the WebFlux way. Every step is an operator.
fun liveStock(store: String): Flux<String> =
    Flux.interval(Duration.ofMillis(10))
        .take(3)
        .flatMap { tick -> fetchLevel(store, tick) }
        .map { "${it.store} ${it.sku}: ${it.units}" }
        .onErrorResume { Flux.empty() } // "a bad reading shouldn't take the dashboard down"

interface Flow<T>      { suspend fun collect(collector: FlowCollector<T>) }
interface FlowCollector<T> { suspend fun emit(value: T) }

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext

data class StockLevel(val store: String, val sku: String, val units: Int)

// A cold flow: the block runs once per collector, when it collects. Nothing runs before that.
fun stockLevels(store: String): Flow<StockLevel> = flow {
    println("scanner connected for $store")
    for (tick in 0..3) {
        delay(10) // suspends between values; no thread is held
        emit(StockLevel(store, "SHW-1001", 120 - tick * 7))
    }
}

fun main() = runBlocking {
    val pune = stockLevels("PUN-014") // nothing printed yet: building a flow runs nothing

    // Operators read like the Stream API, but they are suspend-aware and run in the collector's coroutine.
    pune
        .filter { it.units < 115 }
        .map { "${it.store}: ${it.units}" }
        .take(2)
        .collect { println(it) }
    // -> scanner connected for PUN-014
    // -> PUN-014: 113
    // -> PUN-014: 106

    // A second collection runs the producer again.
    println(pune.toList().size)
    // -> scanner connected for PUN-014
    // -> 4

    // flowOn changes where the UPSTREAM runs; the collector stays where it is.
    stockLevels("MUM-002")
        .map { Thread.currentThread().name.startsWith("DefaultDispatcher") }
        .flowOn(Dispatchers.Default)
        .take(1)
        .collect { println("produced on Default: $it") }
    // -> scanner connected for MUM-002
    // -> produced on Default: true

    // Emitting from a different context inside flow { } breaks the flow's contract.
    try {
        flow { withContext(Dispatchers.Default) { emit(1) } }.collect { }
    } catch (e: IllegalStateException) {
        println(e.message!!.substringBefore(":")) // -> Flow invariant is violated
    }
}

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

// An operator is a new flow whose collect collects the upstream. The library's map and filter
// have the same shape; they use an internal builder (unsafeTransform) that skips the safety
// checks flow { } performs on every emit, which this hand-written version keeps.
fun <T, R> Flow<T>.mapped(transform: suspend (T) -> R): Flow<R> = flow {
    collect { value -> emit(transform(value)) }
}

fun readings() = flow {
    println("emit 1"); emit(1)
    println("emit 2"); emit(2)
}

fun main() = runBlocking {
    // emit calls straight down into the collector: each value goes all the way through
    // before the producer gets the next line. One coroutine, one call stack.
    readings().mapped { println("map $it"); it * 10 }.collect { println("collect $it") }
    // -> emit 1
    // -> map 1
    // -> collect 10
    // -> emit 2
    // -> map 2
    // -> collect 20

    // buffer splits the flow in two: everything above it runs in a new producer coroutine,
    // which sends into a channel; the collector receives from it. Now the producer runs ahead.
    readings().mapped { println("map $it"); it * 10 }.buffer().collect { println("collect $it") }
    // -> emit 1
    // -> map 1
    // -> emit 2
    // -> map 2
    // -> collect 10
    // -> collect 20
}

import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flatMapConcat
import kotlinx.coroutines.flow.flatMapMerge
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.runTest

// The gateway client as a suspend function: one value, so no Mono. (StockLevel is from ColdFlows.kt.)
suspend fun fetchLevel(store: String, tick: Int): StockLevel {
    delay(5) // the remote call
    val raw = if (store == "MUM-002" && tick == 1) "12O" else "${120 - tick * 7}"
    return StockLevel(store, "SHW-1001", raw.toInt())
}

// The opening's Flux chain, rewritten. Only the timer needed a stream; the rest is a loop that waits.
fun liveStock(store: String): Flow<String> = flow {
    for (tick in 0 until 3) {
        delay(10)
        emit(fetchLevel(store, tick)) // Reactor needed flatMap for this call
    }
}.map { "${it.store} ${it.sku}: ${it.units}" }

// A lookup whose latency depends on the item: later items answer first.
fun slowLookup(n: Int): Flow<Int> = flow {
    delay((4 - n) * 10L)
    emit(n)
}

@OptIn(ExperimentalCoroutinesApi::class) // flatMapMerge and flatMapConcat need opt-in in kotlinx.coroutines 1.11.0
fun main() = runTest {
    liveStock("PUN-014").collect { println(it) }
    // -> PUN-014 SHW-1001: 120
    // -> PUN-014 SHW-1001: 113
    // -> PUN-014 SHW-1001: 106

    // No operator swallows anything by default: the failure reaches the code that collected.
    try {
        liveStock("MUM-002").collect { println(it) }
    } catch (e: NumberFormatException) {
        println("MUM-002 failed: ${e.message}")
    }
    // -> MUM-002 SHW-1001: 120
    // -> MUM-002 failed: For input string: "12O"

    // map with a suspend call: one at a time, in order. This is what most Reactor flatMaps meant.
    println(flowOf(1, 2, 3).map { n -> slowLookup(n).toList().single() }.toList()) // -> [1, 2, 3]
    // flatMapMerge: concurrent, so results arrive in completion order, like Reactor's flatMap.
    println(flowOf(1, 2, 3).flatMapMerge { slowLookup(it) }.toList()) // -> [3, 2, 1]
    // flatMapConcat: one inner flow after another, like Reactor's concatMap.
    println(flowOf(1, 2, 3).flatMapConcat { slowLookup(it) }.toList()) // -> [1, 2, 3]
}

import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.FlowPreview
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.conflate
import kotlinx.coroutines.flow.debounce
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.currentTime
import kotlinx.coroutines.test.runTest

// A scanner that emits a reading every 100 ms (of virtual time: runTest skips the waiting).
// Each consumer below takes 300 ms per reading.
fun scans(count: Int) = flow {
    repeat(count) {
        delay(100)
        emit(it)
    }
}

// Runs a block and prints what it collected and how much virtual time it took.
@OptIn(ExperimentalCoroutinesApi::class) // currentTime is experimental in kotlinx-coroutines-test 1.11.0
suspend fun TestScope.timed(label: String, block: suspend () -> List<Int>) {
    val start = currentTime
    val seen = block()
    println("$label: saw $seen in ${currentTime - start} ms")
}

@OptIn(FlowPreview::class) // debounce is a Flow preview API in kotlinx.coroutines 1.11.0
fun main() = runTest {
    // No buffer: producer and consumer take turns, 100 + 300 ms per reading.
    timed("sequential") { buildList { scans(5).collect { delay(300); add(it) } } }
    // -> sequential: saw [0, 1, 2, 3, 4] in 2000 ms

    // buffer: the producer runs ahead in its own coroutine; the consumer sets the pace.
    timed("buffer") { buildList { scans(5).buffer().collect { delay(300); add(it) } } }
    // -> buffer: saw [0, 1, 2, 3, 4] in 1600 ms

    // conflate: the consumer only ever sees the latest value; intermediate ones are dropped.
    timed("conflate") { buildList { scans(5).conflate().collect { delay(300); add(it) } } }
    // -> conflate: saw [0, 2, 4] in 1000 ms

    // collectLatest: a new value cancels the work on the previous one.
    timed("collectLatest") { buildList { scans(5).collectLatest { delay(300); add(it) } } }
    // -> collectLatest: saw [4] in 800 ms

    // debounce: emit only after 250 ms of quiet. A burst of scans becomes one update.
    val burst = flow {
        emit(1); delay(50); emit(2); delay(50); emit(3) // a burst
        delay(400)
        emit(4) // a lone scan
    }
    timed("debounce") { burst.debounce(250).toList() }
    // -> debounce: saw [3, 4] in 500 ms
    // (the last value goes out as soon as the flow completes, without waiting 250 ms)
}

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.runTest

class ShelfSensor {
    // StateFlow: always has a value; new collectors get the current one; equal updates are skipped.
    private val _units = MutableStateFlow(120)
    val units = _units.asStateFlow() // read-only view for the outside world

    fun scan(newUnits: Int) {
        _units.value = newUnits
    }
}

fun main() = runTest {
    val sensor = ShelfSensor()
    val seen = mutableListOf<Int>()
    val watcher = launch { sensor.units.toList(seen) }
    delay(1)
    for (units in listOf(118, 118, 110, 110, 95)) {
        sensor.scan(units)
        delay(1) // give the watcher time to see every update, so nothing is conflated
    }
    watcher.cancel()
    println(seen) // -> [120, 118, 110, 95]
    // (the repeated 118 and 110 were equal to the current value, so they emitted nothing)

    // Mutating the object inside a StateFlow emits nothing: the reference didn't change.
    val skus = MutableStateFlow(mutableListOf("SHW-1001"))
    var updates = 0
    val counter = launch { skus.collect { updates++ } }
    delay(1)
    skus.value.add("SHW-2040") // invisible to collectors
    delay(1)
    counter.cancel()
    println("updates seen: $updates") // -> updates seen: 1

    // SharedFlow: a broadcast of events. No current value; replay decides what late subscribers get.
    val scans = MutableSharedFlow<String>(replay = 1)
    scans.emit("PUN-014 aisle 7")
    scans.emit("PUN-014 aisle 8")
    val late = scans.take(1).toList()
    println(late) // -> [PUN-014 aisle 8]

    // tryEmit (the non-suspending emit) on a SharedFlow with no buffer: with nobody subscribed,
    // the value is dropped and tryEmit reports success; with a subscriber, it fails.
    val alerts = MutableSharedFlow<String>()
    println(alerts.tryEmit("low stock: SHW-2040")) // -> true
    val listener = launch { alerts.collect { } }
    delay(1)
    println(alerts.tryEmit("low stock: SHW-3001")) // -> false
    listener.cancel()

    // stateIn: one upstream shared by every collector. WhileSubscribed(5_000) starts it with the
    // first subscriber and stops it 5 s (of virtual time, here) after the last one leaves.
    var polls = 0
    val stockPoll = flow {
        polls++ // one database poll loop per start
        while (true) {
            emit(100)
            delay(1_000)
        }
    }
    val shared = stockPoll.stateIn(backgroundScope, SharingStarted.WhileSubscribed(5_000), initialValue = 0)
    val dashboards = List(2) { launch { shared.collect { } } }
    delay(10)
    println("dashboards=2, polls=$polls") // -> dashboards=2, polls=1
    dashboards.forEach { it.cancel() }
    delay(6_000) // nobody subscribed for more than 5 s: the poll stops
    val again = launch { shared.collect { } }
    delay(10)
    println("after a quiet spell, polls=$polls") // -> after a quiet spell, polls=2
    again.cancel()
}

import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.produce
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.util.concurrent.ConcurrentHashMap

// A channel is a queue between coroutines: each element goes to exactly ONE receiver.
// That makes it the tool for distributing work, where a Flow is the tool for describing a stream.
@OptIn(ExperimentalCoroutinesApi::class) // produce { } is experimental in kotlinx.coroutines 1.11.0
fun main() = runBlocking {
    val labelsPrinted = ConcurrentHashMap<String, Int>()

    coroutineScope {
        val jobs = produce(capacity = Channel.BUFFERED) {
            for (aisle in 1..12) send("aisle-$aisle")
        } // closed automatically when the producer finishes

        repeat(3) { printer ->
            launch {
                for (job in jobs) { // fan-out: three printers share the queue
                    delay(10)
                    labelsPrinted.merge("printer-$printer", 1, Int::plus)
                }
            }
        }
    }
    println("${labelsPrinted.values.sum()} labels, ${labelsPrinted.size} printers busy") // -> 12 labels, 3 printers busy
}

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

// A vendor's shelf-scanner SDK: listener-based, and the caller must unregister by hand.
public final class ScannerSdk {
    public interface ScanListener {
        void onScan(String sku, int units);

        void onError(Exception e);
    }

    private final List<ScanListener> listeners = new CopyOnWriteArrayList<>();

    public void addListener(ScanListener listener) { listeners.add(listener); }

    public void removeListener(ScanListener listener) { listeners.remove(listener); }

    public int listenerCount() { return listeners.size(); }

    // The real SDK calls listeners from its own I/O thread; the example calls this directly.
    public void deliver(String sku, int units) {
        for (ScanListener listener : listeners) listener.onScan(sku, units);
    }
}

import kotlinx.coroutines.async
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.yield

data class Scan(val sku: String, val units: Int)

// A Java listener API as a Flow. callbackFlow gives the block a channel: callbacks push into it,
// the collector reads from it, and awaitClose keeps the listener registered until the collector stops.
fun ScannerSdk.scans(): Flow<Scan> = callbackFlow {
    val listener = object : ScannerSdk.ScanListener {
        override fun onScan(sku: String, units: Int) {
            trySend(Scan(sku, units)) // callbacks can't suspend, so the non-suspending send
        }

        override fun onError(e: Exception) {
            close(e) // fails the flow with the SDK's exception
        }
    }
    addListener(listener)
    awaitClose { removeListener(listener) } // runs on completion, failure or cancellation
}

fun main() = runBlocking {
    val sdk = ScannerSdk()
    val firstTwo = async { sdk.scans().take(2).toList() }
    while (sdk.listenerCount() == 0) yield() // let the collector start and register

    sdk.deliver("SHW-1001", 40)
    sdk.deliver("SHW-2040", 8)
    sdk.deliver("SHW-3001", 75) // buffered, then discarded when take(2) cancels the flow

    println(firstTwo.await()) // -> [Scan(sku=SHW-1001, units=40), Scan(sku=SHW-2040, units=8)]
    println("listeners after: ${sdk.listenerCount()}") // -> listeners after: 0
}

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.reactive.asFlow
import kotlinx.coroutines.reactive.awaitSingle
import kotlinx.coroutines.reactor.awaitSingleOrNull
import kotlinx.coroutines.reactor.asFlux
import kotlinx.coroutines.reactor.flux
import kotlinx.coroutines.reactor.mono
import kotlinx.coroutines.runBlocking
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono

// A Reactor-based client from another team.
fun legacyPrice(sku: String): Mono<Long> = Mono.just(if (sku == "SHW-1001") 16_500L else 72_000L)

fun legacyStock(): Flux<Int> = Flux.just(120, 113, 106)

fun main() = runBlocking {
    // Reactor -> coroutines: await a Mono, collect a Flux as a Flow.
    println(legacyPrice("SHW-1001").awaitSingle()) // -> 16500
    // An empty Mono ("not found" in most Java code) makes awaitSingle throw; use awaitSingleOrNull.
    println(Mono.empty<Long>().awaitSingleOrNull()) // -> null
    try {
        Mono.empty<Long>().awaitSingle()
    } catch (e: NoSuchElementException) {
        println("awaitSingle on empty: ${e::class.simpleName}") // -> awaitSingle on empty: NoSuchElementException
    }
    println(legacyStock().asFlow().map { it - 100 }.toList()) // -> [20, 13, 6]

    // Coroutines -> Reactor: suspend code behind a Mono or a Flux, for APIs that expect them.
    val price: Mono<String> = mono { "₹${legacyPrice("SHW-2040").awaitSingle() / 100}" }
    println(price.block()) // -> ₹720

    val ticks: Flux<Int> = flux {
        for (i in 1..3) {
            delay(5)
            send(i) // flux { } is a producer: send, not emit
        }
    }
    println(ticks.collectList().block()) // -> [1, 2, 3]

    // A Flow as a Flux, e.g. to return from a WebFlux controller that expects Publisher types.
    println(flow { emit("a"); emit("b") }.asFlux().collectList().block()) // -> [a, b]
}

import kotlinx.coroutines.asCoroutineDispatcher
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
import java.util.concurrent.Executors

// A blocking SDK call that can't be rewritten: it holds its thread for the whole call.
fun blockingSupplierCall(i: Int): Int {
    Thread.sleep(100)
    return i
}

// The two models combine: coroutines for structure, a virtual-thread dispatcher for blocking calls.
fun main() = runBlocking {
    Executors.newVirtualThreadPerTaskExecutor().asCoroutineDispatcher().use { virtualThreads ->
        val results = withContext(virtualThreads) {
            (1..200).map { i -> async { blockingSupplierCall(i) } }.awaitAll()
        }
        println("${results.size} blocking calls, sum ${results.sum()}") // -> 200 blocking calls, sum 20100
        println(withContext(virtualThreads) { Thread.currentThread().isVirtual }) // -> true
    }
}

// Fragment of dashboard/StockBoard.kt
// Step 1: a scan from a shelf scanner.
data class ShelfScan(val store: String, val sku: String, val units: Int)

// Step 2: the board holds state: the latest units per "store/sku", as an immutable map.
class StockBoard(private val reorderAt: Int = 10) {
    private val _levels = MutableStateFlow<Map<String, Int>>(emptyMap())
    val levels: StateFlow<Map<String, Int>> = _levels.asStateFlow()

    // Atomic read-modify-write: safe when scans arrive from many coroutines (or threads) at once.
    fun record(scan: ShelfScan) = _levels.update { it + ("${scan.store}/${scan.sku}" to scan.units) }

    // Step 3: alerts derived from the state, emitted only when the low-stock set changes.
    val lowStock: Flow<Set<String>> = levels
        .map { all -> all.filterValues { it <= reorderAt }.keys }
        .distinctUntilChanged()
}

// Step 4: the same stream as a Flux, for Reactor-based callers.
fun StockBoard.lowStockFeed(): Flux<Set<String>> = lowStock.asFlux()

// Fragment of dashboard/StockBoard.kt
// Step 5: drive it under virtual time.
@OptIn(ExperimentalCoroutinesApi::class) // runCurrent is experimental in kotlinx-coroutines-test 1.11.0
fun main() = runTest {
    val board = StockBoard()
    val alerts = mutableListOf<Set<String>>()
    backgroundScope.launch { board.lowStock.toList(alerts) }

    for (scan in listOf(
        ShelfScan("PUN-014", "SHW-1001", 40),
        ShelfScan("PUN-014", "SHW-2040", 8),
        ShelfScan("MUM-002", "SHW-1001", 5),
        ShelfScan("PUN-014", "SHW-2040", 60),
    )) {
        board.record(scan)
        runCurrent() // let the collector see each state; a StateFlow keeps only the latest
    }

    println(board.levels.value) // -> {PUN-014/SHW-1001=40, PUN-014/SHW-2040=60, MUM-002/SHW-1001=5}
    alerts.forEach(::println)
    // -> []
    // -> [PUN-014/SHW-2040]
    // -> [PUN-014/SHW-2040, MUM-002/SHW-1001]
    // -> [MUM-002/SHW-1001]
}

import com.shelfwise.part11.dashboard.ShelfScan
import com.shelfwise.part11.dashboard.StockBoard
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.advanceTimeBy
import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Test

@OptIn(ExperimentalCoroutinesApi::class) // advanceTimeBy, runCurrent, currentTime
class StockBoardTest {
    @Test
    fun `a scan at or below the reorder point raises an alert`() = runTest {
        val board = StockBoard(reorderAt = 10)

        board.record(ShelfScan("PUN-014", "SHW-2040", 10))

        assertEquals(setOf("PUN-014/SHW-2040"), board.lowStock.first())
    }

    @Test
    fun `virtual time skips the waiting`() = runTest {
        val board = StockBoard()
        launch {
            delay(60_000) // a scanner that reports once a minute
            board.record(ShelfScan("MUM-002", "SHW-1001", 3))
        }

        advanceTimeBy(60_001) // instant in real time
        runCurrent()

        assertEquals(3, board.levels.value["MUM-002/SHW-1001"])
        assertEquals(60_001, testScheduler.currentTime)
    }
}

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    // A Java-style try/catch around emit also catches the COLLECTOR's exceptions.
    // Emitting again after that breaks the flow's contract.
    val readings = flow {
        try {
            emit(1)
        } catch (e: IllegalStateException) {
            emit(-1) // "fallback"
        }
    }
    try {
        readings.collect { if (it == 1) throw IllegalStateException("display offline") }
    } catch (e: IllegalStateException) {
        println(e.message!!.lineSequence().first()) // -> Flow exception transparency is violated:
    }

    // catch { } only sees upstream exceptions; the collector's own failure goes straight to the caller.
    try {
        flow { emit(1) }
            .catch { println("never reached") }
            .collect { throw IllegalStateException("display offline") }
    } catch (e: IllegalStateException) {
        println("collector failed: ${e.message}") // -> collector failed: display offline
    }
}