Observable

An Observable is a lazy, push-based stream of values. Nothing runs until you subscribe; each subscription runs the handler you passed to the constructor, independently of any other subscription.

Creating an observable

The constructor takes a handler that receives an observer and may return a teardown function.

import { Observable } from '@remotex-labs/xobservable';

const source = new Observable<number>((observer) => {
    observer.next?.(1);
    observer.next?.(2);
    observer.complete?.();

    return () => console.log('cleaned up'); // optional teardown
});

The handler drives the stream by calling observer.next, observer.error, and observer.complete. Because it runs once per subscribe, each subscriber gets its own execution - this is what makes an observable cold.

Subscribing

subscribe accepts either an observer object or positional next, error, and complete callbacks. It returns an unsubscribe function.

const unsubscribe = source.subscribe({
    next: (value) => console.log(value),
    error: (err) => console.error(err),
    complete: () => console.log('done')
});

unsubscribe(); // stops delivery and runs the teardown

The unsubscribe function is idempotent - calling it more than once runs the teardown only once - so you never double-free a subscription.

If the handler throws synchronously, the error is routed to observer.error and a no-op unsubscribing is returned - see Error Handling.

Automatic cleanup with using

The unsubscribe function also implements Disposable, so you can bind it with a using declaration and let the subscription tear down automatically when the enclosing block exits. This removes the most common source of leaks: forgetting to call the unsubscribe function on a long-lived stream.

{
    using sub = source.subscribe((value) => console.log(value));

    // ...use the subscription
} // sub is disposed here - delivery stops and the teardown runs

using requires TypeScript 5.2+ and a runtime with Symbol.dispose (Node.js 22+). Calling the function manually still works exactly as before, so using is entirely optional.

pipe

pipe composes operators left to right, returning a new observable. With no operators it returns the same instance, so pipe() is always safe to call.

import { Observable, map, filter } from '@remotex-labs/xobservable';

const result = source.pipe(
    filter((x) => x > 0),
    map((x) => x * 2)
);

pipe is overloaded up to five operators with full type inference; beyond that the result type is inferred from the final operator. See Piping & Composition.

API

new Observable<T>(handler)

Parameter	Description
handler	Runs on each subscribe; receives the observer and may return a teardown function.

subscribe(observerOrNext?, error?, complete?)

Returns an UnsubscribeType function. Pass either a full observer object as the first argument, or a next callback followed by optional error and complete callbacks. The returned function is idempotent and implements Disposable, so it works with a using declaration.

pipe(...operators)

Returns the transformed observable, or this when called with no operators.

Cold vs. hot

An Observable is cold: the handler re-runs for every subscriber. When you need a single execution shared across many subscribers, use a Subject instead.