RxJS avansat: operatori de ordin superior

Semnalele au preluat starea, dar evenimentele în timp rămân teritoriul RxJS: tastări, click-uri, request-uri care se suprapun, reîncercări, websocket-uri. Partea cea mai importantă, și cea mai des greșită, sunt operatorii de ordin superior. Fiecare valoare pornește un nou Observable, de obicei un request HTTP. Întrebarea reală e: ce faci când vine valoarea următoare, iar request-ul de dinainte încă rulează?

Există patru răspunsuri: anulezi, rulezi în paralel, pui la coadă sau ignori. Fiecare are un operator. Dacă alegi greșit, rezultatul e un bug de tip race condition: rezultate vechi care le suprascriu pe cele noi, comenzi duble, salvări în altă ordine.

Problema: Observable de Observable

Un map care întoarce un request produce un Observable<Observable<T>>: un stream de request-uri la care nu s-a abonat nimeni. Operatorii *Map se abonează ei la fluxurile interioare și le „aplatizează” într-un singur stream de rezultate. Diferența dintre ei e doar politica de suprapunere.

// „Pentru fiecare valoare, pornește un Observable” — problema e ce faci când vine valoarea următoare
// iar cea de dinainte încă rulează.
clicks$.pipe(map((id) => http.get(`/api/items/${id}`)));       // ✗ Observable<Observable<Item>>
clicks$.pipe(switchMap((id) => http.get(`/api/items/${id}`))); // ✓ Observable<Item> — plus o politică

Cei patru operatori

// Typeahead: doar ultimul request contează; cele vechi sunt ANULATE (HttpClient anulează și XHR/fetch-ul)
readonly results = toSignal(
  this.query$.pipe(
    debounceTime(300),
    distinctUntilChanged(),
    switchMap((q) => this.api.search(q).pipe(catchError(() => of([])))),
  ),
  { initialValue: [] },
);

Unde pui catchError

O eroare completează stream-ul în care apare. Dacă catchError stă după switchMap, primul request eșuat închide tot fluxul: căutarea nu mai răspunde, fără nicio eroare în consolă. Pune-l în proiecția interioară. Atunci doar acel request e înlocuit cu o valoare de rezervă, iar stream-ul exterior continuă.

// ✗ catchError DUPĂ switchMap: prima eroare completează stream-ul exterior — căutarea „moare”
query$.pipe(
  switchMap((q) => api.search(q)),
  catchError(() => of([])),
);

// ✓ catchError ÎN interiorul proiecției: doar request-ul acela eșuează, stream-ul rămâne viu
query$.pipe(
  switchMap((q) => api.search(q).pipe(catchError(() => of([])))),
);

Retry cu backoff exponențial

retry({ count, delay }) se reabonează la sursă după o eroare. Funcția delay primește eroarea și numărul reîncercării și întoarce un Observable care spune când se reîncearcă. Dublarea pauzei la fiecare pas (300, 600, 1200 ms…) nu mai lovește în server cât timp acesta are probleme. Aruncă eroarea mai departe pentru coduri care nu se vor rezolva singure (401, 404). Pentru HTTP, retry-ul stă de obicei într-un interceptor (vezi HttpClient & interceptori). Pentru o conexiune care trebuie să supraviețuiască restartărilor serverului — cu jitter, ca o mie de clienți să nu revină în aceeași secundă — vezi WebSockets.

import { retry, timer } from 'rxjs';

http.get('/api/report').pipe(
  retry({
    count: 4,                                                // maximum 4 reîncercări
    delay: (error, retryCount) => {
      if (error.status === 401 || error.status === 404) throw error; // nu reîncerca erorile „definitive”
      return timer(300 * 2 ** (retryCount - 1));              // 300, 600, 1200, 2400 ms
    },
    resetOnSuccess: true,
  }),
);

share și shareReplay

Un Observable „rece”, cum e http.get, se execută la fiecare subscribe. Trei abonați înseamnă trei request-uri. share() face ca abonații simultani să împartă o singură execuție. shareReplay păstrează și ultima valoare pentru cei care vin mai târziu, deci funcționează ca un cache. Capcana e la shareReplay(1) simplu: are refCount: false, deci nu se dezabonează niciodată de la sursă. Pe un stream care nu se termină (websocket, interval), sursa rulează și după ce toți abonații au plecat.

Un detaliu fin la refCount: true: cache-ul se păstrează doar dacă sursa a apucat să se termine. Dacă toți abonații pleacă înainte de complete (de exemplu firstValueFrom, take(1) sau o componentă distrusă în timpul request-ului), shareReplay se dezabonează de la sursă și o ia de la capăt la următorul abonat.

