API Reference

Complete documentation of all rx4u operators, creation functions, and utilities.

from

creation

from<T>(create: () => T): Sub<T>

Creates a stream from a synchronous factory function.

Parameters:

create: () => T - Factory function that returns the value to emit

Example:

JavaScript

const lazy$ = from(() => Math.random());
const greeting$ = from(() => 'hello');

fromPromise

creation

fromPromise<T>(create: () => Promise<T>): Sub<T>

Creates a stream from a Promise factory.

Parameters:

create: () => Promise<T> - Factory function that returns a Promise

Example:

JavaScript

const request$ = fromPromise(() => fetch('/api/data'));
request$(console.log);

of

creation

of<T>(...values: T[]): Sub<T>

Creates a stream that emits the provided values in sequence.

Parameters:

values: T[] - Values to emit

Example:

JavaScript

const stream$ = of(1, 2, 3);
stream$(console.log); // Emits: 1, 2, 3

interval

creation

interval(period: number): Sub<number>

Creates a stream that emits sequential numbers at specified intervals.

Parameters:

period: number - Time interval in milliseconds

Example:

JavaScript

const timer$ = interval(1000);
timer$(console.log); // Emits: 0, 1, 2, ... every second

fromEvent

creation

fromEvent<T extends Event = Event>(target: EventTargetLike, type: string, options?): Sub<T>

Creates a stream from events on a DOM-like event target.

Parameters:

target: EventTargetLike - Object that provides addEventListener and removeEventListener
type: string - Event type to listen for
options: AddEventListenerOptions | EventListenerOptions | boolean - Optional event listener options

Example:

JavaScript

const clicks$ = fromEvent<MouseEvent>(button, 'click');
clicks$(event => console.log(event.clientX));

createState

creation

createState<T>(initialValue: T): [Sub<T>, StateSetter<T>, () => T, () => void]

Creates reactive state with an initial value.

Parameters:

initialValue: T - Initial state value

Example:

JavaScript

const [count$, setCount, getCount, destroy] = createState(0);
count$(console.log);
setCount(value => value + 1);

createLazyState

creation

createLazyState<T>(): [Sub<T>, StateSetter<T>, () => T | undefined, () => void]

Creates reactive state that does not emit until its first update.

Example:

JavaScript

const [name$, setName] = createLazyState<string>();
name$(console.log);
setName('Ada');

createSubject

creation

createSubject<T>(): [Sub<T>, (value: T) => void]

Creates a multicast stream and a function that broadcasts values.

Example:

JavaScript

const [messages$, nextMessage] = createSubject<string>();
messages$(console.log);
nextMessage('hello');

pipeFrom

core

pipeFrom<T>(source: Sub<T>, ...operators: Operator[]): Sub<unknown>

Applies operators to a source stream from left to right.

Parameters:

source: Sub<T> - Source stream
operators: Operator[] - Operators to apply in order

Example:

JavaScript

const doubled$ = pipeFrom(
  of(1, 2, 3),
  map(value => value * 2)
);

firstValueFrom

core

firstValueFrom<T>(source: Sub<T>): Promise<T>

Resolves with the first value, or rejects if the stream errors or completes empty.

Parameters:

source: Sub<T> - Source stream

Example:

JavaScript

const first = await firstValueFrom(of(1, 2, 3));

map

transformation

map<T, R>(mapFn: (value: T) => R): Operator<T, R>

Transforms each emitted value using a projection function.

Parameters:

mapFn: (value: T) => R - Function to transform each value

Example:

JavaScript

const doubled$ = pipeFrom(
  of(1, 2, 3),
  map(x => x * 2)
); // Emits: 2, 4, 6

switchMap

transformation

switchMap<T, R>(project: (value: T, index: number) => Sub<R> | Promise<R>): Operator<T, R>

Projects each value to a new stream and switches to the latest projection.

Parameters:

project: (value: T, index: number) => Sub<R> | Promise<R> - Function that projects each value to a stream or Promise

Example:

JavaScript

const searchResults$ = pipeFrom(
  searchQuery$,
  debounceTime(300),
  switchMap(query => fromPromise(() => fetch(`/search?q=${query}`)))
);

mergeMap

transformation

mergeMap<T, R>(project: (value: T, index: number) => Sub<R>, concurrent?: number): Operator<T, R>

Projects each value to a stream and merges all concurrent streams.

Parameters:

project: (value: T, index: number) => Sub<R> - Function that projects each value to a stream
concurrent: number - Optional maximum number of active inner streams

Example:

JavaScript

