Combining & Accumulating

Merge streams together and build up state over time.

combinedWith

Merge two streams with a combinator function. Publishes whenever either source changes:

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

pairedWith

Shorthand for combining into a Pair:

val pair = a.pairedWith(b)
println(pair())  // Pair(1, 10)

switchMap

Follow a stream that lives inside another stream's value. When the outer value changes, the old inner stream is unsubscribed and the new one is subscribed in its place:

class Player(val diagnostics: Stream<Int>)

val playerA = Player(StreamSource(60))
val playerB = Player(StreamSource(30))

val players = StreamSource(playerA)
val fps = players.switchMap { it.diagnostics }

fps.subscribeToStream { println(it) }  // 60

players(playerB)                       // 30 — playerA's stream is released

The selector runs on every value read while nothing is subscribed, and never at construction time — so keep it free of side effects.

Switching releases the old inner stream before subscribing the new one. An upstream shared by all the inner streams — one ticker() feeding every player — therefore sees its last subscriber leave and restarts on each switch.

switchMapNotNull

When the outer stream is nullable, switchMapNotNull runs the selector only for non-null values and publishes null the rest of the time:

val players = StreamSource<Player?>(null)

val fps: Stream<Int?> = players.switchMapNotNull { it.diagnostics }

println(fps())  // null — no player

players(playerA)
println(fps())  // 60

players(null)
println(fps())  // null again — the last value is NOT held

Unlike foldNotNull and historyOfNonNull further down this page, it resets rather than keeping what it had. Compose with fallbackTo(0) when you want a non-null result.

Returning the same stream twice

Inner streams are compared by identity. Selecting the same instance again keeps the subscription and publishes nothing — but a selector that builds a stream returns a new instance every time it runs, so it resubscribes on every outer value:

// Resubscribes on every outer value: map() allocates a new stream per call
players.switchMapNotNull { it.diagnostics.map { d -> d * 2 } }

// Stable instance: hoist the derived stream into the class
class Player(val diagnostics: Stream<Int>) {
    val doubled = diagnostics.map { d -> d * 2 }
}

players.switchMapNotNull { it.doubled }

Switching publishes the new inner stream's current value even when it equals the last one published. Add .distinct() if you only want changes.

fold

Accumulate values over time, like List.fold() but reactive:

val clicks = StreamSource("click")
val count = clicks.fold(0) { acc, _ -> acc + 1 }

// count is 1 after initial subscribe, increments on each "click"

foldNotNull

Like fold, but skips null values in the source stream:

val source = StreamSource<String?>(null)
val lengths = source.foldNotNull(0) { acc, next -> acc + next.length }

println(lengths())  // 0

source("hello")
println(lengths())  // 5

source(null)        // skipped
println(lengths())  // 5

source("world")
println(lengths())  // 10

history

Keep a sliding window of recent values:

val source = StreamSource(1)
val recent = source.history(3)

source(2)
source(3)
source(4)

println(recent())  // [2, 3, 4] — oldest values dropped

historyOfNonNull

Like history, but only keeps non-null values in the window:

val source = StreamSource<String?>("a")
val recent = source.historyOfNonNull(3)

source("b")
source(null)   // skipped
source("c")
source(null)   // skipped
source("d")

println(recent())  // ["b", "c", "d"] — nulls excluded