Frontend RxJS — #11

Writing Custom RxJS Operators and Staying Readable: pipe Composition, Lint Rules and Debugging

Mert Özen Nov 14, 2026 21 min 14 views

A custom operator is a function that takes an Observable and returns one; combining existing operators with pipe() is usually enough, and an operator written from scratch with new Observable must return its teardown. We turn Bookstore's search, retry-with-backoff and loading-state blocks into settledTerm, retryWithBackoff and loadingState, write debugTap from scratch, and catch mistakes such as nested subscribe automatically with eslint-plugin-rxjs-x.

Version note: The examples in this post were run with RxJS 7.8.2, TypeScript 6.0, Node.js 24, Angular 22.2, ESLint 10.15, typescript-eslint 8.72 and eslint-plugin-rxjs-x 1.0.6 (the Angular project uses TypeScript 6.0, the range Angular 22.2 supports); the Angular examples ran in headless Chromium driven by Playwright 1.65 against the mock Bookstore API of the snapshot.

Predict First

In Bookstore, the online bookstore used as the shared example across this site's series, a book's stock is polled every second and the screen updates only when the stock changes. Someone on the team wrote the "do not pass the same value twice" part as their own operator, and the page shows only the first two changes, then stops:

import { map, Observable, take, tap, timer, type MonoTypeOperatorFunction } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";

// a hand-written operator: pass a stock value on only when it differs from the previous one
function stockChanges(): MonoTypeOperatorFunction<number> {
  return (source) =>
    new Observable<number>((subscriber) => {
      let previous: number | undefined;
      source.subscribe({
        next: (stock) => {
          if (stock !== previous) subscriber.next(stock);
          previous = stock;
        },
        error: (err: unknown) => subscriber.error(err),
        complete: () => subscriber.complete(),
      });
    });
}

// Dune's stock as the API reports it on each poll
const answers = [12, 12, 11, 11, 11, 10, 10, 9];

runInVirtualTime("first two stock changes of Dune", () => {
  timer(0, 1000)
    .pipe(
      tap((i) => log(`-> GET /books/9780441172719/stock (poll ${i + 1})`)),
      map((i) => answers[i % answers.length]),
      stockChanges(),
      take(2),
    )
    .subscribe({
      next: (stock) => log(`screen: ${stock} in stock`),
      complete: () => log("done"),
    });
}, 5000);

How many more requests go out after the screen says "done"? What happens to the timer at the start of the chain when take(2) completes?

Throughout the series we wrote some pipes several times: the debounced search block, the retry with backoff, the loading state around a request. In this post we turn them into named, typed, testable operators: first from existing operators with pipe(), then, when needed, from scratch with new Observable. After that we look at the rules that keep long pipes readable, and at handing some of those rules to ESLint.

An Operator Is Just a Function

Remember the definition from the marble diagram post: a pipeable operator is a function that takes an Observable and returns a new one. RxJS types it as OperatorFunction<T, R>; when input and output types are the same, the shorthand is MonoTypeOperatorFunction<T>. Writing your own operator means writing a function that returns a value of that type. There are two ways: combine existing operators with the pipe() function (the RxJS docs recommend this first, and it covers most cases), or, for behaviour that cannot be expressed with existing operators, write it from scratch with (source) => new Observable(subscriber => …).

stockChanges in the prediction was written the second way and forgot something. Running it:

--- first two stock changes of Dune
    0 ms  -> GET /books/9780441172719/stock (poll 1)
    0 ms  screen: 12 in stock
 1000 ms  -> GET /books/9780441172719/stock (poll 2)
 2000 ms  -> GET /books/9780441172719/stock (poll 3)
 2000 ms  screen: 11 in stock
 2000 ms  done
 3000 ms  -> GET /books/9780441172719/stock (poll 4)
 4000 ms  -> GET /books/9780441172719/stock (poll 5)
 5000 ms  -> GET /books/9780441172719/stock (poll 6)