const allRequests$ = pipeFrom(
  urls$,
  mergeMap(url => fromPromise(() => fetch(url)))
);

concatMap

transformation

concatMap<T, R>(project: (value: T, index: number) => Sub<R>): Operator<T, R>

Projects each value to a stream and concatenates them sequentially.

Parameters:

project: (value: T, index: number) => Sub<R> - Function that projects each value to a stream

Example:

JavaScript

const sequential$ = pipeFrom(
  of(1, 2, 3),
  concatMap(x => of(x, x * 2))
); // Emits: 1, 2, 2, 4, 3, 6

filter

filtering

filter<T>(predicate: (value: T) => boolean): Operator<T, T>

Emits only values that satisfy the provided predicate function.

Parameters:

predicate: (value: T) => boolean - Function to test each value

Example:

JavaScript

const evens$ = pipeFrom(
  of(1, 2, 3, 4, 5),
  filter(x => x % 2 === 0)
); // Emits: 2, 4

take

filtering

take<T>(count: number): Operator<T, T>

Emits only the first N values from the source.

Parameters:

count: number - Number of values to take

Example:

JavaScript

const first3$ = pipeFrom(
  interval(1000),
  take(3)
); // Emits: 0, 1, 2, then completes

takeWhile

filtering

takeWhile<T>(predicate: (value: T, index: number) => boolean, inclusive?: boolean): Operator<T, T>

Emits values while the predicate passes, then completes.

Parameters:

predicate: (value: T, index: number) => boolean - Function that tests each value
inclusive: boolean - Whether to emit the first value that fails the predicate

Example:

JavaScript

const values$ = pipeFrom(
  of(1, 2, 3, 4),
  takeWhile(value => value < 3, true)
); // Emits: 1, 2, 3

takeUntil

filtering

takeUntil<T>(notifier: Sub<unknown>): Operator<T, T>

Emits source values until the notifier emits.

Parameters:

notifier: Sub<unknown> - Stream whose first value stops the source

Example:

JavaScript

const values$ = pipeFrom(
  interval(1000),
  takeUntil(stop)
);

skip

filtering

skip<T>(count: number): Operator<T, T>

Skips the first N values from the source.

Parameters:

count: number - Number of values to skip

Example:

JavaScript

const skipFirst2$ = pipeFrom(
  of(1, 2, 3, 4, 5),
  skip(2)
); // Emits: 3, 4, 5

distinctUntilChanged

filtering

distinctUntilChanged<T, K = T>(compareFn?: (a: K, b: K) => boolean, keySelector?: (value: T) => K): Operator<T, T>

Only emits when the current value differs from the previous.

Parameters:

compareFn: (a: K, b: K) => boolean - Optional comparison function
keySelector: (value: T) => K - Optional function that selects the compared key

Example:

JavaScript

const unique$ = pipeFrom(
  of(1, 1, 2, 2, 3),
  distinctUntilChanged()
); // Emits: 1, 2, 3

merge

combination

merge<T>(...sources: Sub<T>[]): Sub<T>

Merges multiple streams into a single stream.

Parameters:

sources: Sub<T>[] - Streams to merge

Example:

JavaScript

const combined$ = merge(
  of(1, 2, 3),
  of(4, 5, 6)
); // Emits: 1, 2, 3, 4, 5, 6

combineLatest

combination

combineLatest<T extends readonly unknown[]>(...sources: {[K in keyof T]: Sub<T[K]>}): Sub<T>

Combines latest values from multiple streams whenever any stream emits.

Parameters:

sources: Sub<T>[] - Streams to combine

Example:

JavaScript

const latest$ = combineLatest(
  of(1, 2),
  of('a', 'b')
); // Emits: [2, 'a'], [2, 'b']

zip

combination

zip<T extends readonly unknown[]>(...sources: {[K in keyof T]: Sub<T[K]>}): Sub<T>

Combines values from multiple streams pairwise.

Parameters:

sources: Sub<T>[] - Streams to zip

Example:

JavaScript

const paired$ = zip(
  of(1, 2),
  of('a', 'b')
); // Emits: [1, 'a'], [2, 'b']

withLatestFrom

combination

withLatestFrom<T>(...others: Sub<unknown>[]): Operator<T, [T, ...unknown[]]>

Combines each source value with the latest values from other streams.

Parameters:

others: Sub<unknown>[] - Streams that provide the latest accompanying values

Example:

JavaScript

const values$ = pipeFrom(
  clicks$,
  withLatestFrom(state)
);

tap

utility

tap<T>(tapFn: (value: T) => void): Operator<T, T>

