RxJS in Angular 22: HttpClient, Router and Forms Observables, Signal Interop and Subscription Management

Version note: The examples in this post were run with Angular 22.2, RxJS 7.8.2, TypeScript 6.0 and Node.js 24 (the Angular project uses TypeScript 6.0, the range Angular 22.2 supports); every Angular example ran in headless Chromium driven by Playwright 1.65 against the mock Bookstore API of the snapshot.

Predict First

Bookstore, the online bookstore used as the shared example across this site's series, has a "My order" page that shows the status of order 1042. In the time-based operators post we wrote the polling; OrderService.status$ in bookstore-web now does the same with HttpClient (timer(0, 5000), exhaustMap, takeWhile). A first version of the page subscribes in the constructor and stores each answer in a field:

@Component({
  selector: 'app-order-status-page',
  template: `
    <h1>Order 1042</h1>
    <p class="status">Status: {{ status?.status ?? 'Loading...' }}</p>
  `,
})
export class OrderStatusPage {
  protected status: OrderStatus | undefined;

  constructor() {
    inject(OrderService)
      .status$(1042)
      .subscribe((order) => {
        this.status = order;
        console.log(`[order] status ${order.status}`);
      });
  }
}

The page is open for 11 seconds, then the user clicks "Catalog". Two questions: what does the page show after 11 seconds, and what happens to the polling after the user has left?

This post brings the series into Angular 22. Which Angular APIs give you Observables and what their signal-based counterparts are; toSignal, toObservable, rxResource, httpResource and outputFromObservable; why an Observable no longer updates the screen by itself; Bookstore's search box written three ways; and subscription management without leaks. The basics of signals and takeUntilDestroyed are in the Angular series' "Angular 16 Yenilikleri: Signals, Required Input ve takeUntilDestroyed" post, the Angular 22 defaults in "Angular 22: Varsayılan OnPush, Stable Signal Forms ve Angular 12 ile Proje Farkı"; here we look at how RxJS fits in.

Zoneless and OnPush: An Observable Does Not Update the Screen by Itself

We opened the page in headless Chromium:

 0.1 s console: [analytics] page view /orders/1042
 0.1 s request /api/orders/1042/status
 0.3 s console: [order] status Paid
 5.1 s request /api/orders/1042/status
 5.3 s console: [order] status Paid
10.1 s request /api/orders/1042/status
10.3 s console: [order] status Paid
11.1 s screen: Status: Loading...
11.1 s console: [analytics] page view /
11.1 s left the order page
15.1 s request /api/orders/1042/status
15.3 s console: [order] status Shipped
20.1 s request /api/orders/1042/status
20.3 s console: [order] status Shipped

Both answers are bad. Three answers arrived and the console printed every one of them, but the screen still says "Loading...". And after the user left, the polling went on: two more requests at 15 and 20 seconds, for a page that no longer exists, and they would continue until the order is delivered.

The first problem comes from Angular 22's defaults. A new project is zoneless, so nothing patches setTimeout or fetch to start change detection after every asynchronous callback, and components are OnPush unless they say otherwise. Angular refreshes a view when something tells it to: a template event, a signal read in the template that changed, the async pipe, or an explicit markForCheck. Assigning a plain field inside subscribe is none of these. In an older zone-based application the same code happened to work, which is why it is so common. The second problem is older than Angular 22: a manual subscribe lives until the Observable completes, and nothing here ties it to the component's lifetime.

The fixed page is one line:

import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { OrderService } from './order-service';

@Component({
  selector: 'app-order-status-page',
  template: `
    <h1>Order 1042</h1>
    <p class="status">Status: {{ status()?.status ?? 'Loading...' }}</p>
  `,
})
export class OrderStatusPage {
  // toSignal subscribes now and unsubscribes when the page is destroyed
  protected readonly status = toSignal(inject(OrderService).status$(1042));
}

