Angular Observables RxJS Operators

Observable Contract and Creation

RxJS Observables model asynchronous streams and cancellation. Angular uses them heavily for HTTP and event pipelines, while signals provide synchronous reactive state. Learn subscription ownership, operator intent, and safe interop between the two models.

An Observable represents values delivered over time. A subscriber receives next notifications, followed by at most one terminal notification: complete or error. After completion or error, that execution cannot emit again. Unsubscribing stops observation and runs the producer teardown without sending complete.

Subscription behavior comes from the source. Cold sources create independent work for each subscriber; HttpClient is cold, so two subscriptions can send two requests. Hot sources such as DOM events, WebSockets, and Subjects can produce independently of one subscriber. Do not assume every Observable is cold or that every subscription is harmless.

Use creation functions instead of constructing Observable manually for common sources. Construct one directly only when wrapping a callback API or resource that needs explicit teardown.

of emits the supplied values synchronously and completes.
from converts an iterable, Promise, or Observable-like input into an Observable.
defer runs a factory separately for each subscriber, which is useful when creation must use current state.
interval and timer create time-based streams; fromEvent wraps browser or framework event targets.
EMPTY completes immediately, NEVER neither emits nor completes, and throwError creates an erroring source.

Observable with Teardown

import { Observable } from 'rxjs';

const resize$ = new Observable<DOMRectReadOnly>(subscriber => {
  const observer = new ResizeObserver(entries => {
    const rect = entries[0]?.contentRect;
    if (rect) subscriber.next(rect);
  });

  observer.observe(document.body);

  // Runs when the subscriber unsubscribes.
  return () => observer.disconnect();
});

const subscription = resize$.subscribe({
  next: rect => console.log(rect.width),
  error: error => console.error(error),
  complete: () => console.log('finished')
});

subscription.unsubscribe();

The constructor is appropriate because ResizeObserver is a callback API with a resource that must be disconnected. For arrays, Promises, timers, and events, prefer an RxJS creation function.

Subscription Ownership in Angular

Every subscription needs a clear owner. Use AsyncPipe for values consumed only by a template; it subscribes when the view needs the value, exposes the latest emission, marks the component for checking, and unsubscribes when the view is destroyed or the bound Observable reference changes.

Use takeUntilDestroyed for an imperative subscription that produces a side effect in TypeScript. Call it inside an injection context, or pass an injected DestroyRef when the pipeline is created later in a method. Finite HttpClient requests normally complete, but unsubscribing can still abort an in-flight request when a component disappears.

Avoid nested subscribe calls and arrays of subscriptions. Compose dependent work with switchMap, concatMap, mergeMap, or exhaustMap, and let one outer subscription own the pipeline.

Template value: expose Observable<T> and bind with AsyncPipe.
Imperative component effect: pipe through takeUntilDestroyed.
One-value operation: use take(1) or firstValueFrom only when Promise-style control flow is genuinely clearer.
Application-lifetime service: document why the subscription is intentionally long-lived and how it is stopped in tests or shutdown.

AsyncPipe for Template Data

import { AsyncPipe } from '@angular/common';
import { HttpClient } from '@angular/common/http';
import { Component, inject } from '@angular/core';

@Component({
  selector: 'app-users',
  standalone: true,
  imports: [AsyncPipe],
  template: `
    @if (users$ | async; as users) {
      <ul>
        @for (user of users; track user.id) {
          <li>{{ user.name }}</li>
        }
      </ul>
    } @else {
      <p>Loading users...</p>
    }
  `
})
export class UsersComponent {
  private http = inject(HttpClient);
  users$ = this.http.get<readonly User[]>('/api/users');
}

interface User { id: number; name: string; }

takeUntilDestroyed for an Imperative Effect

import { Component, DestroyRef, inject } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { NotificationService } from './notification.service';

@Component({ selector: 'app-notifications', template: `` })
export class NotificationsComponent {
  private notifications = inject(NotificationService);
  private destroyRef = inject(DestroyRef);

  startListening(): void {
    this.notifications.messages$.pipe(
      takeUntilDestroyed(this.destroyRef)
    ).subscribe(message => {
      console.log('Notification:', message);
    });
  }
}