Runs a side effect for each value and forwards the value unchanged.

Parameters:

tapFn: (value: T) => void - Function called for each value

Example:

JavaScript

const logged$ = pipeFrom(
  of(1, 2, 3),
  tap(x => console.log('Value:', x))
);

delay

utility

delay<T = void>(ms: number): Operator<T, T>

Delays every source value by the specified duration.

Parameters:

ms: number - Delay in milliseconds

Example:

JavaScript

const delayed$ = pipeFrom(of(1, 2, 3), delay(500));

debounceTime

utility

debounceTime<T>(dueTime: number): Operator<T, T>

Emits a value only after a specified silence period.

Parameters:

dueTime: number - Time in milliseconds to wait

Example:

JavaScript

const debounced$ = pipeFrom(
  searchInput$,
  debounceTime(300)
); // Wait 300ms after last input

throttleTime

utility

throttleTime<T>(duration: number, options?: {leading?: boolean; trailing?: boolean}): Operator<T, T>

Emits at most one value per time period.

Parameters:

duration: number - Time period in milliseconds
options: {leading?: boolean; trailing?: boolean} - Optional leading and trailing emission behavior

Example:

JavaScript

const throttled$ = pipeFrom(
  clicks$,
  throttleTime(1000)
); // Max 1 click per second

timeout

utility

timeout<T>(ms: number, errorFactory?: () => Error): Operator<T, T>

Errors if the source does not emit within the specified duration.

Parameters:

ms: number - Timeout duration in milliseconds
errorFactory: () => Error - Optional custom error factory

Example:

JavaScript

const guarded$ = pipeFrom(request$, timeout(5000));

startWith

utility

startWith<T>(...values: T[]): Operator<T, T>

Emits the provided values before subscribing to the source.

Parameters:

values: T[] - Values to prepend

Example:

JavaScript

const values$ = pipeFrom(of(2, 3), startWith(1));

endWith

utility

endWith<T>(...values: T[]): Operator<T, T>

Emits the provided values after the source completes.

Parameters:

values: T[] - Values to append

Example:

JavaScript

const values$ = pipeFrom(of(1, 2), endWith(3));

utility

share<T>(): Operator<T, T>

Shares a single subscription among multiple subscribers.

Example:

JavaScript

const shared$ = pipeFrom(
  expensiveOperation$,
  share()
); // Multiple subscribers share the same execution

catchError

error-handling

catchError<T, R = T>(selector: (error: unknown, caught: Sub<T>) => Sub<R>): Operator<T, T | R>

Catches errors and continues with a fallback stream.

Parameters:

selector: (error: unknown, caught: Sub<T>) => Sub<R> - Function to handle errors

Example:

JavaScript

const resilient$ = pipeFrom(
  riskyOperation$,
  catchError(err => of('fallback'))
);

retry

error-handling

retry<T>(count?: number): Operator<T, T>

Retries the source stream on error.

Parameters:

count: number - Number of retry attempts

Example:

JavaScript

const retried$ = pipeFrom(
  unreliableRequest$,
  retry(3)
); // Retry up to 3 times

retryWhen

error-handling

retryWhen<T>(notifier: (errors: Sub<unknown>) => Sub<unknown>): Operator<T, T>

Retries based on custom retry logic.

Parameters:

notifier: (errors: Sub<unknown>) => Sub<unknown> - Function that controls retry timing

Example:

JavaScript

const smartRetry$ = pipeFrom(
  source$,
  retryWhen(errors => pipeFrom(
    errors,
    debounceTime(1000),
    take(3)
  ))
);

scan

mathematical

scan<T, R = T>(accumulator: (acc: R, value: T, index: number) => R, seed?: R): Operator<T, R>

Applies an accumulator function and emits each intermediate result.

Parameters:

accumulator: (acc: R, value: T, index: number) => R - Accumulator function
seed: R - Optional initial accumulator value

Example:

JavaScript

const running$ = pipeFrom(
  of(1, 2, 3, 4),
  scan((acc, x) => acc + x, 0)
); // Emits: 1, 3, 6, 10

reduce

mathematical

reduce<T, R = T>(accumulator: (acc: R, value: T, index: number) => R, seed?: R): Operator<T, R>

Applies an accumulator function and emits the final result when the source completes.

Parameters:

accumulator: (acc: R, value: T, index: number) => R - Accumulator function
seed: R - Optional initial accumulator value

Example:

JavaScript

const sum$ = pipeFrom(
  of(1, 2, 3, 4),
  reduce((acc, x) => acc + x, 0)
); // Emits: 10