The template now reads status()?.status. Same test:

 0.1 s console: [analytics] page view /orders/1042
 0.1 s request /api/orders/1042/status
 5.1 s request /api/orders/1042/status
10.1 s request /api/orders/1042/status
11.1 s screen: Status: Paid
11.1 s console: [analytics] page view /
11.1 s left the order page

The screen shows "Paid", and the last request is the one at 10 seconds: when the user left, toSignal unsubscribed, exhaustMap unsubscribed from timer, and the polling stopped. That is the general rule of this post: in a zoneless, OnPush application an Observable reaches the screen through the async pipe or toSignal, and both clean up for you.

Observables in Angular's APIs and Their Signal Counterparts

RxJS is part of Angular's public API in several places. Since Angular 16 most of them have a signal-based alternative, and Angular 22 marks the resource APIs and Signal Forms as stable:

Two of them in bookstore-web. Router.events has no signal version and does not need one: it is a stream of events, exactly what RxJS is for. The root component logs every finished navigation, for example for analytics:

  constructor() {
    // Router.events is an Observable: one line per finished navigation, e.g. for analytics
    inject(Router)
      .events.pipe(
        filter((event) => event instanceof NavigationEnd),
        map((event) => event.urlAfterRedirects),
        takeUntilDestroyed(),
      )
      .subscribe((url) => console.log(`[analytics] page view ${url}`));
  }

Since TypeScript 5.5 the compiler infers that event instanceof NavigationEnd narrows the type, so map can read urlAfterRedirects without a type guard. The book detail page, on the other hand, no longer needs ActivatedRoute: with provideRouter(routes, withComponentInputBinding()) the :isbn parameter arrives as a signal input, and rxResource loads the book whenever it changes:

import { HttpClient } from '@angular/common/http';
import { Component, inject, input } from '@angular/core';
import { rxResource } from '@angular/core/rxjs-interop';
import { RouterLink } from '@angular/router';
import { apiUrl } from '../api';
import type { Book } from './book';

@Component({
  selector: 'app-book-page',
  imports: [RouterLink],
  template: `
    @if (book.value(); as b) {
      <h1>{{ b.title }}</h1>
      <p>{{ b.author }} - {{ b.price }} TRY - {{ b.stock }} in stock</p>
    } @else if (book.isLoading()) {
      <p>Loading...</p>
    }
    <a routerLink="/">Back to catalog</a>
  `,
})
export class BookPage {
  readonly #http = inject(HttpClient);
  // filled from the :isbn route parameter by withComponentInputBinding()
  readonly isbn = input.required<string>();

  protected readonly book = rxResource({
    params: () => this.isbn(),
    stream: ({ params: isbn }) => this.#http.get<Book>(`${apiUrl}/books/${isbn}`),
  });
}

console: [analytics] page view /
console: [analytics] page view /books/9780441172719
request /api/books/9780441172719
screen: Dune | Frank Herbert - 450 TRY - 12 in stock
console: [analytics] page view /
console: [analytics] page view /books/9780141439587
request /api/books/9780141439587
screen: Emma | Jane Austen - 320 TRY - 8 in stock

rxResource takes params, a reactive function, and stream, which turns the params into an Observable. When isbn changes, the resource unsubscribes from the previous Observable (like switchMap) and subscribes to the new one; value(), isLoading(), error() and status() are signals the template can read. The type definitions of Angular 22.2 also warn about one combination: if the Observable completes without a value, for example because of catchError(() => EMPTY), the resource fails with NG0991. Return a fallback value instead, as in the error handling post.

toSignal and toObservable

toSignal subscribes immediately, in an injection context, and unsubscribes when that context (the component, the service) is destroyed. Its options decide what the signal holds before the first value. Without options the type is T | undefined, as on the order page above. initialValue gives a starting value, as in the catalog's toSignal(…, { initialValue: [] }). requireSync: true removes undefined from the type for Observables that emit synchronously on subscribe (a BehaviorSubject, a pipe starting with startWith) and throws if they do not. manualCleanup: true keeps the subscription after the context is destroyed, until the Observable completes; it is for the rare case where a component starts something that must outlive it. And as we saw in the error handling post, an error in the Observable is rethrown every time the signal is read, so handle errors in the pipe before toSignal.

