Observables from zero

An Observable is a stream of values over time. It can emit nothing, one value or an endless series, and it ends with complete or error — or never. In Angular you meet it wherever events happen: HttpClient, router events, valueChanges in reactive forms, WebSockets.

Signals have taken over the job of holding state; Observables remain the tool for events in time — clicks, keystrokes, responses that can be cancelled. This page covers what you need before Signals and RxJS and advanced RxJS.

Anatomy of an Observable

An Observable is a function waiting to be called. new Observable(subscriber => …) stores it; every subscribe() calls it with a new subscriber, which receives values through next() and the end through error() or complete(). The function returns a teardown: what to stop when someone unsubscribes.

import { Observable } from 'rxjs';

// the function runs on EVERY subscribe(), not here
const ticker$ = new Observable<number>((subscriber) => {
  let n = 0;
  const id = setInterval(() => subscriber.next(++n), 1000);

  // teardown: runs on unsubscribe(), complete() or error()
  return () => clearInterval(id);
});

Lazy, and one run per subscriber

Nothing happens when you create an Observable — not even an HTTP request. The work starts at subscribe(), and every subscriber gets its own run: two subscribers to http.get() mean two requests. This kind of Observable is called cold.

Without sharing, every subscriber starts its own setInterval: the values differ and the number of running executions grows. share() makes the stream hot: one execution whose values go to everyone, stopped when the last subscriber leaves. shareReplay({ bufferSize: 1, refCount: true }) also hands the latest value to anyone who arrives late — notice that a new subscriber no longer shows “…”.

const user$ = this.http.get<User>('/api/user'); // nothing is sent yet

user$.subscribe(); // request 1
user$.subscribe(); // request 2 — every subscriber gets its own run

// one request, shared by everyone who subscribes while it's running
const shared$ = user$.pipe(shareReplay({ bufferSize: 1, refCount: true }));

Operators: pipe() builds a new stream

An operator takes an Observable and returns another one; the source doesn't change. pipe() chains them from top to bottom, and each value goes through the whole chain before the next one starts.

source$.pipe(
  filter((n) => n % 2 === 0),
  map((n) => n * 10),
).subscribe({ next, error, complete });

A few things to notice. filter and take decide whether a value gets through; map and scan change it; debounceTime holds a value back until the source has been quiet for 500 ms, so the burst becomes a single value. take(3) completes the subscriber even though the source keeps going. And after complete() or error(), the source is finished for good: nothing more can come through.

import { debounceTime, distinctUntilChanged, filter, map } from 'rxjs';

const query$ = this.searchControl.valueChanges.pipe(
  map((text) => text.trim()),
  filter((text) => text.length >= 2), // not for one letter
  debounceTime(300), // wait until typing pauses
  distinctUntilChanged(), // the same query again: skip it
);

The operators you'll use every week

Who unsubscribes?

An Observable that completes — an HTTP request — cleans up by itself. One that never completes — interval, a WebSocket, router events — keeps running until someone unsubscribes. In Angular you rarely call unsubscribe() by hand: the async pipe and toSignal() do it when the component is destroyed, and takeUntilDestroyed() does it for a subscribe() in the class. What happens when nobody does is on Memory leaks.

// 1. in the template: the async pipe unsubscribes when the component is destroyed
// <p>{{ ticker$ | async }}</p>

// 2. as a signal: the same, and it reads like any other signal
readonly tick = toSignal(ticker$, { initialValue: 0 });

// 3. a subscribe() in the class: tie it to the component's lifetime
constructor() {
  ticker$.pipe(takeUntilDestroyed()).subscribe((n) => this.log(n));
}

Subject and BehaviorSubject

A Subject is an Observable you push values into yourself with next() — handy for turning callbacks into a stream. A BehaviorSubject also remembers the latest value. For state in a component or a service, a signal does the BehaviorSubject's job with less code; keep Subjects for events.

// a Subject: you push the values in
const clicks$ = new Subject<MouseEvent>();
button.addEventListener('click', (e) => clicks$.next(e));

// a BehaviorSubject remembers the latest value…
const user$ = new BehaviorSubject<User | null>(null);
user$.getValue();

// …but for state, a signal is simpler
readonly user = signal<User | null>(null);

Signal or Observable?

The usual shape in Angular 22: Observables at the edges (HTTP, sockets, DOM events), turned into signals with toSignal() as soon as they reach a component. More in Signals and RxJS.

@Component({
  template: `
    @if (user(); as u) {
      <h2>{{ u.name }}</h2>
      <p>{{ u.email }}</p>
    }
  `,
})
export class Profile {
  private readonly http = inject(HttpClient);
  // the Observable stays at the edge; the component reads a signal
  readonly user = toSignal(this.http.get<User>('/api/user'));
}

Common pitfalls

• Forgetting to subscribe. this.http.post(…) without subscribe() sends nothing.
• Two async pipes on the same HTTP Observable. Two requests. Convert it once with toSignal(), or read it once with @let.
• subscribe() inside subscribe(). Nested subscriptions can't be cancelled together and run out of order. Use switchMap and its siblings — see advanced RxJS.
• An endless stream with no way out. interval keeps running after the component is gone. Use takeUntilDestroyed() or toSignal().
• shareReplay(1) without refCount. The source keeps running even after every subscriber has left.

Cheat sheet

Observable             a function called on every subscribe(); emits 0..∞ values, then complete or error
lazy                   creating it does nothing; subscribe() starts the work (even an HTTP request)
cold                   one run per subscriber: two subscribers = two requests
share()                one run for everyone; stops when the last subscriber leaves
shareReplay(…)         also replays the latest value; always { bufferSize: 1, refCount: true }
pipe(a, b, c)          a new stream; values go through a, then b, then c
teardown               the function returned in new Observable(…); runs on unsubscribe
unsubscribing          async pipe, toSignal(), takeUntilDestroyed(); HTTP completes by itself
Subject                you call next() yourself; BehaviorSubject also keeps the latest value
signal vs Observable   a value now vs events over time; toSignal() at the component boundary