Observables de la zero

Un Observable e un flux de valori în timp. Poate să nu emită nimic, o singură valoare sau o serie fără sfârșit și se termină cu complete sau error — ori niciodată. În Angular îl întâlnești oriunde apar evenimente: HttpClient, evenimentele router-ului, valueChanges din reactive forms, WebSockets.

Signal-urile au preluat sarcina de a ține starea; Observables rămân unealta pentru evenimente în timp — click-uri, taste, răspunsuri care pot fi anulate. Pagina asta acoperă ce îți trebuie înainte de Signals și RxJS și de RxJS avansat.

Anatomia unui Observable

Un Observable e o funcție care așteaptă să fie apelată. new Observable(subscriber => …) o păstrează; fiecare subscribe() o apelează cu un subscriber nou, care primește valori prin next() și sfârșitul prin error() sau complete(). Funcția întoarce un teardown: ce trebuie oprit când cineva se dezabonează.

import { Observable } from 'rxjs';

// funcția rulează la FIECARE subscribe(), nu aici
const ticker$ = new Observable<number>((subscriber) => {
  let n = 0;
  const id = setInterval(() => subscriber.next(++n), 1000);

  // teardown: rulează la unsubscribe(), complete() sau error()
  return () => clearInterval(id);
});

Leneș și câte o execuție pentru fiecare abonat

Nu se întâmplă nimic când creezi un Observable — nici măcar o cerere HTTP. Munca începe la subscribe(), iar fiecare abonat primește propria execuție: doi abonați la http.get() înseamnă două cereri. Un astfel de Observable se numește cold.

Fără partajare, fiecare abonat pornește propriul setInterval: valorile diferă, iar numărul de execuții care rulează crește. share() face fluxul hot: o singură execuție, ale cărei valori ajung la toți, oprită când pleacă ultimul abonat. shareReplay({ bufferSize: 1, refCount: true }) îi dă în plus ultima valoare oricui vine mai târziu — observă că un abonat nou nu mai arată „…”.

const user$ = this.http.get<User>('/api/user'); // încă nu se trimite nimic

user$.subscribe(); // cererea 1
user$.subscribe(); // cererea 2 — fiecare abonat primește propria execuție

// o singură cerere, împărțită de toți cei care se abonează cât timp rulează
const shared$ = user$.pipe(shareReplay({ bufferSize: 1, refCount: true }));

Operatori: pipe() construiește un flux nou

Un operator primește un Observable și întoarce altul; sursa nu se schimbă. pipe() îi înlănțuie de sus în jos, iar fiecare valoare trece prin tot lanțul înainte să pornească următoarea.

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

Câteva lucruri de observat. filter și take decid dacă o valoare trece; map și scan o schimbă; debounceTime ține o valoare pe loc până când sursa tace 500 ms, așa că rafala devine o singură valoare. take(3) îl încheie pe abonat deși sursa continuă. Iar după complete() sau error(), sursa s-a terminat definitiv: nu mai poate trece nimic.

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

const query$ = this.searchControl.valueChanges.pipe(
  map((text) => text.trim()),
  filter((text) => text.length >= 2), // nu pentru o literă
  debounceTime(300), // așteaptă o pauză în tastare
  distinctUntilChanged(), // aceeași căutare din nou: o sărim
);

Operatorii pe care îi folosești în fiecare săptămână

Cine se dezabonează?

Un Observable care se încheie — o cerere HTTP — face curat singur. Unul care nu se încheie niciodată — interval, un WebSocket, evenimentele router-ului — rulează până se dezabonează cineva. În Angular rareori apelezi unsubscribe() de mână: pipe-ul async și toSignal() o fac când componenta e distrusă, iar takeUntilDestroyed() o face pentru un subscribe() din clasă. Ce se întâmplă când n-o face nimeni vezi la Memory leaks.

// 1. în template: pipe-ul async se dezabonează când componenta e distrusă
// <p>{{ ticker$ | async }}</p>

// 2. ca signal: la fel, și se citește ca orice alt signal
readonly tick = toSignal(ticker$, { initialValue: 0 });

// 3. un subscribe() în clasă: îl legi de durata de viață a componentei
constructor() {
  ticker$.pipe(takeUntilDestroyed()).subscribe((n) => this.log(n));
}

Subject și BehaviorSubject

Un Subject e un Observable în care împingi tu valori, cu next() — util ca să transformi callback-uri într-un flux. Un BehaviorSubject ține minte și ultima valoare. Pentru starea unei componente sau a unui serviciu, un signal face treaba BehaviorSubject-ului cu mai puțin cod; păstrează Subject-urile pentru evenimente.

// un Subject: împingi tu valorile
const clicks$ = new Subject<MouseEvent>();
button.addEventListener('click', (e) => clicks$.next(e));

// un BehaviorSubject ține minte ultima valoare…
const user$ = new BehaviorSubject<User | null>(null);
user$.getValue();

// …dar pentru stare, un signal e mai simplu
readonly user = signal<User | null>(null);

Signal sau Observable?

Forma obișnuită în Angular 22: Observables la margini (HTTP, socket-uri, evenimente DOM), transformate în signal-uri cu toSignal() imediat ce ajung într-o componentă. Mai multe în Signals și RxJS.

@Component({
  template: `
    @if (user(); as u) {
      <h2>{{ u.name }}</h2>
      <p>{{ u.email }}</p>
    }
  `,
})
export class Profile {
  private readonly http = inject(HttpClient);
  // Observable-ul rămâne la margine; componenta citește un signal
  readonly user = toSignal(this.http.get<User>('/api/user'));
}

Capcane frecvente

• Uiți să te abonezi. this.http.post(…) fără subscribe() nu trimite nimic.
• Două pipe-uri async pe același Observable HTTP. Două cereri. Transformă-l o dată cu toSignal() sau citește-l o dată cu @let.
• subscribe() în subscribe(). Abonamentele imbricate nu pot fi anulate împreună și rulează în dezordine. Folosește switchMap și frații lui — vezi RxJS avansat.
• Un flux fără sfârșit și fără ieșire. interval rulează și după ce componenta a dispărut. Folosește takeUntilDestroyed() sau toSignal().
• shareReplay(1) fără refCount. Sursa continuă să ruleze chiar și după ce au plecat toți abonații.

Fișă rapidă

Observable             o funcție apelată la fiecare subscribe(); emite 0..∞ valori, apoi complete sau error
leneș                  crearea lui nu face nimic; subscribe() pornește munca (chiar și o cerere HTTP)
cold                   o execuție per abonat: doi abonați = două cereri
share()                o execuție pentru toți; se oprește când pleacă ultimul abonat
shareReplay(…)         repetă și ultima valoare; mereu { bufferSize: 1, refCount: true }
pipe(a, b, c)          un flux nou; valorile trec prin a, apoi b, apoi c
teardown               funcția întoarsă în new Observable(…); rulează la unsubscribe
dezabonare             pipe-ul async, toSignal(), takeUntilDestroyed(); HTTP se încheie singur
Subject                apelezi tu next(); BehaviorSubject ține și ultima valoare
signal vs Observable   o valoare acum vs evenimente în timp; toSignal() la granița componentei