Observables and Signals

Signals hold a synchronously readable current value and fit application state. Observables model asynchronous event sequences and provide cancellation, timing, combination, and concurrency operators. Neither API replaces the other.

Keep a stream as an Observable while operator composition is the main job. Convert once at a UI ownership boundary when a template benefits from a signal; repeated conversion can create repeated subscriptions.

Feature	Observables (RxJS)	Signals
Best for	Async streams, HTTP, events	Synchronous reactive state
Value model	Notifications over time: next, error, and complete	One synchronously readable current value
Timing and cancellation	Operators model timing, concurrency, cancellation, and retries	Updates state synchronously; effects run during Angular scheduling
Consumption	Subscription, AsyncPipe, or an interop API	Read by calling the signal
Template usage	With async pipe	Direct call: {{ mySignal() }}

toSignal Interop

toSignal subscribes immediately and exposes the latest source value as a signal. Supply initialValue when the source cannot emit synchronously, or use requireSync only for a source guaranteed to emit during subscription.

Angular ties the subscription to the current injection context by default. Create the signal once in a field or constructor, and handle source errors before conversion because an unhandled error is thrown when the signal is read.

Create toSignal once and reuse the returned signal; each call creates a subscription.
Without initialValue, the signal returns undefined until the first emission. requireSync verifies a synchronous first emission.
If the source errors, reading the signal throws. If the source completes, the signal keeps its most recent value.
toObservable tracks a signal with an effect; after any initial synchronous replay, later notifications are asynchronous and rapid signal writes are coalesced after stabilization.

toSignal() Example

import { Component, inject } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { toSignal } from '@angular/core/rxjs-interop';

interface Post { id: number; title: string; }

@Component({
  selector: 'app-posts',
  standalone: true,
  template: `
    @for (post of posts(); track post.id) {
      <p>{{ post.id }}. {{ post.title }}</p>
    }
  `
})
export class PostsComponent {
  private http = inject(HttpClient);

  // Observable converted to Signal - no subscribe() needed
  posts = toSignal(
    this.http.get<Post[]>('https://jsonplaceholder.typicode.com/posts?_limit=5'),
    { initialValue: [] }
  );
}

toObservable() Search Pipeline

import { HttpClient } from '@angular/common/http';
import { Component, inject, signal } from '@angular/core';
import { toObservable } from '@angular/core/rxjs-interop';
import { debounceTime, distinctUntilChanged, switchMap } from 'rxjs';

@Component({ selector: 'app-catalog', template: `` })
export class CatalogComponent {
  private http = inject(HttpClient);

  query = signal('');
  private query$ = toObservable(this.query);

  results$ = this.query$.pipe(
    debounceTime(250),
    distinctUntilChanged(),
    switchMap(query =>
      this.http.get<readonly Product[]>('/api/products', {
        params: { q: query }
      })
    )
  );
}

interface Product { id: number; name: string; }

Operator Decisions

Choose an operator by the behavior the user should observe when new values arrive. Flattening operators differ in cancellation, ordering, concurrency, and whether new work is accepted.

Operators run in pipe order and return a new Observable; they do not mutate the source. Keep transformations pure, use tap only for diagnostics or explicit side effects, and place catchError inside or outside a flattening operator according to which stream should terminate.

Operator	Use it for
map	Synchronous value transformation without starting another stream.
filter	Dropping values that do not satisfy a predicate.
debounceTime + distinctUntilChanged	Waiting for typing to pause and ignoring an unchanged query.
switchMap	Replacing obsolete work, such as search requests or route-dependent loads.
concatMap	Queueing writes that must finish in source order.
mergeMap	Running independent inner work concurrently when order is unimportant.
exhaustMap	Ignoring repeated triggers while one operation is active, such as a submit button.
combineLatest	Recalculating from the latest values of several continuing streams.
forkJoin	Waiting for several completing streams and emitting their final values once.
tap	Logging or non-transforming observation; do not hide business state changes here.
take / takeUntil	Completing after a count or notifier; use takeUntilDestroyed for Angular lifecycle ownership.
startWith	Providing an initial emission, often before combineLatest consumes form valueChanges.
scan	Accumulating state across emissions without an external mutable variable.
withLatestFrom	Letting one source trigger while reading the latest values from supporting streams.
timeout	Failing a stream that does not emit or complete within an accepted time boundary.

