The Flow contract, and writing your own operators
An operator is just an extension function on Flow. The two rules the runtime enforces are what stop you writing a broken one — and both failures are famous.
Explore the full explanation 8 ideas
There is no operator registry in Kotlin. `map`, `filter`, `debounce` and every other operator is an ordinary extension function on `Flow<T>` that returns a `Flow<R>`, and yours will be too: `fun Flow<Order>.onlyPaid(): Flow<Order> = filter { it.paid }` is a real operator, indistinguishable from a built-in at the call site. That is the whole extensibility story, and it is why the standard library can afford to be small — anything missing you write once, name well, and unit-test on its own.
The `flow { }` builder enforces two invariants, and knowing them is the difference between an operator that works and one that fails in a way nobody can debug. The first is **context preservation**: only the coroutine that runs the builder may call `emit`. Writing `flow { withContext(Dispatchers.IO) { emit(x) } }` throws `IllegalStateException: Flow invariant is violated`, and the library is right to refuse — a flow whose emissions arrive from arbitrary threads has no coherent cancellation or exception story. To move upstream work, use `.flowOn(dispatcher)`, which changes where the producer runs without changing who calls `emit`.
The second invariant is **exception transparency**, and it is the subtler one. A collector's exception propagates *back through* the `emit` call that invoked it — so a `try/catch(Throwable)` wrapped around your `emit` catches failures thrown by **downstream code you know nothing about**. Your operator then silently handles somebody else's bug, and the failure disappears entirely: nothing escapes the `collect`, nothing is logged, and the stream just ends early. This is legal Kotlin that no compiler can warn you about, which is exactly why `catch { }` exists as a separate operator.
The declarative error operators are **positional**: each sees only what is *above* it in the chain. `.catch { }` handles upstream failures and may `emit` a replacement value, so a pipeline that ran out of retries can degrade instead of throwing. Placed before the stage that fails, it is dead code. `.onCompletion { cause -> … }` runs on success, on failure and on cancellation, receives the cause (`null` when all went well) and — unlike `catch` — does **not** handle anything, which makes it the right home for logging and metrics. `.retryWhen { cause, attempt -> … }` re-subscribes to the whole upstream, with `attempt` counting retries from zero.
For an operator that only transforms values, `transform { }` is the tool: it hands you each upstream value and lets you `emit` zero, one or many results. `map` is `transform { emit(f(it)) }`; `filter` is `transform { if (p(it)) emit(it) }`; a flat-map over a list is one line. Because you are inside the builder's own coroutine, all of that is legal — the invariants only bite when a *different* coroutine wants to produce.
And that is exactly when you need `channelFlow { }`. It runs your block in a `ProducerScope` backed by a channel, so `send` is safe from any child coroutine — which is what an operator with a **timer**, a **callback** or **several concurrent producers** requires. A batching operator that flushes on size *or* on a timeout cannot be written with `flow { }`, because the timer coroutine would have to emit; with `channelFlow { }` it is a `launch { delay(t); flush() }` and a lock around the flush. `callbackFlow { }` is the same builder specialised for listener APIs, with a mandatory `awaitClose { }` that the runtime checks for — because a callback bridge without cleanup is a guaranteed leak.
Two design rules keep custom operators honest. **Return a cold Flow**: your function should do no work when called, only when collected, so it composes like every other operator and can be safely retried, cached or collected twice. And **do not swallow cancellation**: if your operator catches broadly for its own reasons, rethrow `CancellationException` first, or a collector that walks away leaves your operator running.
Finally, test them like ordinary functions, because that is what they are. `runTest` gives you virtual time, so an operator with a timeout is asserted by *schedule* — `[1, 2]@100` — rather than by sleeping and hoping. Build the source with `flow { delay(…); emit(…) }` to script an exact timeline, collect with `toList()`, and assert on both the values and `currentTime`. An operator whose behaviour you can pin to the millisecond is one you can refactor without fear.
Let’s figure it out.
Ask me about this course or where to start.
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.
Arena IDEimport 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 = 0return 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.
Arena IDEimport 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? = nullval 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 shortflush()} 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.
Arena IDEimport kotlinx.coroutines.*import kotlinx.coroutines.channels.awaitCloseimport kotlinx.coroutines.flow.*import kotlinx.coroutines.test.*class Ticker { // a legacy push APIprivate var listener: ((Int) -> Unit)? = nullval 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 suspendawaitClose { 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 subscriberepeat(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.
Arena IDE🧠 Check your understanding
0/1 · 0/1 answered1. You write a custom operator as `flow { try { source.collect { emit(transform(it)) } } catch (e: Throwable) { emit(fallback) } }`. What is wrong with it?