toObservable goes the other way. It reads the signal in an effect and emits the current value first, then every change. Two consequences are visible in the search example below: the values arrive asynchronously, not at the moment set is called, and several quick changes in the same turn can collapse into one emission, which is fine for a search box and wrong for an event channel. The first value, the empty starting text, also goes through the pipe; in our run the searched output fired once with "" right after the page loaded.

The Search Box Three Ways

The /search page of bookstore-web contains the same search box three times. All three wait 300 ms after the last key, ignore repeated terms and ask the mock API GET /api/books?q=…. First, RxJS only, starting from a reactive form control:

import { AsyncPipe } from '@angular/common';
import { HttpClient } from '@angular/common/http';
import { Component, inject } from '@angular/core';
import { FormControl, ReactiveFormsModule } from '@angular/forms';
import { catchError, debounceTime, distinctUntilChanged, map, of, switchMap } from 'rxjs';
import { apiUrl } from '../api';
import type { Book } from '../catalog/book';

// 1. RxJS only: the form control's valueChanges, the async pipe renders the result
@Component({
  selector: 'app-search-rxjs',
  imports: [ReactiveFormsModule, AsyncPipe],
  template: `
    <input id="search-rxjs" [formControl]="term" placeholder="Search books" />
    <ul>
      @for (book of results$ | async; track book.isbn) {
        <li>{{ book.title }}</li>
      }
    </ul>
  `,
})
export class SearchRxjs {
  readonly #http = inject(HttpClient);
  protected readonly term = new FormControl('', { nonNullable: true });
  protected readonly results$ = this.term.valueChanges.pipe(
    debounceTime(300),
    map((term) => term.trim()),
    distinctUntilChanged(),
    switchMap((term) =>
      term
        ? this.#http
            .get<Book[]>(`${apiUrl}/books`, { params: { q: term } })
            .pipe(catchError(() => of<Book[]>([])))
        : of<Book[]>([]),
    ),
  );
}

Second, the state is a signal and so is the result; only the part that is about time goes through RxJS. outputFromObservable turns the settled term into a normal component output:

import { HttpClient } from '@angular/common/http';
import { Component, inject, signal } from '@angular/core';
import { outputFromObservable, toObservable, toSignal } from '@angular/core/rxjs-interop';
import { catchError, debounceTime, distinctUntilChanged, map, of, switchMap } from 'rxjs';
import { apiUrl } from '../api';
import type { Book } from '../catalog/book';

// 2. signal in, signal out: RxJS only for the part that is about time
@Component({
  selector: 'app-search-interop',
  template: `
    <input id="search-interop" #box [value]="term()" (input)="term.set(box.value)" />
    <ul>
      @for (book of results(); track book.isbn) {
        <li>{{ book.title }}</li>
      }
    </ul>
  `,
})
export class SearchInterop {
  readonly #http = inject(HttpClient);
  protected readonly term = signal('');

  readonly #settledTerm$ = toObservable(this.term).pipe(
    debounceTime(300),
    map((term) => term.trim()),
    distinctUntilChanged(),
  );

  protected readonly results = toSignal(
    this.#settledTerm$.pipe(
      switchMap((term) =>
        term
          ? this.#http
              .get<Book[]>(`${apiUrl}/books`, { params: { q: term } })
              .pipe(catchError(() => of<Book[]>([])))
          : of<Book[]>([]),
      ),
    ),
    { initialValue: [] },
  );

  // the parent can listen with (searched)="…", like any output()
  readonly searched = outputFromObservable(this.#settledTerm$);
}

