Subscriptions & Resilience

Batch cleanup of subscriptions and protection against failing handlers.

The Subscriptions collector

Forgetting to unsubscribe is the most common source of leaks in reactive code. The Subscriptions class collects all your subscriptions and cleans them up in a single call:

import io.peekandpoke.ultra.streams.Subscriptions

val subs = Subscriptions()

// Subscribe via the collector
subs.subscribe(nameStream) { println("Name: $it") }
subs.subscribe(ageStream) { println("Age: $it") }

// Or use the += operator
subs += counterStream.subscribeToStream { println("Count: $it") }

println(subs.size)  // 3

// Clean up everything at once
subs.unsubscribeAll()

This is especially useful in UI components:

class UserDashboard {
    private val subs = Subscriptions()

    fun mount(userStream: Stream<User>, notificationsStream: Stream<List<Notification>>) {
        subs.subscribe(userStream) { renderHeader(it) }
        subs.subscribe(notificationsStream) { renderNotifications(it) }
    }

    fun unmount() {
        subs.unsubscribeAll()
    }
}

unsubscribeAll() can be called multiple times safely.

Error resilience

A failing subscriber never breaks other subscribers. If one handler throws during a publish cycle, the remaining handlers still receive the value:

val source = StreamSource(1)

source.subscribeToStream { throw RuntimeException("oops") }
source.subscribeToStream { println("Got: $it") }

source(2)  // prints "Got: 2"

The exception is logged as [Streams] ERROR: subscriber threw ... so it shows up in your console during development, but it doesn't take down the stream.

Lazy subscription chains

Operator chains are fully lazy. No work happens until someone subscribes to the end of the chain:

val source = StreamSource(1)

// No subscriptions created yet — this chain is inert
val result = source
    .map { it * 2 }
    .filter(0) { it > 5 }
    .distinct()

// NOW the chain subscribes upstream
val unsub = result.subscribeToStream { println(it) }

// Unsubscribing tears down the entire chain
unsub()
// source has zero subscribers again

Each operator only subscribes to its upstream when it gains its first subscriber, and unsubscribes when its last subscriber leaves. Unused branches cost nothing.

Permanent subscriptions

Sometimes you want a stream to stay subscribed to its upstream forever — for example, a derived stream that must always be ready. Use permanent() to pin a subscription that never tears down:

val source = StreamSource(0)

// Always-active derived stream with a side effect
val logged = source
    .map { it * 2 }
    .permanent { println("Value changed: $it") }

// Or just pin the subscription without a handler
val pinned = source.map { it + 1 }.permanent()

Use this sparingly — permanent subscriptions bypass the lazy teardown that keeps unused branches free.