Operators

Every operator takes a stream and returns a new stream. Chain them freely.

steady

Not an operator on an existing stream — a factory for a Stream<T> that always holds the same value. It never changes and never notifies beyond the initial delivery.

import io.peekandpoke.ultra.streams.steady

val pi = steady(3.14)

println(pi())  // 3.14

// Subscribers are called once with the value and never again
pi.subscribeToStream { println(it) }  // prints "3.14"

map

Transform each value:

val source = StreamSource(5)
val doubled = source.map { it * 2 }

println(doubled())  // 10

source(10)
println(doubled())  // 20

onEach

Execute a side effect for each value without changing it. Useful for logging:

val source = StreamSource(1)

val stream = source
    .onEach { println("Saw: $it") }
    .map { it * 10 }

indexed

Pairs each value with its index (starting at 0):

val source = StreamSource("a")
val indexed = source.indexed()

// Pair(0, "a"), then Pair(1, "b"), Pair(2, "c")...

filter

Only pass values matching a predicate. Since streams always have a value, you must provide an initial value for when nothing matches yet:

val source = StreamSource(3)

val big = source.filter(initial = 0) { it > 10 }
println(big())  // 0

source(20)
println(big())  // 20

source(5)
println(big())  // 20 (keeps last matching value)

Nullable filter

When you don't want to provide an initial value, use the nullable variant. It returns null until a value matches:

val source = StreamSource(3)
val big = source.filter { it > 10 }  // Stream<Int?>

println(big())  // null (nothing matched yet)

source(20)
println(big())  // 20

filterNotNull

Remove nulls from a nullable stream. The result is still Stream<String?> because the current value may be null if no non-null value has arrived yet:

val source = StreamSource<String?>(null)
val safe = source.filterNotNull()  // Stream<String?>

println(safe())  // null (no non-null value yet)

source("hello")
println(safe())  // "hello"

source(null)
println(safe())  // "hello" (keeps last non-null)

filterIsInstance

Filter by type, using Kotlin's reified type parameter:

val source = StreamSource<Any>("text")
val strings = source.filterIsInstance<String>()  // Stream<String?>

println(strings())  // "text"

source(42)
println(strings())  // "text" (42 is not a String, keeps last match)

source("new text")
println(strings())  // "new text"

distinct

Only publish when the value actually changes:

val source = StreamSource(1)
val distinct = source.distinct()

val received = mutableListOf<Int>()
distinct.subscribeToStream { received.add(it) }

source(1)  // same — skipped
source(2)  // different — published
source(2)  // same — skipped
source(3)  // different — published

// received = [1, 2, 3]

Use distinctStrict() for reference equality (!==) instead of structural equality (!=).

fallback

Replace null values with a default:

val source = StreamSource<String?>(null)

val safe = source.fallbackTo("default")
println(safe())  // "default"

source("hello")
println(safe())  // "hello"

// Or compute the fallback dynamically
val computed = source.fallbackBy { "fallback-value" }