Les Observables depuis zéro

Un Observable est un flux de valeurs dans le temps. Il peut n’émettre rien, une seule valeur ou une série sans fin, et il se termine par complete ou error — ou jamais. Dans Angular, vous le croisez partout où il y a des événements : HttpClient, les événements du routeur, valueChanges des reactive forms, les WebSockets.

Les signals ont repris le rôle de garder l’état ; les Observables restent l’outil des événements dans le temps — clics, frappes, réponses qu’on peut annuler. Cette page couvre ce qu’il faut savoir avant Signals et RxJS et RxJS avancé.

Anatomie d’un Observable

Un Observable est une fonction qui attend d’être appelée. new Observable(subscriber => …) la garde ; chaque subscribe() l’appelle avec un nouveau subscriber, qui reçoit les valeurs par next() et la fin par error() ou complete(). La fonction renvoie un teardown : ce qu’il faut arrêter quand quelqu’un se désabonne.

import { Observable } from 'rxjs';

// la fonction s’exécute à CHAQUE subscribe(), pas ici
const ticker$ = new Observable<number>((subscriber) => {
  let n = 0;
  const id = setInterval(() => subscriber.next(++n), 1000);

  // teardown : s’exécute au unsubscribe(), au complete() ou à l’error()
  return () => clearInterval(id);
});

Paresseux, et une exécution par abonné

Il ne se passe rien quand vous créez un Observable — pas même une requête HTTP. Le travail commence au subscribe(), et chaque abonné reçoit sa propre exécution : deux abonnés à http.get(), ce sont deux requêtes. On dit d’un tel Observable qu’il est froid (cold).

Sans partage, chaque abonné lance son propre setInterval : les valeurs diffèrent et le nombre d’exécutions en cours augmente. share() rend le flux chaud (hot) : une seule exécution, dont les valeurs vont à tout le monde, arrêtée quand le dernier abonné part. shareReplay({ bufferSize: 1, refCount: true }) donne en plus la dernière valeur à ceux qui arrivent en retard — remarquez qu’un nouvel abonné n’affiche plus « … ».

const user$ = this.http.get<User>('/api/user'); // rien n’est encore envoyé

user$.subscribe(); // requête 1
user$.subscribe(); // requête 2 — chaque abonné a sa propre exécution

// une seule requête, partagée par tous ceux qui s’abonnent pendant qu’elle tourne
const shared$ = user$.pipe(shareReplay({ bufferSize: 1, refCount: true }));

Les opérateurs : pipe() construit un nouveau flux

Un opérateur prend un Observable et en renvoie un autre ; la source ne change pas. pipe() les enchaîne de haut en bas, et chaque valeur traverse toute la chaîne avant que la suivante ne parte.

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

Quelques points à observer. filter et take décident si une valeur passe ; map et scan la transforment ; debounceTime retient une valeur jusqu’à ce que la source soit restée silencieuse 500 ms, donc la rafale devient une seule valeur. take(3) termine l’abonné alors que la source continue. Et après complete() ou error(), la source est finie pour de bon : plus rien ne peut passer.

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

const query$ = this.searchControl.valueChanges.pipe(
  map((text) => text.trim()),
  filter((text) => text.length >= 2), // pas pour une seule lettre
  debounceTime(300), // attendre une pause dans la frappe
  distinctUntilChanged(), // la même recherche : on l’ignore
);

Les opérateurs que vous utiliserez chaque semaine

Qui se désabonne ?

Un Observable qui se termine — une requête HTTP — fait le ménage tout seul. Un Observable qui ne se termine jamais — interval, un WebSocket, les événements du routeur — continue jusqu’à ce que quelqu’un se désabonne. Dans Angular, vous appelez rarement unsubscribe() vous-même : le pipe async et toSignal() le font à la destruction du composant, et takeUntilDestroyed() le fait pour un subscribe() dans la classe. Ce qui se passe quand personne ne le fait est décrit dans Fuites mémoire.

// 1. dans le template : le pipe async se désabonne à la destruction du composant
// <p>{{ ticker$ | async }}</p>

// 2. en signal : pareil, et il se lit comme n’importe quel signal
readonly tick = toSignal(ticker$, { initialValue: 0 });

// 3. un subscribe() dans la classe : le lier à la durée de vie du composant
constructor() {
  ticker$.pipe(takeUntilDestroyed()).subscribe((n) => this.log(n));
}

Subject et BehaviorSubject

Un Subject est un Observable dans lequel vous poussez vous-même des valeurs avec next() — pratique pour transformer des callbacks en flux. Un BehaviorSubject garde aussi la dernière valeur. Pour l’état d’un composant ou d’un service, un signal fait le travail du BehaviorSubject avec moins de code ; gardez les Subjects pour les événements.

// un Subject : c’est vous qui poussez les valeurs
const clicks$ = new Subject<MouseEvent>();
button.addEventListener('click', (e) => clicks$.next(e));

// un BehaviorSubject garde la dernière valeur…
const user$ = new BehaviorSubject<User | null>(null);
user$.getValue();

// …mais pour de l’état, un signal est plus simple
readonly user = signal<User | null>(null);

Signal ou Observable ?

La forme habituelle dans Angular 22 : des Observables aux bords (HTTP, sockets, événements DOM), transformés en signals avec toSignal() dès qu’ils atteignent un composant. Plus de détails dans Signals et RxJS.

@Component({
  template: `
    @if (user(); as u) {
      <h2>{{ u.name }}</h2>
      <p>{{ u.email }}</p>
    }
  `,
})
export class Profile {
  private readonly http = inject(HttpClient);
  // l’Observable reste au bord ; le composant lit un signal
  readonly user = toSignal(this.http.get<User>('/api/user'));
}

Pièges fréquents

• Oublier de s’abonner. this.http.post(…) sans subscribe() n’envoie rien.
• Deux pipes async sur le même Observable HTTP. Deux requêtes. Convertissez-le une fois avec toSignal(), ou lisez-le une fois avec @let.
• subscribe() dans subscribe(). Les abonnements imbriqués ne peuvent pas être annulés ensemble et s’exécutent dans le désordre. Utilisez switchMap et ses cousins — voir RxJS avancé.
• Un flux sans fin et sans sortie. interval continue après la disparition du composant. Utilisez takeUntilDestroyed() ou toSignal().
• shareReplay(1) sans refCount. La source continue de tourner même quand tous les abonnés sont partis.

Aide-mémoire

Observable             une fonction appelée à chaque subscribe() ; émet 0..∞ valeurs, puis complete ou error
paresseux              le créer ne fait rien ; subscribe() lance le travail (même une requête HTTP)
froid (cold)           une exécution par abonné : deux abonnés = deux requêtes
share()                une exécution pour tous ; s’arrête quand le dernier abonné part
shareReplay(…)         rejoue aussi la dernière valeur ; toujours { bufferSize: 1, refCount: true }
pipe(a, b, c)          un nouveau flux ; les valeurs passent par a, puis b, puis c
teardown               la fonction renvoyée dans new Observable(…) ; s’exécute au unsubscribe
désabonnement          pipe async, toSignal(), takeUntilDestroyed() ; HTTP se termine tout seul
Subject                vous appelez next() vous-même ; BehaviorSubject garde aussi la dernière valeur
signal ou Observable   une valeur maintenant ou des événements dans le temps ; toSignal() à la frontière du composant