@real-router/rx

Reactive Observable API for Real-Router. State streams, event streams, built-in operators, TC39 Observable and RxJS interop. Zero-cost opt-in — only bundled when imported.

Installation

npm install @real-router/rx

Peer dependency: @real-router/core

Quick Start

import { state$, map, filter, distinctUntilChanged } from "@real-router/rx";

state$(router)
  .pipe(
    map(({ route }) => route.params.categoryId),
    filter((id) => id !== undefined),
    distinctUntilChanged(),
  )
  .subscribe((categoryId) => {
    loadCategory(categoryId);
  });

Streams

| Factory | Returns | Description | |---------|---------|-------------| | state$(router, options?) | RxObservable<{ route, previousRoute }> | Router state changes | | events$(router) | RxObservable<RouterEvent> | All router events (start, stop, transition, error, cancel) | | observable(router) | RxObservable<SubscribeState> | TC39-style Observable wrapper for RxJS interop |

state$ accepts { signal: AbortSignal } for automatic unsubscription.

Operators

| Operator | Description | |----------|-------------| | map(project) | Transform emitted values | | filter(predicate) | Filter values by predicate | | debounceTime(ms) | Emit only the last value after a delay | | distinctUntilChanged(cmp?) | Skip consecutive duplicates | | takeUntil(notifier) | Complete when notifier emits |

All operators are composable via .pipe():

state$(router).pipe(
  filter(({ route }) => route.name.startsWith("admin")),
  debounceTime(100),
).subscribe(({ route }) => {
  analytics.trackPage(route.name);
});

Event Filtering

import { events$, filter } from "@real-router/rx";

events$(router)
  .pipe(filter((e) => e.type === "TRANSITION_ERROR"))
  .subscribe(({ error }) => {
    errorTracker.capture(error);
  });

RxJS Interop

observable() returns a TC39-style Observable — pass it to RxJS from():

import { from } from "rxjs";
import { debounceTime } from "rxjs/operators";
import { observable } from "@real-router/rx";

from(observable(router))
  .pipe(debounceTime(100))
  .subscribe(({ route }) => {
    console.log("Route:", route.name);
  });

The interop member is declared under the "@@observable" string and aliased onto Symbol.observable when the host defines one. A consumer picks the host's symbol if there is one and the string otherwise, which is what RxJS from() does. Symbol.observable is not a well-known symbol — a host has one only if something polyfilled it, so on a bare host only the string spelling exists.

Divergence from TC39 / RxJS — error is non-terminal. Unlike the TC39 proposal and RxJS (where error is a terminal event that triggers cleanup), this library keeps the subscription open after error(): values keep flowing, multiple errors are each forwarded, and closed stays false. Only complete() and unsubscribe() are terminal. This is intentional — state$/events$ are infinite router streams, so a single throwing subscriber must not permanently kill the stream.

⚠ An RxJS chain over the same stream disagrees: from(observable(router)) does end on error, because that is RxJS's own Subscriber contract and not something this library can change. The stream behind it keeps going, so an ended chain is no evidence that it stopped — resubscribe, or use RxJS catchError / retry, to keep receiving.

Documentation

Full documentation: Wiki — rx

• API Reference
• Operators
• Async Iteration
• Migration Guide

Related Packages

| Package | Description | |---------|-------------| | @real-router/core | Core router (required peer dependency) | | @real-router/sources | useSyncExternalStore-based alternative | | @real-router/react | React integration |

Contributing

See contributing guidelines for development setup and PR process.

License

MIT © Oleg Ivanov