take(2) completed on the second change and the screen said "done", but the requests keep going. We stopped the virtual clock at 5 seconds; in a real application the polling would continue until the page is closed, or until the application is closed if it lives in a root service. The cause is that the function given to new Observable returns nothing. When take(2) completes, it unsubscribes from its source, the Observable returned by stockChanges. That Observable has no teardown, and since the Subscription of the inner source.subscribe(…) is not connected to anything, the unsubscribe never reaches the timer.

The fix is one word: return the inner Subscription. The function given to new Observable may return a teardown function or a Subscription directly:

// a hand-written operator: pass a stock value on only when it differs from the previous one
function stockChanges(): MonoTypeOperatorFunction<number> {
  return (source) =>
    new Observable<number>((subscriber) => {
      let previous: number | undefined;
      // returning the Subscription makes it the teardown: unsubscribe now reaches the source
      return source.subscribe({
        next: (stock) => {
          if (stock !== previous) subscriber.next(stock);
          previous = stock;
        },
        error: (err: unknown) => subscriber.error(err),
        complete: () => subscriber.complete(),
      });
    });
}

--- first two stock changes of Dune, with teardown
    0 ms  -> GET /books/9780441172719/stock (poll 1)
    0 ms  screen: 12 in stock
 1000 ms  -> GET /books/9780441172719/stock (poll 2)
 2000 ms  -> GET /books/9780441172719/stock (poll 3)
 2000 ms  screen: 11 in stock
 2000 ms  done

Nothing after the third request. The RxJS docs list three requirements for an operator written from scratch: handle all three of next, error and complete when subscribing to the source, write a function that does the cleanup, and return that function from the function given to new Observable. This example missed the third. And by the way, what stockChanges does is exactly distinctUntilChanged(); checking whether an existing operator already does the job is the cheapest check before writing one from scratch.

A New Operator from Existing Ones with pipe()

In the search box we wrote three steps side by side every time: debounceTime, trim and distinctUntilChanged. The standalone pipe() function from rxjs, unlike an Observable's .pipe() method, takes no source; it combines the operators you give it and returns a single OperatorFunction:

// 1. a search term the user has stopped typing: wait, trim, skip repeats
export function settledTerm(dueTimeMs = 300): MonoTypeOperatorFunction<string> {
  return pipe(
    debounceTime(dueTimeMs),
    map((term) => term.trim()),
    distinctUntilChanged(),
  );
}

Combined with the next step, the search block we wrote throughout the series becomes one named operator. We put debugTap, which we write from scratch shortly, on the inner request, so you can see when switchMap subscribes to each request and when it lets go:

import { concat, map, NEVER, pipe, switchMap, timer, type Observable, type OperatorFunction } from "rxjs";
import { books } from "../../data/books.ts";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { debugTap, settledTerm } from "./operators.ts";

// a fake GET /books?q=…: answers after 500 ms
function findBooks(term: string): Observable<string[]> {
  return timer(500).pipe(
    map(() => books.filter((b) => b.title.toLowerCase().includes(term)).map((b) => b.title)),
  );
}

// the whole search block of the earlier posts, now one named operator
function searchBooks(): OperatorFunction<string, string[]> {
  return pipe(
    settledTerm(300),
    switchMap((term) => findBooks(term).pipe(debugTap(`GET /books?q=${term}`, log))),
  );
}

runInVirtualTime("search box", () => {
  // what the user types; NEVER keeps the input open like a real text box
  const input$ = concat(
    emitAt([0, "d"], [100, "du"], [200, "dun"], [300, "dune"], [1200, "dune "], [1600, "em"], [2000, "emma"]),
    NEVER,
  );
  input$.pipe(searchBooks()).subscribe((titles) => log(`screen: ${titles.join(", ")}`));
});

