El contrato de Flow y escribir tus propios operadores

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

fun main() = runBlocking {
    // 1. Context preservation: only the builder's coroutine may emit.
    val illegal = flow { withContext(Dispatchers.Default) { emit(1) } }
    try { illegal.collect { } }
    catch (e: Throwable) { println("1 = " + e::class.simpleName) }

    // 2. Exception transparency: the collector's failure unwinds back THROUGH
    //    emit, so a broad catch in the builder eats somebody else's bug.
    val leaky = flow {
        try { emit(1); emit(2) }
        catch (e: Throwable) { println("2 = swallowed '" + e.message + "' (not mine!)") }
    }
    leaky.collect { if (it == 2) throw IllegalStateException("collector blew up") }
    println("2 = ...and nothing escaped the collect")

    // 3. The legal way: catch() sees UPSTREAM failures only, and may emit.
    val guarded = flow<Int> { emit(1); throw IllegalStateException("upstream") }
        .catch { println("3 = caught '" + it.message + "', degrading"); emit(-1) }
    println("3 = " + guarded.toList())
}

Both invariants, demonstrated. Emitting from another context is refused outright; a try/catch around emit silently swallows the COLLECTOR's failure, which is the bug catch { } exists to prevent.

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

// Real operators, written the way the standard library writes them: the
// per-collection state lives INSIDE the function, so each collection is
// independent and the operator stays cold.
fun <T, R> Flow<T>.mapIndexedNotNull(f: (Int, T) -> R?): Flow<R> {
    var i = 0
    return transform { value -> f(i++, value)?.let { emit(it) } }
}

// A domain operator reads at the call site like a built-in.
data class Order(val id: Int, val paid: Boolean)
fun Flow<Order>.onlyPaid(): Flow<Order> = filter { it.paid }

fun main() = runBlocking {
    val orders = flowOf(Order(1, true), Order(2, false), Order(3, true))
    println("onlyPaid = " + orders.onlyPaid().map { it.id }.toList())

    println("labelled = " + flowOf("a", "b", "c")
        .mapIndexedNotNull { i, s -> if (i % 2 == 0) "" + i + s else null }
        .toList())
}

transform { } is enough for anything that only reshapes values — zero, one or many emissions per input, all inside the builder's own coroutine.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.sync.*
import kotlinx.coroutines.test.*

@OptIn(ExperimentalCoroutinesApi::class)
fun <T> Flow<T>.chunked(size: Int, timeoutMs: Long): Flow<List<T>> = channelFlow {
    val buf = mutableListOf<T>()
    var timer: Job? = null
    val lock = Mutex()

    suspend fun flush() = lock.withLock {
        if (buf.isNotEmpty()) { send(buf.toList()); buf.clear() }
    }

    collect { v ->
        lock.withLock { buf.add(v) }
        if (buf.size >= size) {
            timer?.cancel(); timer = null      // or it would cut the NEXT batch short
            flush()
        } else if (timer == null) {
            timer = launch { delay(timeoutMs); flush(); timer = null }
        }
    }
    timer?.cancel()
    flush()                                    // never lose the final partial batch
}

@OptIn(ExperimentalCoroutinesApi::class)
fun main() = runTest {
    // 1,2 arrive together; 3,4,5 arrive after a long gap.
    val src = flow {
        emit(1); delay(10); emit(2); delay(290)
        emit(3); delay(10); emit(4); delay(10); emit(5)
    }
    src.chunked(size = 3, timeoutMs = 100).collect { println(it.toString() + "@" + currentTime) }
}

The operator flow { } cannot express: a batch that flushes on SIZE or on a TIMEOUT needs a second coroutine to be able to emit — so channelFlow, and a lock so the two triggers cannot send the same buffer twice.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*

class Ticker {                                  // a legacy push API
    private var listener: ((Int) -> Unit)? = null
    val log = mutableListOf<String>()
    fun register(l: (Int) -> Unit) { listener = l; log.add("registered") }
    fun unregister() { listener = null; log.add("unregistered") }
    fun push(v: Int) { listener?.invoke(v) }
}

fun Ticker.prices(): Flow<Int> = callbackFlow {
    register { v -> trySend(v) }                // the callback cannot suspend
    awaitClose { unregister() }                 // runs on completion AND cancel
}

@OptIn(ExperimentalCoroutinesApi::class)
fun main() = runTest {
    val ticker = Ticker()
    val got = mutableListOf<Int>()

    val job = launch { ticker.prices().take(2).collect { got.add(it) } }
    runCurrent()                                // let the collector subscribe
    repeat(3) { ticker.push(it + 1) }
    runCurrent()
    job.join()

    println("got       = " + got)
    println("lifecycle = " + ticker.log.joinToString(","))
}

callbackFlow for a listener API. awaitClose is mandatory — the runtime throws without it — because a callback bridge that never unregisters is a guaranteed leak.