Third, no RxJS at all: Angular 22's debounced(), which its type definitions mark as experimental since 22.0, produces a resource that follows the signal 300 ms late, and httpResource builds the request from its value. Returning undefined as the URL means "no request":

import { httpResource } from '@angular/common/http';
import { Component, debounced, signal } from '@angular/core';
import { apiUrl } from '../api';
import type { Book } from '../catalog/book';

// 3. no RxJS: debounced() (experimental in Angular 22) and httpResource
@Component({
  selector: 'app-search-resource',
  template: `
    <input id="search-resource" #box [value]="term()" (input)="term.set(box.value)" />
    @if (results.isLoading()) {
      <p>Searching...</p>
    }
    <ul>
      @for (book of results.value(); track book.isbn) {
        <li>{{ book.title }}</li>
      }
    </ul>
  `,
})
export class SearchResource {
  protected readonly term = signal('');
  readonly #settledTerm = debounced(this.term, 300);

  protected readonly results = httpResource<Book[]>(
    () => {
      const term = this.#settledTerm.value().trim();
      return term ? `${apiUrl}/books?q=${encodeURIComponent(term)}` : undefined;
    },
    { defaultValue: [] },
  );
}

Playwright typed "dune" into each box, one key every 100 ms, then an "s" 350 ms later and a backspace right after it, so the term changed and came back to "dune":

  396 ms console: [search] searched for ""
--- page loaded, nothing typed yet
--- 1. #search-rxjs: type "dune", one key every 100 ms, then "s" 350 ms later and backspace
  641 ms request /api/books?q=dune
 1703 ms screen: ["Dune"] 
--- 2. #search-interop: type "dune", one key every 100 ms, then "s" 350 ms later and backspace
  632 ms request /api/books?q=dune
  633 ms console: [search] searched for "dune"
 1688 ms screen: ["Dune"] 
--- 3. #search-resource: type "dune", one key every 100 ms, then "s" 350 ms later and backspace
  641 ms request /api/books?q=dune
 1699 ms screen: ["Dune"] | "Searching..." seen while typing: true

