Operatori RXJS: Guida pratica con casi d'uso reali

Cos’è la programmazione reattiva con RxJS?

RxJS (Reactive Extensions for JavaScript) è una libreria per la programmazione reattiva che usa gli Observable per gestire sequenze asincrone di eventi. In Angular è integrata nativamente: ogni chiamata HttpClient, ogni FormControl.valueChanges, ogni Router event è un Observable.

La vera potenza di RxJS sta nei suoi operatori: funzioni pure che trasformano, filtrano, combinano e gestiscono gli stream di dati. Conoscerli bene fa la differenza tra un’applicazione reattiva ben progettata e un groviglio di callback nested.

1. map — Trasformare i dati

Analogia: come Array.map(), ma su uno stream continuo nel tempo.

Caso d’uso: normalizzare la risposta di un’API

// Scenario: l'API restituisce { data: User[], total: number }
// ma il componente si aspetta solo User[]
this.http.get<ApiResponse<User[]>>('/api/users')
  .pipe(
    map(response => response.data),
    map(users => users.map(u => ({
      ...u,
      fullName: `${u.firstName} ${u.lastName}`,
      avatarUrl: u.avatar || '/assets/default-avatar.png'
    })))
  )
  .subscribe(users => this.users = users);

Regola pratica: usa map ogni volta che vuoi trasformare la forma dei dati senza cambiare il ritmo delle emissioni.

2. filter — Filtrare le emissioni

Lascia passare solo i valori che soddisfano un predicato booleano.

Caso d’uso: processare solo eventi di tastiera rilevanti

fromEvent<KeyboardEvent>(document, 'keydown')
  .pipe(
    filter(event => event.key === 'Enter' || event.key === 'Escape'),
    map(event => event.key)
  )
  .subscribe(key => {
    if (key === 'Enter') this.confirmAction();
    if (key === 'Escape') this.closeModal();
  });

Regola pratica: usa filter il prima possibile nella catena per ridurre il lavoro degli operatori successivi.

3. tap — Effetti collaterali senza modificare

tap esegue un’azione sul valore corrente ma lascia passare il valore invariato. Ideale per logging e aggiornamento dello stato UI.

Caso d’uso: logging + spinner di caricamento

this.http.get<Product[]>('/api/products')
  .pipe(
    tap(() => this.loading = true),
    map(products => products.filter(p => p.inStock)),
    tap(products => console.log(`Prodotti disponibili: ${products.length}`)),
    tap(() => this.loading = false)
  )
  .subscribe(products => this.products = products);

Regola pratica: non usare tap per logica di business, solo per side-effect (log, analytics, spinner).

4. debounceTime + distinctUntilChanged — Ottimizzare la ricerca

Questi due operatori lavorano spesso in coppia per evitare di inondare il backend con richieste inutili mentre l’utente digita.

• debounceTime(ms): aspetta che l’utente smetta di digitare per N ms prima di emettere
• distinctUntilChanged(): non emette se il valore è identico al precedente

Caso d’uso: search box con chiamata API

this.searchControl.valueChanges.pipe(
  debounceTime(400),
  distinctUntilChanged(),
  filter(term => term.length >= 2),
  switchMap(term => this.productService.search(term)),
  takeUntil(this.destroy$)
).subscribe(results => this.results = results);

Senza questi operatori, ogni battuta genererebbe una richiesta HTTP. Con essi si riduce il numero di chiamate di oltre il 90%.

5. switchMap — Cancellare le richieste precedenti

switchMap è il più usato per richieste HTTP in risposta a eventi utente. Quando arriva una nuova emissione, cancella automaticamente l’Observable precedente e si iscrive al nuovo.

Caso d’uso: navigazione dinamica con route params

this.route.paramMap.pipe(
  map(params => params.get('id')),
  filter(id => !!id),
  switchMap(id => this.articleService.getById(id!))
).subscribe(article => this.article = article);

Caso d’uso: autocomplete con risultati aggiornati

// Se l'utente digita "ang" poi "angu" poi "angul",
// vengono eseguite solo le richieste per "angul"
searchTerm$.pipe(
  debounceTime(300),
  switchMap(term => this.searchService.getSuggestions(term))
).subscribe(suggestions => this.suggestions = suggestions);

Regola: usa switchMap quando solo l’ultima risposta è rilevante.

6. mergeMap — Esecuzione parallela

A differenza di switchMap, mergeMap NON cancella le richieste precedenti: le esegue tutte in parallelo e emette i risultati man mano che arrivano.

Caso d’uso: upload multiplo di file

from(selectedFiles).pipe(
  mergeMap(file => this.uploadService.upload(file).pipe(
    map(result => ({ file: file.name, url: result.url, status: 'ok' })),
    catchError(err => of({ file: file.name, status: 'error', error: err.message }))
  ))
).subscribe(result => this.uploadResults.push(result));

Attenzione: usa mergeMap(fn, maxConcurrency) per limitare le richieste parallele simultanee.

7. concatMap — Esecuzione sequenziale garantita

concatMap esegue le operazioni una alla volta in ordine, aspettando il completamento della precedente prima di iniziare la successiva.

Caso d’uso: salvataggio record in sequenza

