RxJS 7 operators tested with TestScheduler marble diagrams: working versions (tested 2026-10-15)

{
  "name": "rxjs-testscheduler-marbles",
  "private": true,
  "type": "module",
  "scripts": {
    "test": "node --import tsx --test test/*.test.ts"
  },
  "dependencies": {
    "rxjs": "7.8.2"
  },
  "devDependencies": {
    "@types/node": "26.6.5",
    "tsx": "4.23.15"
  }
}

import { Observable, of, timer } from "rxjs";
import { catchError, debounceTime, distinctUntilChanged, filter, map, retry, switchMap } from "rxjs/operators";

export type SearchFn = (q: string) => Observable<string[]>;

/** Classic typeahead: wait for a pause, skip repeats and short terms, cancel stale requests. */
export function typeahead(input$: Observable<string>, search: SearchFn, dueTime = 30): Observable<string[]> {
  return input$.pipe(
    map((q) => q.trim()),
    debounceTime(dueTime),
    filter((q) => q.length >= 2),
    distinctUntilChanged(),
    switchMap((q) => search(q).pipe(catchError(() => of(["error"])))),
  );
}

/** Retry a failing source with a fixed delay between attempts. */
export function withRetry<T>(source$: Observable<T>, count: number, delayMs: number): Observable<T> {
  return source$.pipe(retry({ count, delay: () => timer(delayMs) }));
}

import { test } from "node:test";
import assert from "node:assert/strict";
import { TestScheduler } from "rxjs/testing";
import { typeahead, withRetry } from "../src/search.ts";

const scheduler = () => new TestScheduler((actual, expected) => assert.deepEqual(actual, expected));

test("debounces, filters short terms and dedupes", () => {
  scheduler().run(({ hot, cold, expectObservable }) => {
    const input = hot("a 10ms b 40ms c 40ms d", { a: "r", b: "rx", c: "rx ", d: "rxjs" });
    const search = (q: string) => cold("-r|", { r: [q.toUpperCase()] });
    // Frames: a=0, b=11, c=52, d=93. "rx" settles at 41 and the search answers at 42;
    // "rx " trims to a duplicate; "rxjs" settles at 123 and answers at 124.
    expectObservable(typeahead(input, search)).toBe("42ms x 81ms y", { x: ["RX"], y: ["RXJS"] });
  });
});

test("switchMap cancels a slow in-flight search", () => {
  scheduler().run(({ hot, cold, expectObservable, expectSubscriptions }) => {
    const input = hot("a 40ms b", { a: "ab", b: "abc" });
    const slow = cold("50ms r|", { r: ["slow"] });
    const fast = cold("1ms r|", { r: ["fast"] });
    const search = (q: string) => (q === "ab" ? slow : fast);
    expectObservable(typeahead(input, search)).toBe("72ms x", { x: ["fast"] });
    // slow is subscribed when "ab" settles (30) and dropped when "abc" settles (71).
    expectSubscriptions(slow.subscriptions).toBe("30ms ^ 40ms !");
  });
});

test("errors become a fallback value without killing the stream", () => {
  scheduler().run(({ hot, cold, expectObservable }) => {
    const input = hot("a 40ms b", { a: "bad", b: "good" });
    const search = (q: string) => (q === "bad" ? cold("#") : cold("r|", { r: [q] }));
    expectObservable(typeahead(input, search)).toBe("30ms e 40ms g", { e: ["error"], g: ["good"] });
  });
});

test("retry resubscribes after a delay", () => {
  scheduler().run(({ cold, expectObservable, expectSubscriptions }) => {
    const source = cold("-#");
    expectObservable(withRetry(source, 2, 5)).toBe("13ms #");
    expectSubscriptions(source.subscriptions).toBe(["^!", "6ms ^!", "12ms ^!"]);
  });
});