All three sent exactly one request, for "dune", about 300 ms after the last key; "dunes" lived for 100 ms and never reached the API. The first two did not search for "dune" again after the backspace because of distinctUntilChanged; httpResource did not because the URL it computed was the same. A first version of the third box read the debounced term with chain(this.#settledTerm) from the params context instead of .value(). chain passes the pending state of the debounce on to the HTTP resource, so it showed "Searching..." while the user was typing, and it requested "dune" a second time after the backspace. Both can be what you want, but you should choose it knowingly.

The split behind the table is the rule of this series for Angular 22: time and event flow (debouncing, cancelling, combining, retrying) belongs to RxJS; the state that the screen shows belongs to signals. The conversion happens at the edges, once, with toSignal or toObservable.

Subscription Management

Most subscriptions in a modern Angular component never appear in code: the async pipe, toSignal, rxResource and httpResource all subscribe and unsubscribe for you. For the remaining manual subscribe calls, takeUntilDestroyed() ends the subscription when the component is destroyed. It looks up the DestroyRef of the current injection context, which exists only in a constructor, a field initializer or a factory. We used it in ngOnInit of a test component and in a template method with toSignal:

broken-a [console.error] ERROR RuntimeError: NG0602: toSignal() cannot be called from within a reactive context. Invoking `toSignal` causes new subscriptions every time. Consider moving `toSignal` outside of the reactive context and read the signal value where needed. Find more at https://v22.angular.dev/errors/NG0602
broken-b [console.error] ERROR RuntimeError: NG0203: takeUntilDestroyed() can only be used within an injection context such as a constructor, a factory function, a field initializer, or a function used with `runInInjectionContext`. Find more at https://v22.angular.dev/errors/NG0203

Outside an injection context, pass the DestroyRef yourself; this version ran without errors in the same test component:

readonly #destroyRef = inject(DestroyRef);

ngOnInit(): void {
  interval(1000)
    .pipe(takeUntilDestroyed(this.#destroyRef))
    .subscribe((n) => this.ticks.set(n));
}

The NG0602 error covers a common mistake: toSignal called in a method that the template runs on every change detection would create a new subscription each time. Angular 22 refuses it; create the signal once, in a field.

Two more patterns cause most leaks we see in reviews. The first is an endless source in a root service: an interval or a WebSocket started in a @Service() lives as long as the application, and every component that subscribes without cleanup keeps its own subscription to it. Such a source should be shared with shareReplay({ bufferSize: 1, refCount: true }) from the multicasting post, so that it stops when nobody listens. The second is the nested subscribe:

// anti-pattern: a subscribe inside a subscribe
this.#route.paramMap.subscribe((params) => {
  this.#http.get<Book>(`${apiUrl}/books/${params.get('isbn')}`).subscribe((book) => this.book.set(book));
});

It works in the happy case, but the outer subscription is never cleaned up, a slow request for the previous book can overwrite the current one, and the inner requests cannot be cancelled. It is switchMap written by hand, without its guarantees; the book page above does the same job with rxResource, and a pure RxJS version would be paramMap.pipe(switchMap(…)).

In Production

Angular DevTools shows signals and their dependencies, but not RxJS subscriptions. To find leaks, navigate between pages a few times and watch the Network tab for requests that belong to pages you have left, as in the predict example; or take heap snapshots in Chrome DevTools and look for component instances whose count keeps growing. Logging in finalize during development (or a debug operator, the topic of the custom operator post) shows exactly when each subscription ends.

Migrating an existing application to zoneless often surfaces the stale-screen problem from the predict example in many places at once. Search for subscribe( in components and check each one: a value that ends up in the template should go through toSignal, the async pipe or a signal.set. Angular's provideCheckNoChangesConfig({ exhaustive: true }) helps during development by reporting bindings that changed without Angular being notified.

Finally, be careful with experimental APIs in production code. debounced() may still change in a minor version; if you use it, keep it behind a small wrapper so that switching to toObservable + debounceTime touches one file.

Common Pitfalls

• Assigning an Observable's value to a plain field in subscribe in a zoneless, OnPush component. The screen does not update; use toSignal, async or a signal.
• A manual subscribe without takeUntilDestroyed. Polling and timers continue after the user leaves the page.
• Calling takeUntilDestroyed() without an argument in ngOnInit or an event handler. Outside an injection context it throws NG0203; pass a DestroyRef.
• Calling toSignal in a method or getter that the template uses. Angular 22 throws NG0602; create the signal once in a field.
• Reading a toSignal signal whose Observable can fail, without catchError. Every read rethrows the error.
• Nesting subscribe calls. Use switchMap, rxResource or httpResource.
• Using toObservable as an event channel. Changes are delivered asynchronously and can collapse into one; events belong in a Subject or an output().

A Small Experiment

★☆☆ Add filter((term) => term !== '') before outputFromObservable in search-interop.ts so that the searched output no longer fires for the empty starting text. Does the result list still clear when the user deletes the whole term? If not, where should the filter go?

★★☆ Rewrite the book page with ActivatedRoute.paramMap, switchMap and toSignal, without withComponentInputBinding and rxResource. Navigate quickly from Dune to Emma with the mock API's DELAY_MS=1500 and check in the Network tab that the request for Dune is cancelled.

★★★ Give the "My order" page a "Refresh now" button that starts a new status request immediately and restarts the five-second timer, without ever having two requests in flight. Write it once with RxJS (merge of the timer and the click stream into exhaustMap) and once with rxResource and its reload(); compare what each version does when the button is clicked during a request.

Some pipes in this series have started to repeat: the debounce-and-search block, the retry with backoff, the loading state around a request. In the next post we turn such blocks into custom operators with pipe() and OperatorFunction, write one from scratch with new Observable, and keep long pipes readable with naming, a debug operator and ESLint rules for RxJS.