// Fără partajare: fiecare subscribe = încă o execuție (încă un request)
const user$ = http.get<User>('/api/me');
user$.subscribe(); user$.subscribe();                 // 2 request-uri

// share(): abonații SIMULTANI împart o execuție; după complete se resetează
const shared$ = http.get<User>('/api/me').pipe(share());

// shareReplay: păstrează ultima valoare pentru abonații TÂRZII (cache)
const cached$ = http.get<User>('/api/me').pipe(shareReplay({ bufferSize: 1, refCount: true }));
// refCount: true → când pleacă ultimul abonat, se dezabonează și de la sursă.
// shareReplay(1) (refCount: false) pe o sursă care nu se termină = sursa rulează la nesfârșit.

Combinarea fluxurilor

combineLatest emite ori de câte ori se schimbă oricare sursă, după ce fiecare a emis o dată. E potrivit pentru filtre combinate. withLatestFrom emite doar la sursa principală și ia valoarea curentă din celelalte („la click, salvează formularul curent”). forkJoin așteaptă să se termine toate și emite o singură dată: e un Promise.all pentru request-uri.

// combineLatest: emite la fiecare schimbare, după ce TOATE au emis o dată — filtre combinate
combineLatest([category$, sort$]).pipe(switchMap(([c, s]) => api.list(c, s)));

// withLatestFrom: emite doar la sursa principală, luând ultima valoare din celelalte
save$.pipe(withLatestFrom(form$), concatMap(([, value]) => api.save(value)));

// forkJoin: așteaptă ca TOATE să se termine, o singură emisie — request-uri paralele independente
forkJoin({ user: api.me(), settings: api.settings() });

// zip: împerechează a n-a valoare din fiecare — rar util în UI

Fără subscribe-uri uitate

Orice subscribe manual într-o componentă trebuie închis. takeUntilDestroyed() o face automat la distrugerea componentei. Pentru valori afișate în template, toSignal() e și mai simplu: se abonează și se dezabonează singur (vezi Signals ↔ RxJS).

import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

export class Live {
  constructor() {
    interval(1000)
      .pipe(takeUntilDestroyed()) // în context de injecție; altfel: takeUntilDestroyed(this.destroyRef)
      .subscribe((n) => this.tick.set(n));
  }
}
// Pentru stare afișată în template, toSignal(obs$) se dezabonează singur (vezi pagina Signals ↔ RxJS).

Când DA, când NU

Capcane frecvente

• mergeMap pentru căutare. Răspunsurile vin în orice ordine, iar unul vechi îl poate suprascrie pe cel nou. Folosește switchMap.
• switchMap pentru salvări. Un save anulat la jumătate poate fi deja procesat pe server, sau poate nu. Pentru scrieri, concatMap sau exhaustMap. Mai multe despre scrieri care sosesc în altă ordine la actualizări optimiste.
• Subscribe în subscribe. Pierzi anularea și gestiunea erorilor. Orice „subscribe, apoi în callback alt subscribe” e un operator *Map.
• catchError la nivelul exterior într-un stream care trebuie să trăiască: oprește totul la prima eroare.
• shareReplay(1) pe surse infinite: scurgere de memorie și conexiuni. Folosește { bufferSize: 1, refCount: true }. Cum le găsești afli pe pagina despre memory leaks.
• Retry pe 4xx: reîncerci la nesfârșit un request care nu va reuși niciodată. Filtrează după status.

Fișă rapidă

OPERATOR      LA O VALOARE NOUĂ CÂT CEA VECHE RULEAZĂ        FOLOSEȘTE PENTRU
switchMap     anulează-o pe cea veche                       căutare, parametri de rută, „ultimul câștigă”
mergeMap      rulează-le în paralel (opțional: concurrency) operații independente, upload-uri
concatMap     pune-o la coadă, în ordine                     salvări ordonate, cozi de comenzi
exhaustMap    ignor-o                                        submit, login, refresh manual

catchError    ÎN proiecția interioară, ca stream-ul exterior să rămână viu
retry         { count, delay: (err, n) => timer(base * 2 ** (n - 1)) } — nu reîncerca 4xx
share()       abonați simultani împart execuția
shareReplay   { bufferSize: 1, refCount: true } — cache + dezabonare corectă
combineLatest filtre combinate · withLatestFrom „la click, ia valoarea curentă” · forkJoin „toate, o dată”
takeUntilDestroyed() / toSignal()   fără subscribe-uri uitate