Cancellable Search Stream

import { HttpClient } from '@angular/common/http';
import { Component, inject } from '@angular/core';
import { FormControl } from '@angular/forms';
import { catchError, debounceTime, distinctUntilChanged, map, of, switchMap } from 'rxjs';

interface Result { id: number; label: string; }

@Component({ selector: 'app-search', template: `` })
export class SearchComponent {
  private http = inject(HttpClient);
  searchControl = new FormControl('', { nonNullable: true });

  results$ = this.searchControl.valueChanges.pipe(
    map(query => query.trim()),
    debounceTime(250),
    distinctUntilChanged(),
    switchMap(query =>
      query === ''
        ? of([])
        : this.http.get<Result[]>('/api/search', { params: { q: query } }).pipe(
            catchError(() => of([]))
          )
    )
  );
}

switchMap unsubscribes from an obsolete request when a newer query arrives. The inner catchError keeps the outer valueChanges stream alive for later searches.

Combining Streams and Initial Values

Combination operators differ in what triggers output and whether inputs must complete. Choose from the user interaction, not from similar-looking method names. The most common bug is waiting forever because one input has not produced its first value.

combineLatest emits whenever any input changes after every input has emitted once. withLatestFrom emits only when its primary source emits. forkJoin waits for every input to complete and then emits their final values once; it never emits if an input never completes.

Operator	Output rule	Typical Angular use
combineLatest	Any input changes after all have emitted	Filter, sort, and route-state view models
withLatestFrom	Primary source emits	Submit click plus latest form or session state
forkJoin	All inputs complete	Parallel finite HTTP requests
zip	One value is available from every input	Pairing values by position
merge	Any input emits	Treating several same-type event sources as one stream

Seed FormControl.valueChanges with startWith(control.value) when the current value belongs in the initial view model.
Do not use forkJoin with an interval, Subject, or another source that does not complete.
Give combined values names by using an object shape instead of relying on tuple positions in large pipelines.

combineLatest Filter View Model

import { FormControl } from '@angular/forms';
import { combineLatest, map, startWith } from 'rxjs';

search = new FormControl('', { nonNullable: true });
category = new FormControl('all', { nonNullable: true });

viewModel$ = combineLatest({
  query: this.search.valueChanges.pipe(startWith(this.search.value)),
  category: this.category.valueChanges.pipe(startWith(this.category.value)),
  products: this.productsService.products$
}).pipe(
  map(({ query, category, products }) => ({
    query,
    category,
    visibleProducts: products.filter(product =>
      (category === 'all' || product.category === category) &&
      product.name.toLowerCase().includes(query.trim().toLowerCase())
    )
  }))
);

Shared Streams

A Subject is both an observer and an observable. It is useful at imperative boundaries, but exposing writable subjects throughout an application makes ownership difficult to trace. Prefer a private subject with a read-only observable or signal-based service state.

share and shareReplay turn one source execution into a shared execution. Scope the shared observable to the intended owner, decide whether late subscribers need a replay, and define refresh, invalidation, error reset, and teardown behavior before calling the result a cache.

Tool	Behavior
Subject<T>	Multicasts future values and has no current value for late subscribers.
BehaviorSubject<T>	Requires a starting value and synchronously supplies the latest value to new subscribers.
ReplaySubject<T>	Replays a configured number or time window of earlier values; bound the retained history.
share()	Shares one live source execution while subscribers overlap.
shareReplay({ bufferSize: 1, refCount: true })	Shares and replays the latest value while allowing teardown when the last subscriber leaves.

Do not use shareReplay as an unbounded global cache without an invalidation and lifetime policy.
Do not call next on a public subject from unrelated components; expose an intent method on the owning service.
Remember that every subscription to a cold HTTP observable can issue another request unless execution is deliberately shared.

