RxJS essentials for Angular

The handful of operators that solve real problems, the subject patterns that avoid memory leaks, and when the answer is a signal instead.

The operators that earn their place

import { Component, inject, DestroyRef } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { FormControl, ReactiveFormsModule } from '@angular/forms';
import { debounceTime, distinctUntilChanged, filter, switchMap, catchError, of, shareReplay } from 'rxjs';

@Component({
  selector: 'app-product-search',
  imports: [ReactiveFormsModule],
  template: `<input [formControl]="term" placeholder="Search products" />`
})
export class ProductSearchComponent {
  private readonly catalog = inject(CatalogService);

  readonly term = new FormControl('', { nonNullable: true });

  readonly results = toSignal(
    this.term.valueChanges.pipe(
      debounceTime(250),                        // wait for a typing pause
      distinctUntilChanged(),                   // ignore unchanged values
      filter((value) => value.length === 0 || value.length >= 3),
      switchMap((value) =>                        // cancel the previous request
        this.catalog.search(value).pipe(
          catchError(() => of([] as Product[]))  // recover inside the switch
        )
      ),
      shareReplay({ bufferSize: 1, refCount: true })
    ),
    { initialValue: [] as Product[] }
  );
}

Subjects and lifetime

import { Subject, BehaviorSubject } from 'rxjs';
import { Injectable, Component, inject, DestroyRef } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

@Injectable({ providedIn: 'root' })
export class NotificationService {
  // BehaviorSubject: replays the current value to new subscribers.
  private readonly queue = new BehaviorSubject<Notification[]>([]);
  readonly queue$ = this.queue.asObservable();

  push(note: { message: string; kind: 'info' | 'error' }) {
    const list = [...this.queue.value, { ...note, id: crypto.randomUUID() }];
    this.queue.next(list);
  }

  dismiss(id: string) {
    this.queue.next(this.queue.value.filter((note) => note.id !== id));
  }
}

@Component({ selector: 'app-bell', template: `<span>{{ (queue$ | async)?.length ?? 0 }}</span>` })
export class BellComponent {
  private readonly notifications = inject(NotificationService);
  private readonly destroyRef = inject(DestroyRef);
  readonly queue$ = this.notifications.queue$;

  constructor() {
    // takeUntilDestroyed without arguments works in an injection context.
    this.notifications.queue$
      .pipe(takeUntilDestroyed(this.destroyRef))
      .subscribe((list) => console.log('queue size', list.length));
  }
}

• An unsubscribed infinite observable keeps running and holds its closures. takeUntilDestroyed is the shortest reliable fix and needs no ngOnDestroy.
• async pipe subscriptions are cleaned up automatically, which is why binding the observable rather than subscribing manually is the safer default.
• A BehaviorSubject in a root service survives navigation, so its state is global. If it should reset per route, provide the service at the route level instead of in root.
• Check whether a signal would do: a BehaviorSubject that is only read by templates is usually a signal that has not been converted yet.

Choosing between RxJS and signals

// Signals: state and derivation. Simple, synchronous, always current.
readonly term = signal('');
readonly isLongEnough = computed(() => this.term().length >= 3);

// RxJS: timing and composition. Explicit about when things happen.
readonly term$ = toObservable(this.term).pipe(
  debounceTime(250),
  distinctUntilChanged(),
  switchMap((value) => this.catalog.search(value))
);

// Back to a signal at the template boundary.
readonly results = toSignal(this.term$, { initialValue: [] as Product[] });

Both models coexist deliberately. Write state as signals, keep the temporal logic in observables, and convert at the boundary with toSignal. What you should avoid is a component that mixes both styles in the same property — that is where the confusing bugs live.

FAQ

Related

Signals, computed values and effects HTTP, interceptors and async resources

Last refreshed 2026-09-18.