Getting Started

Add Streams to your project and create your first reactive values.

1. Add the dependency

In your build.gradle.kts:

dependencies {
    implementation("io.peekandpoke.ultra:streams:0.107.2")
}

Find the latest version on Maven Central.

Streams is a Kotlin Multiplatform library — it works in commonMain, jvmMain, jsMain, and nativeMain (linuxX64, linuxArm64, macosX64, macosArm64, mingwX64).

2. Create a StreamSource

A StreamSource is a mutable reactive value. It always has a current value.

import io.peekandpoke.ultra.streams.StreamSource

val counter = StreamSource(0)

// Read the current value
println(counter())  // 0

// Set a new value
counter(1)
println(counter())  // 1

// Modify based on current value
counter.modify { this + 1 }
println(counter())  // 2

// Reset to initial
counter.reset()
println(counter())  // 0

For a stream that never changes — a constant value wrapped in a Stream<T> — use steady():

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"

3. Subscribe to changes

Subscribing immediately delivers the current value, then every subsequent change:

val name = StreamSource("Alice")

val unsubscribe = name.subscribeToStream { value ->
    println("Name is: $value")
}
// prints "Name is: Alice" immediately

name("Bob")
// prints "Name is: Bob"

name("Charlie")
// prints "Name is: Charlie"

// Stop listening
unsubscribe()

4. Transform with operators

Operators create new read-only streams derived from a source:

val source = StreamSource(5)

// map — transform each value
val doubled = source.map { it * 2 }
println(doubled())  // 10

// filter — only pass values matching a predicate
val big = source.filter(initial = 0) { it > 10 }
println(big())  // 0 (initial, since 5 doesn't match)

source(20)
println(big())     // 20
println(doubled()) // 40

5. Combine streams

Merge two streams into one:

val a = StreamSource(1)
val b = StreamSource(10)

val sum = a.combinedWith(b) { x, y -> x + y }
println(sum())  // 11

a(5)
println(sum())  // 15

b(100)
println(sum())  // 105

How subscriptions work

Streams use lazy subscriptions. When you chain operators like source.map .filter , no work happens until someone subscribes to the end of the chain. The subscription propagates upstream — and when the last subscriber unsubscribes, the entire chain tears down.

This means unused streams cost nothing.