Shared HTTP Data with Explicit Refresh

import { HttpClient } from '@angular/common/http';
import { Injectable, inject } from '@angular/core';
import { shareReplay, startWith, Subject, switchMap } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class UsersStore {
  private http = inject(HttpClient);
  private refreshRequest = new Subject<void>();

  readonly users$ = this.refreshRequest.pipe(
    startWith(undefined),
    switchMap(() => this.http.get<readonly User[]>('/api/users')),
    shareReplay({ bufferSize: 1, refCount: true })
  );

  refresh(): void {
    this.refreshRequest.next();
  }
}

interface User { id: number; name: string; }

Overlapping consumers share one current request and late consumers receive the latest result. refresh() is an explicit invalidation path; refCount allows teardown when no consumers remain.

Error Handling Retry and Finalization

An Observable sends either complete or error once; neither channel can emit another value afterward. Place recovery where the application can choose whether to replace a failed inner operation, terminate the feature stream, or ask the user to retry.

Retry re-subscribes to the source. Limit attempts, delay them, and use retry only when repeating the operation is safe. A GET is commonly retryable; a payment or create request may duplicate work unless the server provides idempotency guarantees.

API	Decision
catchError	Return a fallback observable or rethrow a meaningful failure. Placement controls which part of the chain terminates.
retry({ count, delay })	Retry transient idempotent work with a limit and delay; do not blindly repeat unsafe writes.
finalize	Release loading state whether the stream completes, errors, or is unsubscribed.
throwError	Return a typed error Observable when an operator branch must fail instead of emit a normal value.

Bounded Retry with Loading Cleanup

import { HttpClient } from '@angular/common/http';
import { inject, signal } from '@angular/core';
import { catchError, defer, EMPTY, finalize, retry, timer } from 'rxjs';

private http = inject(HttpClient);
loading = signal(false);
errorMessage = signal<string | null>(null);

users$ = defer(() => {
  this.loading.set(true);
  this.errorMessage.set(null);

  return this.http.get<readonly User[]>('/api/users');
}).pipe(
  retry({
    count: 2,
    delay: (_error, retryCount) => timer(retryCount * 500)
  }),
  catchError(error => {
    console.error(error);
    this.errorMessage.set('Users could not be loaded.');
    return EMPTY;
  }),
  finalize(() => this.loading.set(false))
);

defer resets UI state for each subscription. retry is bounded and delayed, catchError chooses the visible failure state, and finalize runs for completion, error, or unsubscription.

Angular Interop and Promise Boundaries

Keep interop at a clear ownership boundary. Converting back and forth throughout a feature creates extra subscriptions and hides which reactive model owns the state.

Use firstValueFrom or lastValueFrom only when an API requires a Promise or async function. firstValueFrom resolves on the first emission and unsubscribes; lastValueFrom waits for completion and resolves with the final value. Both reject on source error, and both can remain pending when the source contract never satisfies their condition.

API	Use it when	Important boundary
takeUntilDestroyed	An imperative Angular subscription needs lifecycle teardown	Pass DestroyRef outside an injection context
toSignal	A component needs the latest stream value as signal state	Creates an immediate subscription; reuse the result
toObservable	Signal state must enter an RxJS operator pipeline	Later notifications occur after signal stabilization
outputFromObservable	A component output is driven by an Observable	Call only in a component or directive property initializer
outputToObservable	A component OutputRef must enter an RxJS pipeline	Use direct OutputRef subscription when no operators are needed
rxResource	Observable loading should expose resource value, status, and error signals	The stream factory runs again when resource parameters change
firstValueFrom / lastValueFrom	A Promise-only boundary must consume a stream	Guarantee an emission or completion with take, timeout, or source design

Do not convert HttpClient to a Promise merely to convert it back into an Observable later.
Do not call toSignal repeatedly in a getter, template helper, or method.
Handle Observable errors before toSignal when the UI needs a normal error state instead of a thrown signal read.

firstValueFrom with a Time Boundary

import { firstValueFrom, timeout } from 'rxjs';

