Observable Primitives — API Reference

@tagna/udiot v1.0.0-rc.1

import {
  Observable,
  Subject,
  BehaviorSubject,
  ReplaySubject,
  AsyncSubject
} from '@tagna/udiot';

Core Interfaces

interface Observer<T> {
  next: (value: T) => void;
  error?: (error: any) => void;
  complete?: () => void;
}

interface Subscription {
  unsubscribe: () => void;
  closed: boolean;
}

Observable<T>

A cold observable. The producer function runs once per subscriber, so each subscriber gets its own independent execution.

class Observable<T> {
  constructor(producer: (observer: Observer<T>) => (() => void) | void)
}

subscribe(observerOrNext, onError?, onComplete?)

subscribe(
  observerOrNext: Observer<T> | ((value: T) => void),
  onError?: (error: any) => void,
  onComplete?: () => void
): Subscription

Executes the producer. Returns a Subscription with an unsubscribe() method and a closed boolean. Calling unsubscribe() stops delivery and runs any teardown returned by the producer.

const obs$ = new Observable<number>(observer => {
  let i = 0;
  const id = setInterval(() => observer.next(i++), 1000);
  return () => clearInterval(id); // teardown
});

const sub = obs$.subscribe(n => console.log(n));
setTimeout(() => sub.unsubscribe(), 5000); // stops after ~5 values

Instance Operators

All operators return a new Observable and do not modify the source.

Static Creation Methods

Observable.of<T>(...values: T[]): Observable<T>
Observable.from<T>(array: T[] | ArrayLike<T>): Observable<T>
Observable.fromPromise<T>(promise: Promise<T>): Observable<T>
Observable.interval(ms: number): Observable<number>
Observable.timer(delay: number): Observable<number>
Observable.never<T>(): Observable<T>
Observable.empty<T>(): Observable<T>
Observable.throw<T>(error: any): Observable<T>

Subject<T>

A hot observable. Multicasts to all current subscribers. Extends Observable<T> so it can be passed anywhere an Observable is expected.

class Subject<T> extends Observable<T> {
  next(value: T): void
  error(err: any): void
  complete(): void
  subscribe(observerOrNext: Observer<T> | ((value: T) => void), ...): Subscription
  asObservable(): Observable<T>
  get observerCount(): number
  get closed(): boolean
}

A Subject is stopped (closed = true) after error() or complete() is called. Further calls to next() are silently ignored.

const events$ = new Subject<string>();

events$.subscribe(e => console.log('handler 1:', e));
events$.subscribe(e => console.log('handler 2:', e));

events$.next('click');
// handler 1: click
// handler 2: click

events$.complete();
events$.next('after complete'); // ignored

BehaviorSubject<T>

Extends Subject<T>. Replays the most recent value to each new subscriber synchronously at the moment of subscription.

class BehaviorSubject<T> extends Subject<T> {
  constructor(initialValue: T)
  get value(): T           // Throws if closed or errored
  getValue(): T | undefined // Safe — returns undefined if closed or errored
  next(value: T): void     // Updates current value then multicasts
  subscribe(...): Subscription  // Emits current value immediately to new subscriber
}

value (the getter) throws if the subject is closed or has errored. Use getValue() for a non-throwing alternative.

const count$ = new BehaviorSubject<number>(0);

count$.subscribe(n => console.log('A:', n)); // A: 0  (immediate)

count$.next(1); // A: 1

count$.subscribe(n => console.log('B:', n)); // B: 1  (immediate — gets latest)

count$.next(2); // A: 2, B: 2

ReplaySubject<T>

Extends Subject<T>. Stores up to bufferSize past values and replays them to each new subscriber.

class ReplaySubject<T> extends Subject<T> {
  constructor(bufferSize?: number) // Default: Infinity
  next(value: T): void
  subscribe(...): Subscription  // Replays buffered values first
  getBuffer(): T[]
  clearBuffer(): void
}

const log$ = new ReplaySubject<string>(3); // keep last 3

log$.next('a');
log$.next('b');
log$.next('c');
log$.next('d'); // 'a' is evicted

// Late subscriber gets 'b', 'c', 'd' immediately
log$.subscribe(s => console.log(s));

AsyncSubject<T>

Extends Subject<T>. Buffers all values but emits only the last one — and only when complete() is called.

class AsyncSubject<T> extends Subject<T> {
  next(value: T): void     // Stores but does not emit
  complete(): void         // Emits last stored value then completes
}

const result$ = new AsyncSubject<number>();

result$.subscribe(n => console.log('result:', n));

result$.next(1);
result$.next(2);
result$.next(3);
result$.complete(); // emits 3, then completes
// result: 3

Useful for wrapping a single async operation that you want to multicast.

Examples

BehaviorSubject Powering a Count Display

const count$ = new BehaviorSubject(0);

function renderCount(n: number): void {
  document.getElementById('counter')!.textContent = String(n);
}

count$.subscribe(renderCount); // renders 0 immediately

document.getElementById('inc')!.addEventListener('click', () => {
  count$.next(count$.value + 1);
});

Subject as Event Bus

const eventBus$ = new Subject<{ type: string; payload?: any }>();

// Producer
function emitEvent(type: string, payload?: any): void {
  eventBus$.next({ type, payload });
}

// Consumers
eventBus$
  .filter(e => e.type === 'USER_LOGIN')
  .subscribe(e => console.log('User logged in:', e.payload));

eventBus$
  .filter(e => e.type === 'PURCHASE')
  .subscribe(e => analytics.track('purchase', e.payload));

emitEvent('USER_LOGIN', { userId: 42 });
emitEvent('PURCHASE', { amount: 99.99 });

Observable from a Timer

const ticker$ = Observable.interval(1000); // emits 0, 1, 2, … every second

const sub = ticker$
  .take(5)
  .map(n => `tick ${n}`)
  .subscribe(msg => console.log(msg));

// Automatically unsubscribes after 5 values because take(5) calls complete()

Unsubscribe Cleanup Pattern in Class Lifecycle

class DataComponent {
  private subscriptions: (() => void)[] = [];

  mount(store: Store<AppState>): void {
    const sub1 = store
      .select(s => s.users)
      .subscribe(users => this.renderUsers(users));

    const sub2 = store
      .select(s => s.loading)
      .subscribe(loading => this.setSpinner(loading));

    this.subscriptions.push(
      () => sub1.unsubscribe(),
      () => sub2.unsubscribe()
    );
  }

  unmount(): void {
    this.subscriptions.forEach(unsub => unsub());
    this.subscriptions = [];
  }

  private renderUsers(users: User[]): void { /* ... */ }
  private setSpinner(loading: boolean): void { /* ... */ }
}