rxjs-fp

Functional-style reactive extensions. Operators are plain curried functions composed with pipe, nothing is patched onto any prototype, and each behavior gets its own operator instead of a config key.

See docs/DESIGN.md for the project charter and operator architecture principles, docs/pipe-usage.md for why composition is pipe(source, ...) instead of source.pipe(...), and docs/rxjs-fp-patterns.md for a catalog of the FP patterns used throughout the library.

import { pipe, of, map, filter, bufferCount } from 'rxjs-fp';

const subscription = pipe(
  of(1, 2, 3, 4, 5),
  map((n) => n * 2),
  filter((n) => n > 4),
  bufferCount(2)
).subscribe(console.log); // [6, 8], [10]

subscription.unsubscribe();

Why

Two things this library deliberately does differently.

No prototype patching. Operators are free functions. There is no source.pipe(...) method and no import-time mutation of a shared Observable prototype, so what a stream does is determined entirely by the functions you passed to pipe — not by which modules happened to be imported first. It also means the package is genuinely side-effect free, and bundlers drop what you don't use:

| Import | Minified | | ------------------- | -------- | | pipe, of, map | 1,453 B | | the entire library | 31,612 B |

One operator per behavior. Where other reactive libraries fold a family of operators into one function driven by an options bag, rxjs-fp splits them back apart. The signature tells you what you get.

| Instead of a config key | rxjs-fp | | ------------------------------------------------------ | ------------------------------------------------------------------- | | buffer({ delay, maxSize, startEvery, emitEmpty, … }) | buffer, bufferCount, bufferTime, bufferWhen, bufferToggle | | mergeMap(fn, { concurrent }) | mergeMapConcurrent(fn, n) | | mergeScan(fn, seed, concurrent) | mergeScanConcurrent(fn, seed, n) | | expand(fn, { concurrent }) | expandConcurrent(fn, n) | | takeWhile(p, { includeLast }) | takeWhileInclusive(p) | | retry({ count, delay }) | retry(count) / retryDelay(count, delay) | | timeout({ each, with }) | timeout(ms) / timeoutWith(ms, fallback) | | throttle(d, { leading, trailing }) | throttle(d) (leading) / audit(d) (trailing) | | shareReplay({ bufferSize, refCount }) | shareReplay(n) / shareReplayRefCount(n) | | timer(due, period) | timer(due) / interval(period) | | generate({ initialState, condition, … }) | generate(seed, condition, iterate, select?) | | first(predicate) | filter(predicate) then first() | | distinct(keySelector, flushes) | distinctReset(notifier, keySelector?) | | count(predicate) | filter(predicate) then count() |

Where a split would only produce an alias, none was added: auditTime already is trailing throttleTime, so there is no throttleTrailing.

Streams are cold

Every subscribe runs the producer again, with its own teardown. Two subscribers never share a producer, there is no ref counting, and no shared active subscription:

const source = pipe(
  interval(10),
  map((n) => n * 2),
  take(3)
);

source.subscribe((v) => console.log('a', v)); // producer #1
source.subscribe((v) => console.log('b', v)); // producer #2, independent

subscribe returns a Subscription; cancellation is unsubscribe(), not an AbortSignal. Sharing is something you opt into — with share, shareReplay, or a Subject — never the default.

A Subject is the one hot thing here: it is the producer, exists before anyone subscribes, and so can multicast.

API

Creating — of, from, defer, timer, interval, fromEvent, fromEventPattern, generate, throwError, animationFrames, EMPTY, NEVER

Combining sources — merge, concat, combineLatest, zip, race, forkJoin, onErrorResumeNext, partition

Multicasting — Subject, BehaviorSubject, ReplaySubject, AsyncSubject, connectable, multicast, refCount, connect, publish, publishBehavior, publishLast, publishReplay, share, shareReplay, shareReplayRefCount

Transforming — map, scan, mergeMap, mergeMapConcurrent, concatMap, switchMap, exhaustMap, mergeScan, mergeScanConcurrent, switchScan, expand, expandConcurrent, groupBy, pluck, pairwise, materialize, dematerialize

Filtering — filter, take, takeLast, takeWhile, takeWhileInclusive, takeUntil, skip, skipLast, skipWhile, skipUntil, distinct, distinctBounded, distinctReset, distinctUntilChanged, distinctUntilKeyChanged, first, last, single, elementAt, find, findIndex, ignoreElements

Reducing — reduce, count, max, min, every, isEmpty, sequenceEqual, defaultIfEmpty, throwIfEmpty

Buffering and windowing — buffer, bufferCount, bufferTime, bufferWhen, bufferToggle, window, windowCount, windowTime, windowWhen, windowToggle

Timing — delay, delayWhen, debounce, debounceTime, throttle, throttleTime, audit, auditTime, sample, sampleTime, timeout, timeoutWith, timestamp, timeInterval, observeOn, subscribeOn

Scheduling — Scheduler, queueScheduler, asapScheduler, asyncScheduler, animationFrameScheduler

Errors and effects — catchError, retry, retryDelay, retryWhen, repeat, repeatDelay, repeatWhen, tap, finalize