--- search box
  600 ms  [GET /books?q=dune] subscribe
 1100 ms  [GET /books?q=dune] next ["Dune"]
 1100 ms  screen: Dune
 1100 ms  [GET /books?q=dune] complete
 1900 ms  [GET /books?q=em] subscribe
 2300 ms  [GET /books?q=em] unsubscribe
 2300 ms  [GET /books?q=emma] subscribe
 2800 ms  [GET /books?q=emma] next ["Emma"]
 2800 ms  screen: Emma
 2800 ms  [GET /books?q=emma] complete

"dune" was searched 300 ms after the last key. The space added at 1200 ms gave the same term after trim, and distinctUntilChanged stopped it. "emma" settled while the "em" request was running, and switchMap dropped the old request and started the new one; the unsubscribe line in the log is that moment. The settled term now has a name, and both search components in bookstore-web use it: the three-line block in search-rxjs.ts and search-interop.ts became settledTerm(300). We reran the search page with the Playwright test from the previous post; all three boxes still sent a single request for "dune".

retryWithBackoff: An Operator with Options

The retry with backoff from the error handling post lived in two places: the retryDelay helper in the examples project and a copy inside the interceptor in bookstore-web. Both were a function passed to retry({ count, delay }); as an operator it gets a name and clear options:

// 2. retry with exponential backoff and jitter, only for errors that may go away
export interface BackoffOptions {
  count?: number;
  baseMs?: number;
  maxMs?: number;
  isRetryable?: (err: unknown) => boolean;
  random?: () => number;
  onRetry?: (err: unknown, retryCount: number, delayMs: number) => void;
}

export function retryWithBackoff<T>(options: BackoffOptions = {}): MonoTypeOperatorFunction<T> {
  const { count = 3, baseMs = 500, maxMs = 8000, isRetryable = () => true, random = Math.random, onRetry } = options;
  return retry({
    count,
    delay: (err: unknown, retryCount) => {
      if (!isRetryable(err)) return throwError(() => err);
      const delayMs = backoffDelay(retryCount, random, baseMs, maxMs);
      onRetry?.(err, retryCount, delayMs);
      return timer(delayMs);
    },
  });
}

The defaults are deliberately conservative: three attempts, a 500 ms base, at most 8 seconds. isRetryable retries every error by default, because the operator cannot know which errors are temporary; for HTTP the caller supplies it. The random option is there to pass a fixed number generator in tests, onRetry for logging. We ran the scenario from the error handling post with the same seed:

import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { isRetryable, seededRandom } from "../09/backoff.ts";
import { flakyHttp } from "../09/fake-http.ts";
import { retryWithBackoff } from "./operators.ts";

runInVirtualTime("503, 503, then 200", () => {
  flakyHttp("GET /books/9780441172719/stock", 200, [503, 503, 200], () => 12)
    .pipe(
      retryWithBackoff({
        count: 4,
        isRetryable,
        random: seededRandom(42),
        onRetry: (err, retryCount, delayMs) => log(`retry ${retryCount} in ${delayMs} ms (${String(err)})`),
      }),
    )
    .subscribe({
      next: (stock) => log(`stock: ${stock}`),
      error: (err: Error) => log(`stock failed: ${err.message}`),
    });
});

runInVirtualTime("404 is not retried", () => {
  flakyHttp("GET /books/9780000000000", 200, [404], () => "book")
    .pipe(retryWithBackoff({ count: 4, isRetryable }))
    .subscribe({
      next: (book) => log(`book: ${book}`),
      error: (err: Error) => log(`book failed: ${err.message}`),
    });
});

--- 503, 503, then 200
    0 ms  -> GET /books/9780441172719/stock
  200 ms  <- GET /books/9780441172719/stock 503
  200 ms  retry 1 in 400 ms (HttpError: HTTP 503)
  600 ms  -> GET /books/9780441172719/stock
  800 ms  <- GET /books/9780441172719/stock 503
  800 ms  retry 2 in 724 ms (HttpError: HTTP 503)
 1524 ms  -> GET /books/9780441172719/stock
 1724 ms  <- GET /books/9780441172719/stock 200
 1724 ms  stock: 12