async function readCurrentSession(): Promise<Session> {
  return firstValueFrom(
    sessionService.session$.pipe(timeout(3000))
  );
}

The timeout prevents a session stream that never emits from leaving the Promise pending forever. Keep the Observable form when cancellation or multiple values still matter to the caller.

Testing Observable Pipelines

Test a pipeline at the level where its timing contract matters. Synchronous of-based tests are enough for pure mapping. Use TestScheduler marble tests for debounce, delays, cancellation, retries, and combination timing. Use Angular HTTP testing utilities for HttpClient requests instead of calling a real backend.

Assert emitted values, completion or error, and subscription timing when cancellation is part of the requirement. A test that checks only the final value can miss duplicate requests, a stream that never completes, or an inner subscription that was not cancelled.

Keep operator logic in exported functions when it can be tested without creating a component.
Test the empty, error, and rapid-input paths, not only the successful single emission.
For switchMap, verify the stale inner stream is unsubscribed when a newer source value arrives.
For shared HTTP streams, verify the number of backend requests and the refresh behavior.

RxJS TestScheduler Example

import { filter, map } from 'rxjs';
import { TestScheduler } from 'rxjs/testing';

describe('normalizedQuery$', () => {
  it('trims, lowercases, and removes short queries', () => {
    const scheduler = new TestScheduler((actual, expected) => {
      expect(actual).toEqual(expected);
    });

    scheduler.run(({ cold, expectObservable }) => {
      const source$ = cold('-a-b-c-|', {
        a: '  Angular ',
        b: 'Rx',
        c: ' Signals '
      });

      const result$ = source$.pipe(
        map(value => value.trim().toLowerCase()),
        filter(value => value.length >= 3)
      );

      expectObservable(result$).toBe('-a---c-|', {
        a: 'angular',
        c: 'signals'
      });
    });
  });
});

Confirm the page outcome

Stream Review

8 checks

I can explain next, error, complete, unsubscribe, and producer teardown without treating them as the same event.
I can distinguish cold and hot streams and predict when a subscription starts duplicate work.
I can choose AsyncPipe, takeUntilDestroyed, or a finite self-completing stream for subscription ownership.
I can select switchMap, concatMap, mergeMap, or exhaustMap from cancellation and ordering requirements.
I can choose combineLatest, withLatestFrom, forkJoin, zip, or merge from the required trigger and completion behavior.
I can place catchError, retry, and finalize without accidentally terminating a long-lived source or repeating an unsafe write.
I can decide whether state should remain an Observable, become a signal, or cross a Promise or Angular interop boundary.
I can test values, timing, cancellation, errors, completion, and shared request counts.

Try this next

Angular Observables Coding Exercises

0 of 4 completed

Build a Cancellable Product Search
Create a non-nullable search control, normalize and debounce its values, cancel stale HTTP requests with switchMap, and expose loading, result, empty, and error states without nested subscriptions. Put catchError inside switchMap so one failed request does not stop future searches.
Diagnose Duplicate HTTP Requests
Bind one cold users request in two places, confirm the backend sees two calls, then repair ownership by binding once or sharing the request with an explicit refresh policy. Count requests before and after the change; do not assume shareReplay is correct without testing invalidation.
Combine Filters Without a Silent Stream
Combine search, category, and product streams into one view model. Demonstrate why valueChanges initially blocks combineLatest, then fix it with current control values. Use startWith(control.value) on each control stream that must contribute immediately.
Prove switchMap Cancellation
Write a TestScheduler case with two rapid queries and verify that only the latest inner result is emitted and the stale inner subscription ends early. Assert subscription marbles as well as output marbles.

Stream Review Questions

Subscriptions. Each Start click creates another interval, and all of them update the same counter. Unsubscribe the previous interval, disable Start while it is running, or use switchMap so a new start replaces the old stream. ngOnDestroy only helps when the component itself is removed.

A signal is read synchronously, but many Observables—especially HTTP requests—do not have a value at construction time.

Keep the Observable when the feature depends on stream operators, cancellation, retries, debouncing, or combining multiple asynchronous sources.

Previous Next