Piped combinators — startWith, withLatestFrom, mergeWith, concatWith, raceWith, zipWith, combineLatestWith, combineLatestAll, zipAll

Promises and async iteration — firstValueFrom, lastValueFrom, eachValueFrom, latestValueFrom, nextValueFrom

Core — Observable, Observer, Subscription, Subscriber, pipe, compose, operate, identity, noop, nextNotification, errorNotification, COMPLETE_NOTIFICATION, EmptyError, SequenceError, TimeoutError, ArgumentOutOfRangeError

from accepts Observables, promises, iterables, async iterables and array-likes, so anywhere an inner source is expected you can pass any of those:

pipe(
  of(1, 2),
  mergeMap((n) => [n, n * 10])
); // 1, 10, 2, 20

combineLatest, zip and forkJoin preserve tuple types:

combineLatest([of(1), of('a')]); // Observable<[number, string]>

Async iteration comes in four strategies, differing only in what they do when the loop falls behind — lossless, or bounded and lossy:

for await (const value of eachValueFrom(source)) { … }   // queues everything
for await (const value of latestValueFrom(source)) { … } // keeps only the newest
for await (const value of nextValueFrom(source)) { … }   // keeps only what arrives while waiting

Writing an operator

An operator is a function returning (source: Observable<T>) => Observable<R>. The operate helper carries the plumbing every operator shares — it subscribes upstream, registers that subscription as teardown before the producer runs so cancellation works mid-emission, and turns a throwing handler into a stream error:

import { Observable, operate } from 'rxjs-fp';
import type { OperatorFunction } from 'rxjs-fp';

export function double(): OperatorFunction<number, number> {
  return (source) =>
    new Observable<number>((subscriber) => {
      operate(source, subscriber, {
        next: (value) => subscriber.next(value * 2),
      });
    });
}

error and complete are optional and default to forwarding downstream. next is required — an operator should always be explicit about whether it forwards, transforms or drops.

Use compose to bundle operators into one reusable operator without a source:

const evensDoubled = compose(
  filter((n: number) => n % 2 === 0),
  map((n) => n * 2)
);

pipe(of(1, 2, 3, 4), evensDoubled);

Samples

The fp-style guides in docs/ describe a refactor recipe: write the technical pipeline first, then extract stable business intent into named, testable operators. docs/rxjs-fp-patterns.md catalogs the naming and composition patterns behind that recipe. Each file in samples/ walks that recipe for one domain — the raw pipeline, the extracted domain operators, the composed business-readable pipeline, and a runnable demo — with isolated tests for the rules in test/.

The policies that repeat across domains — debounce, distinct, switch-to-latest, bounded concurrency, timeout, retry, lifecycle, error recovery — live once in samples/shared-operators.ts as waitFor, onlyChanged, latest, atMost, failAfterSilence, retryAfter, until, and recoverFailure. Each domain binds them to its own rules and keeps only what is genuinely its own: onlyX predicate filters, mapXToY projections, and the per-item effect journeys.

| Sample | Scenario | Shows | | --------------------------------------------------------------------- | --------------------------- | ----------------------------------------------------------------------------------- | | orders-domain | invoicing submitted orders | the docs' running example; bound concurrency policy, per-invoice recovery | | search-autocomplete-domain | debounced typeahead search | named debounce policy, stale searches cancelled by latest | | upload-queue-domain | bounded-concurrency uploads | per-upload timeout and retry policies, running summary behind scan | | angular-form-domain | valueChanges autosave | saving/saved/autosave-failed states, the latest edit superseding a stale save | | websocket-reconnect-domain | resilient price ticker | heartbeat and reconnect policies, tick history that survives reconnects |

The top-level pipeline ends up reading as the scenario, with the generic operators one layer down:

pipe(orders$, invoiceSubmittedBillableOrders(saveInvoice));

Each sample runs as a script (any TypeScript runner that resolves the .js specifiers works, e.g. tsx), and its rules are tested without the full stream:

npx tsx samples/orders-domain.ts
npx vitest --run test/orders-domain.spec.ts

Differences worth knowing

• repeat(count) counts additional runs, matching retry, so repeat(2) runs three times in total. Upstream RxJS counts total subscriptions in repeat and additional ones in retry; this library is consistent rather than compatible.
• scan and reduce require a seed, so the output type is never in doubt.
• max and min complete without emitting on an empty source; follow with throwIfEmpty if that should be an error.
• first, last, single and count take no predicate — compose filter in front.
• Teardown runs in reverse registration order and never throws into the caller; a failing teardown is reported and the rest still run.
• expand drains its recursion with a FIFO loop, so deep synchronous projections cost constant stack.

Install

Not published to npm yet.

npm install github:hansschenker/rxjs-fp

Requires Node 22+. ESM only.

Development

npm test          # vitest
npm run lint      # eslint
npm run typecheck # tsc --noEmit
npm run build     # tsup: ESM + .d.ts

Run one file or one case:

npx vitest --run src/operators/map.spec.ts
npx vitest --run -t "unsubscribes the previous inner"

License

Apache-2.0 — see LICENSE.

Author

Hans Schenker — netxpert.ch