--- 404 is not retried
    0 ms  -> GET /books/9780000000000
  200 ms  <- GET /books/9780000000000 404
  200 ms  book failed: HTTP 404

The waits (400 and 724 ms) are exactly the ones from the error handling post; the operator gave the behaviour a name without changing it. The 404 was not retried. The interceptor now uses the same operator and makes only the HTTP-specific decisions:

export const errorInterceptor: HttpInterceptorFn = (req, next) => {
  const serverStatus = inject(ServerStatus);
  // only GET is safe to repeat; a POST could create a second order
  const retries = req.method === 'GET' ? 3 : 0;

  return next(req).pipe(
    retryWithBackoff({
      count: retries,
      isRetryable,
      onRetry: (err, retryCount) => {
        const status = err instanceof HttpErrorResponse ? err.status : err;
        console.warn(`[http] ${req.method} ${req.url} failed (${status}), retry ${retryCount}`);
      },
    }),

loadingState: Producing States for the Template

A request has three states around it: waiting, arrived, failed. Instead of keeping them in separate signals in every component, we write an operator that turns the stream itself into these states. The return type is a discriminated union; when TypeScript sees status, it narrows access to data or error:

// 3. turn a request into states a template can show: loading, then success or error
export type LoadingState<T> =
  | { status: "loading" }
  | { status: "success"; data: T }
  | { status: "error"; error: unknown };

export function loadingState<T>(): OperatorFunction<T, LoadingState<T>> {
  return (source) =>
    source.pipe(
      map((data): LoadingState<T> => ({ status: "success", data })),
      startWith<LoadingState<T>>({ status: "loading" }),
      catchError((error: unknown) => of<LoadingState<T>>({ status: "error", error })),
    );
}

Order matters. Because startWith comes after map, loading is the first value; because catchError comes last, the error also becomes a state and the stream ends with complete, not error. Order matters again when you combine this operator with retryWithBackoff:

import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { isRetryable, seededRandom } from "../09/backoff.ts";
import { flakyHttp } from "../09/fake-http.ts";
import { loadingState, retryWithBackoff, type LoadingState } from "./operators.ts";

const show = (state: LoadingState<string[]>) =>
  log(
    state.status === "success" ? `success: ${state.data.join(", ")}`
    : state.status === "error" ? `error: ${String(state.error)}`
    : "loading",
  );

runInVirtualTime("categories: one 503, then 200", () => {
  flakyHttp("GET /categories", 200, [503, 200], () => ["classics", "fantasy"])
    .pipe(retryWithBackoff({ count: 2, isRetryable, random: seededRandom(42) }), loadingState())
    .subscribe(show);
});

runInVirtualTime("categories: 503 every time", () => {
  flakyHttp("GET /categories", 200, [503], () => ["classics"])
    .pipe(retryWithBackoff({ count: 2, isRetryable, random: seededRandom(42) }), loadingState())
    .subscribe(show);
});

runInVirtualTime("wrong order: loadingState before retryWithBackoff", () => {
  flakyHttp("GET /categories", 200, [503, 200], () => ["classics", "fantasy"])
    .pipe(loadingState(), retryWithBackoff({ count: 2, isRetryable, random: seededRandom(42) }))
    .subscribe(show);
});

--- categories: one 503, then 200
    0 ms  loading
    0 ms  -> GET /categories
  200 ms  <- GET /categories 503
  600 ms  -> GET /categories
  800 ms  <- GET /categories 200
  800 ms  success: classics, fantasy
--- categories: 503 every time
    0 ms  loading
    0 ms  -> GET /categories
  200 ms  <- GET /categories 503
  600 ms  -> GET /categories
  800 ms  <- GET /categories 503
 1524 ms  -> GET /categories
 1724 ms  <- GET /categories 503
 1724 ms  error: HttpError: HTTP 503
--- wrong order: loadingState before retryWithBackoff
    0 ms  loading
    0 ms  -> GET /categories
  200 ms  <- GET /categories 503
  200 ms  error: HttpError: HTTP 503

In the right order the screen shows a single loading, the retries happen in the background, and the result is either data or a single error. In the wrong order loadingState turned the error into a value on the first attempt; no error ever reached retryWithBackoff, so nothing was retried. The rule is simple: an operator that turns an error into a value must come after everything that needs to see that error.

The catalog page now shows these states instead of falling back to an empty list. Because startWith gives the first value synchronously, we can use toSignal with requireSync: true; there is no undefined in the type. The Angular template narrows the union inside @switch; reading state.data inside @case ('success') compiles, in any other case it would not:

    @let state = books();
    @switch (state.status) {
      @case ('loading') {
        <p class="books-state">Loading books...</p>
      }
      @case ('error') {
        <p class="books-state">{{ errorMessage(state.error) }}</p>
      }
      @case ('success') {
        <ul>
          @for (book of state.data; track book.isbn) {
            <li>
              <a [routerLink]="['/books', book.isbn]">{{ book.title }}</a> - {{ book.price }} TRY
              <button type="button" (click)="cart.add(book)" [disabled]="book.stock === 0">
                Add to cart
              </button>
            </li>
          }
        </ul>
      }
    }
  `,
})
export class CatalogPage {
  protected readonly cart = inject(CartStore);
  // the interceptor has already retried; the page shows loading, the list or a message
  protected readonly books = toSignal(
    inject(HttpClient).get<Book[]>(`${apiUrl}/books`).pipe(loadingState()),
    { requireSync: true },
  );
  protected readonly errorMessage = errorMessage;
  protected readonly categories$ = inject(CategoryService).categories$;
}

We started the API first with FAIL=/api/books:2 (the first two requests answer 503), then so that every request answers 503, and opened the page with Playwright:

 0.1 s screen: Loading books...
 0.3 s GET /api/books 503
 0.3 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 0.3 s [console.warning] [http] GET http://localhost:5180/api/books failed (503), retry 1
 0.9 s GET /api/books 503
 0.9 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 0.9 s [console.warning] [http] GET http://localhost:5180/api/books failed (503), retry 2
 2.0 s GET /api/books 200
 2.0 s screen: 6 books: Dune, Emma, The Hobbit…

 0.1 s screen: Loading books...
 0.3 s GET /api/books 503
 0.3 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 0.3 s [console.warning] [http] GET http://localhost:5180/api/books failed (503), retry 1
 0.8 s GET /api/books 503
 0.8 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 0.8 s [console.warning] [http] GET http://localhost:5180/api/books failed (503), retry 2
 1.8 s GET /api/books 503
 1.8 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 1.8 s [console.warning] [http] GET http://localhost:5180/api/books failed (503), retry 3
 3.8 s GET /api/books 503
 3.8 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 3.8 s [console.error] [http] GET http://localhost:5180/api/books failed: 503
 3.8 s screen: The Bookstore service is not responding. Please try again.

In the first run the user sees "Loading books..." for two seconds, then the list; in the second, a readable message at 3.8 seconds, after four attempts. In the previous version the second case left an empty list under the global banner, and the user could not tell whether the books had sold out or the service was down.

Writing One from Scratch: debugTap

Throughout the series we logged with tap and finalize to see what happens inside a stream. tap sees values, the error and completion, finalize sees the end, but neither tells you on its own when a subscription started and whether the end was an unsubscribe or a complete. You can build that from existing operators, but it reads badly; writing it from scratch is clearer:

// 4. from scratch: log every event of the stream, including when it is unsubscribed
export function debugTap<T>(label: string, log: (line: string) => void = console.log): MonoTypeOperatorFunction<T> {
  return (source) =>
    new Observable<T>((subscriber) => {
      log(`[${label}] subscribe`);
      let finished = false;
      const subscription = source.subscribe({
        next: (value) => {
          log(`[${label}] next ${JSON.stringify(value)}`);
          subscriber.next(value);
        },
        error: (err: unknown) => {
          finished = true;
          log(`[${label}] error ${err instanceof Error ? err.message : String(err)}`);
          subscriber.error(err);
        },
        complete: () => {
          finished = true;
          log(`[${label}] complete`);
          subscriber.complete();
        },
      });
      // teardown: runs on complete, error and unsubscribe; it must stop the source too
      return () => {
        if (!finished) log(`[${label}] unsubscribe`);
        subscription.unsubscribe();
      };
    });
}

All three requirements from the RxJS docs are here: all three observer functions exist, error and complete are passed down, and the teardown unsubscribes from the source. The finished flag is there because the teardown also runs after complete or error; with it, the unsubscribe line is logged only for a subscription that really was dropped from outside. In the search example that was the line of the "em" request. The log parameter lets the virtual-time examples pass a function that also prints the time; the default is console.log.

Generics: Without Falling Back to any

A common shortcut in custom operators is to type them as OperatorFunction<any, any> and move on. We used the same loadingState once that way and once as a generic, and made the same typo:

import { catchError, map, of, startWith, type OperatorFunction } from "rxjs";
import { books } from "../../data/books.ts";
import { loadingState } from "./operators.ts";

// the same operator, typed the lazy way
function loadingStateAny(): OperatorFunction<any, any> {
  return (source) =>
    source.pipe(
      map((data) => ({ status: "success", data })),
      startWith({ status: "loading" }),
      catchError((error) => of({ status: "error", error })),
    );
}

of(books).pipe(loadingStateAny()).subscribe((state) => {
  // a typo in a property name: compiles, prints undefined
  if (state.status === "success") console.log("any:", state.data.lenght);
});

of(books).pipe(loadingState()).subscribe((state) => {
  // @ts-expect-error -- the generic version catches the same typo at compile time
  if (state.status === "success") console.log("generic:", state.data.lenght);
});

any: undefined

The any version compiles and prints undefined at run time. In the generic version, deleting the @ts-expect-error line makes tsc say:

src/11/06-generic-any.ts(22,70): error TS2551: Property 'lenght' does not exist on type 'readonly Book[]'. Did you mean 'length'?

Note that we did not write the generic type by hand. loadingState() takes no arguments, but TypeScript infers T from the type pipe expects and finds readonly Book[]. Our first attempt was loadingState<Book[]>(), and the compiler reported a different error, because the sample data is readonly; most of the time the right type is the one inference finds. @ts-expect-error works like a test here: if the line below it stops being an error, npm run check fails.

Rules for Readable Pipes, and ESLint

The real benefit of custom operators is not technical but readability. The rules that help us in code review:

tap is only for side effects: logging, analytics, setting a signal. A tap that changes the value or starts another request should really be a map or a switchMap.
Split a pipe of more than five or six steps into named operators that each do one thing. settledTerm, searchBooks, retryWithBackoff and loadingState say what they do when you read them; getting the same information out of a ten-line pipe takes time.
Let the name describe the result, not the implementation: settledTerm, not debounceAndTrimAndDistinct.
Take options as an object and give conservative defaults; RxJS 7's retry, share and timeout follow the same path.
Do not write subscribe inside subscribe; a higher-order mapping operator combines the two streams.

We do not have to leave the last two rules and similar ones to human eyes. The best-known ESLint plugin for RxJS was Nicholas Jamieson's eslint-plugin-rxjs; its last version on npm is 5.0.3 (March 2023), and its peer dependency is ESLint 8 only. Its maintained fork is eslint-plugin-rxjs-x: version 1.0.6 came out in August 2026, works with ESLint 10 and typescript-eslint's flat config, and its rule names start with rxjs-x/ instead of rxjs/. Its peer dependencies limit TypeScript to <6.1.0 and RxJS to ^7.2.0, which fits the Angular 22.2 project's TypeScript 6.0. Most rules need type information, since they have to know whether a value is an Observable or a Subject, so typed linting must be on:

// @ts-check
import { defineConfig } from 'eslint/config';
import rxjsX from 'eslint-plugin-rxjs-x';
import tseslint from 'typescript-eslint';

export default defineConfig({
  files: ['src/**/*.ts'],
  extends: [tseslint.configs.recommended, rxjsX.configs.recommended],
  languageOptions: {
    // the RxJS rules need type information: is this an Observable, a Subject, a Subscription?
    parserOptions: { projectService: true, tsconfigRootDir: import.meta.dirname },
  },
  rules: {
    // not in "recommended", but the ones this series cares about
    'rxjs-x/no-ignored-subscription': 'error',
    'rxjs-x/no-exposed-subjects': 'error',
    'rxjs-x/no-floating-observables': 'error',
    // treat takeUntilDestroyed like takeUntil: nothing but finalize & co. may follow it
    'rxjs-x/no-unsafe-takeuntil': ['error', { alias: ['takeUntilDestroyed'] }],
  },
});

We installed it with npm i -D eslint typescript-eslint eslint-plugin-rxjs-x and added "lint": "eslint ." to package.json. The recommended config turns on rules such as no-nested-subscribe, no-unsafe-takeuntil and no-ignored-replay-buffer; we added no-ignored-subscription, no-exposed-subjects and no-floating-observables. By default no-unsafe-takeuntil only looks at takeUntil; we set its alias option so it covers Angular's takeUntilDestroyed too. To try the rules we collected mistakes from earlier posts in a single component:

import { HttpClient } from '@angular/common/http';
import { Component, inject } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { ActivatedRoute } from '@angular/router';
import { map, Subject } from 'rxjs';
import { apiUrl } from '../api';
import type { Book } from '../catalog/book';

@Component({ selector: 'app-book-title-logger', template: '' })
export class BookTitleLogger {
  readonly #http = inject(HttpClient);
  readonly #route = inject(ActivatedRoute);
  readonly titleLogged = new Subject<string>();

  constructor() {
    this.#route.paramMap.pipe(takeUntilDestroyed()).subscribe((params) => {
      this.#http.get<Book>(`${apiUrl}/books/${params.get('isbn')}`).subscribe((book) => {
        this.titleLogged.next(book.title);
      });
    });
    this.#route.paramMap
      .pipe(
        takeUntilDestroyed(),
        map((params) => params.get('isbn')),
      )
      .subscribe((isbn) => console.log(`[book] ${isbn}`));
  }

  markAsSeen(isbn: string): void {
    this.#http.post(`${apiUrl}/books/${isbn}/seen`, {});
  }
}


src/app/lint-demo/book-title-logger.ts
  13:12  error  Subject 'titleLogged' must be private  rxjs-x/no-exposed-subjects
  17:69  error  Nested subscribe calls are forbidden  rxjs-x/no-nested-subscribe
  17:69  error  Ignoring returned subscriptions is forbidden  rxjs-x/no-ignored-subscription
  23:9   error  Applying operators after takeUntil is forbidden  rxjs-x/no-unsafe-takeuntil
  26:8   error  Ignoring returned subscriptions is forbidden  rxjs-x/no-ignored-subscription
  30:5   error  Observables must be subscribed to, returned, converted to a promise and awaited, or be explicitly marked as ignored with the `void` operator  rxjs-x/no-floating-observables

✖ 6 problems (6 errors, 0 warnings)

Behind the six errors are problems we have seen in this series: a public Subject (the Subject types post), a nested subscribe (the previous post), an operator after takeUntilDestroyed, and a POST that is never sent because nothing subscribes to it. no-unsafe-takeuntil was written against the risk of a switchMap after takeUntil whose inner subscription is never cleaned up, and it flags every operator except those on its allow list (finalize, count, last and so on). The map here is actually harmless, but the rule does not tell them apart; putting takeUntilDestroyed last is both safe and readable. The error on line 26 follows from the same thing: no-ignored-subscription lets you ignore a subscription when the last operator ends the stream, like take, first, takeUntil or takeUntilDestroyed; with map last, the rule sees a subscription that never ends.

The real code produced a single error: the placeOrder call on the cart page. The rule does not know that an HTTP request completes after one response. Adding takeUntilDestroyed here would be wrong, because when the user leaves the page it cancels the request in the browser but not the order on the server. The right move is to turn the rule off on that line, with a reason:

    this.#orders
      .placeOrder(this.cart.items())
      // the button comes back whatever happens: success or error
      .pipe(finalize(() => this.placing.set(false)))
      // eslint-disable-next-line rxjs-x/no-ignored-subscription -- one POST: HttpClient completes after the response
      .subscribe({
        next: (order) => this.message.set(`Order ${order.id} placed.`),
        // component level: this page knows what a 409 means for the user
        error: (err: unknown) => this.message.set(errorMessage(err)),
      });

Lint rules are heuristics; when one reports a false positive, do not turn it off for the project, write on that line why it is a false positive. A reason written after -- … saves the next person who reaches that line months later from having the same discussion again.

In Production

Custom operators behave like a small library inside the application, so treat them like one. Keep them in a single file (core/operators.ts), give each a short comment that says what it does, and give them tests. In the next post we test these operators with TestScheduler; the random option of retryWithBackoff exists exactly for that.

Do not ship debugTap in the production build, or if you do, wire its log function to the environment. Calling JSON.stringify on every next is a measurable cost on a stream that returns a large list. During development, though, it is one of the fastest tools for finding leaks: if you leave a page and do not see an unsubscribe line, a subscription stayed open.

Run ESLint in CI and account for the cost of typed linting: rules that need type information build TypeScript's program, so they are noticeably slower than plain linting. In our small project npx eslint . took 2.2 seconds; in a large monorepo you may have to lint only the projects that changed. Finally, watch the peer range of eslint-plugin-rxjs-x: when you upgrade RxJS or TypeScript, the lint step may be the first thing to break.

Common Pitfalls

Not returning a teardown from an operator written from scratch. The source keeps running after the stream ends, like the polling in the prediction.
Not passing the source's error or complete down. The error is lost or the stream never ends.
Letting the generic fall back to any. Typos compile and show up as undefined at run time.
Putting an operator that turns errors into values (loadingState, catchError) before the retry. The retry never runs.
Changing a value or starting a new request inside a tap. That is the job of map or a higher-order mapping operator.
Writing from scratch what an existing operator already does. stockChanges was really distinctUntilChanged().
Silencing a lint error without writing why, or turning the rule off for the whole project.

A Small Experiment

★☆☆ Add an enabled parameter to debugTap; when it is false, return the source as it is without wrapping it. Why is this better than creating a new Observable and skipping the log inside?

★★☆ Add an option to retryWithBackoff that reads the Retry-After header of the HTTP response and, if present, waits for that time instead of the backoff. Add a route to the mock API that returns 503 with Retry-After: 2 and try it in the interceptor.

★★★ Write a minDuration(ms) operator that keeps a value on screen for at least ms milliseconds: even if the request returns in 50 ms, the loading state should be visible for 300 ms so the screen does not flicker. Try it first with existing operators (hint: combineLatest and timer), then write it from scratch, and compare the teardown behaviour of the two versions with debugTap.

In this post we showed the operators' behaviour with log output. In the next post we write the same behaviour as tests: marble testing with TestScheduler, virtual-time tests for settledTerm and retryWithBackoff, a comparison of the same tests with Vitest fake timers, and Bookstore's Angular services with HttpTestingController.