Extending Streams

Build your own operators, bridge to Kotlin Flow, and understand the platform matrix.

Custom operators

Build operators by extending StreamMapper (transform) or StreamWrapper (pass-through):

import io.peekandpoke.ultra.streams.Stream
import io.peekandpoke.ultra.streams.StreamMapper

fun Stream<Int>.clamp(min: Int, max: Int): Stream<Int> =
    StreamMapper(
        wrapped = this,
        mapper = { it.coerceIn(min, max) }
    )

val source = StreamSource(50)
val clamped = source.clamp(0, 100)

source(150)
println(clamped())  // 100

For operators with internal state, extend StreamWrapperBase directly and override handleIncoming() and invoke().

Flow interop

Convert any stream to a Kotlin Flow:

import io.peekandpoke.ultra.streams.ops.asFlow

val source = StreamSource(1)

source.asFlow()
    .filter { it > 5 }
    .take(3)
    .collect { println(it) }

The Flow emits the current value immediately on collection, then subsequent values. Cancelling the collector unsubscribes automatically.

Platform summary