Operators Reference

This page summarizes commonly used functions for creating, transforming, filtering, and combining streams. Operators can be composed using the pipeFrom function.

Creation Operators

Functions that create streams

from

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

Creates a stream from a synchronous factory function.

Example:

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

fromPromise

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

Creates a stream from a Promise factory.

Example:

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

of

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

Creates a stream that emits the provided values in sequence.

Example:

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

interval

interval(period: number): Sub<number>

Creates a stream that emits sequential numbers at specified intervals.

Example:

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

Usage Patterns

Basic Composition

All operators are designed to work with the pipeFrom function:

TypeScript
import { pipeFrom, from, map, filter, take } from 'rx4u';

const result$ = pipeFrom(
  of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10),
  filter(x => x % 2 === 0),  // Keep even numbers
  map(x => x * x),           // Square them
  take(3)                    // Take first 3
); // Emits: 4, 16, 36

Error Handling Pattern

JavaScript
const resilientStream$ = pipeFrom(
  riskyOperation$,
  retry(3),
  catchError(error => {
    console.error('Operation failed:', error);
    return of('fallback value');
  })
);

Async Operations

JavaScript
const searchResults$ = pipeFrom(
  searchInput$,
  debounceTime(300),         // Wait for user to stop typing
  distinctUntilChanged(),    // Skip duplicate queries
  switchMap(query =>         // Switch to new search
    fromPromise(() => fetch(`/api/search?q=${query}`))
  ),
  catchError(err => of([))  // Return empty on error
);

Next Steps