from([record1, record2, record3]).pipe(
  concatMap(record => this.db.save(record))
).subscribe({
  next: result => console.log('Salvato:', result),
  complete: () => console.log('Tutti i record salvati in ordine!')
});

Regola d’oro degli Higher-Order Operators:

• Solo l’ultimo conta → switchMap
• Tutti in parallelo → mergeMap
• Uno alla volta in ordine → concatMap

8. forkJoin — Attendere più richieste parallele

forkJoin è il corrispettivo reattivo di Promise.all(): aspetta che tutti gli Observable completino ed emette un oggetto con gli ultimi valori di ciascuno.

Caso d’uso: caricare dati da più endpoint per una dashboard

ngOnInit() {
  forkJoin({
    user: this.userService.getCurrent(),
    stats: this.statsService.getDashboard(),
    notifications: this.notifService.getUnread(),
    projects: this.projectService.getAll()
  }).pipe(
    takeUntil(this.destroy$)
  ).subscribe({
    next: ({ user, stats, notifications, projects }) => {
      this.user = user;
      this.stats = stats;
      this.notifications = notifications;
      this.projects = projects;
      this.isReady = true;
    },
    error: () => this.error = 'Errore nel caricamento della dashboard'
  });
}

Limitazione: se uno dei sorgenti genera un errore, forkJoin propaga l’errore e cancella gli altri. Usa catchError su ogni Observable per tollerare fallimenti parziali.

9. combineLatest — Combinare stream live

A differenza di forkJoin, combineLatest non aspetta il completamento: ricalcola l’output ogni volta che uno dei sorgenti emette un nuovo valore.

Caso d’uso: tabella filtrata con più filtri indipendenti

combineLatest([
  this.searchControl.valueChanges.pipe(startWith('')),
  this.categoryControl.valueChanges.pipe(startWith('all')),
  this.statusControl.valueChanges.pipe(startWith('active'))
]).pipe(
  debounceTime(200),
  switchMap(([search, category, status]) =>
    this.productService.getFiltered({ search, category, status })
  ),
  takeUntil(this.destroy$)
).subscribe(products => this.filteredProducts = products);

Nota: startWith() è necessario perché combineLatest non emette finché tutti i sorgenti non hanno emesso almeno una volta.

10. takeUntil — Prevenire i memory leak

Ogni subscription non gestita è un potential memory leak. takeUntil è il pattern standard in Angular per completare automaticamente le subscription quando il componente viene distrutto.

Caso d’uso: pattern standard di cleanup

@Component({ ... })
export class MyComponent implements OnInit, OnDestroy {
  private destroy$ = new Subject<void>();

  ngOnInit() {
    this.dataService.stream$.pipe(
      takeUntil(this.destroy$)
    ).subscribe(data => this.data = data);

    interval(5000).pipe(
      takeUntil(this.destroy$),
      switchMap(() => this.statsService.refresh())
    ).subscribe(stats => this.stats = stats);
  }

  ngOnDestroy() {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

Alternativa moderna (Angular 16+): usa takeUntilDestroyed() da @angular/core/rxjs-interop senza gestire manualmente il Subject.

11. catchError — Gestione degli errori

catchError intercetta gli errori nello stream e permette di recuperare con un fallback Observable invece di interrompere la subscription.

Caso d’uso: fallback con dati cached + notifica utente

this.http.get<Product[]>('/api/products').pipe(
  catchError(err => {
    console.error('[ProductService] Errore API:', err);
    this.toastService.showError('Impossibile caricare i prodotti. Uso dati cache.');
    return of(this.cacheService.getProducts() ?? []);
  })
).subscribe(products => this.products = products);

Caso d’uso: retry automatico per errori di rete

this.http.get('/api/critical-data').pipe(
  retry({ count: 3, delay: 1000 }),
  catchError(err => {
    this.errorService.report(err);
    return throwError(() => new Error('Servizio non disponibile'));
  })
).subscribe(data => this.data = data);

12. shareReplay — Caching e condivisione dello stream

shareReplay(1) trasforma un Cold Observable in uno Hot Observable condiviso, con caching dell’ultimo valore. La HTTP call viene eseguita una sola volta indipendentemente dal numero di subscriber.

Caso d’uso: configurazione globale caricata una sola volta

@Injectable({ providedIn: 'root' })
export class ConfigService {
  // La chiamata HTTP viene eseguita UNA SOLA VOLTA,
  // tutti i componenti ricevono la stessa risposta cached
  readonly config$ = this.http.get<AppConfig>('/api/config').pipe(
    shareReplay(1)
  );

  constructor(private http: HttpClient) {}
}

// In qualsiasi componente, zero chiamate duplicate
@Component({ ... })
export class NavbarComponent {
  config$ = inject(ConfigService).config$; // usa la cache
}

Riepilogo: quando usare ogni operatore

Padroneggiare questi operatori significa scrivere codice Angular più leggibile, più efficiente e praticamente privo di bug asincroni. Il segreto è partire sempre dalla domanda: “Cosa deve succedere quando arriva una nuova emissione mentre quella precedente è ancora in volo?” — la risposta ti indica quasi sempre l’operatore giusto.