Observables in Angular

Siamo finalmente giunti a uno degli argomenti più affascinanti e discussi di Angular: l’impiego di RxJS e dei suoi Observable.

La premessa da fare su questo argomento è che RxJS non è un componente di Angular ma una libreria indipendente che, per altro, è implementata e disponibile in diversi linguaggi di programmazione e su diverse piattaforme tecnologiche. Lo scopo di questa libreria è quello di creare un contesto di operazioni applicabili su flussi di eventi (in inglese event stream) in modo estendibile, in modo da poter abilitare un paradigma di programmazione parente della programmazione funzionale chiamato programmazione reattiva; questa definizione è racchiusa in sintesi nel nome stesso della libreria, che sta per “Reactive extensions for JavaScript“.

Per comprendere il principio alla base della programmazione reattiva dobbiamo tornare a parlare di stato e transizioni, cioè aggiornamenti dello stato. 

Immaginiamo di voler implementare un’applicazione molto comune, ma anche estremamente complessa e versatile come un foglio di calcolo; all’interno di questo foglio l’utente potrà inserire dei dati in alcune celle, oppure derivare il valore di queste da quello di altre, combinandole secondo diverse formule.

In un’applicazione del genere dovremmo tenere traccia dinamicamente di quali celle discendono da altre, e in che modo; ad ogni modifica di una cella da parte dell’utente, dovremo aggiornare secondo le formule impostate tutte le celle derivate a cascata.

In un contesto simile l’utilità della programmazione reattiva diventa particolarmente chiara: essenzialmente, anziché rappresentare lo stato di ogni cella come un valore e aggiornarlo imperativamente, lo rappresenteremo come un flusso di dati emessi nel tempo; questo flusso potrà, poi, essere una fonte primaria di eventi oppure una derivata combinando altri flussi.

In questa sezione vedremo in maniera più o meno esaustiva tutto ciò che RxJS comporta nel contesto di Angular; va precisato, però, che per scandagliare in modo approfondito tutti i tipi e le funzioni di questa libreria occorrerebbe una guida a sé stante.

Iniziamo riepilogando come ci siamo imbattuti in RxJS:

• i metodi di HttpClient restituiscono observables invece che promises per rappresentare le operazioni asincrone;
• tutte le API del framework basate su eventi espongono questi sotto forma di observables.

Cominciamo dando un esempio molto stringato di come nasce un Observable:

import { Observable } from 'rxjs';


const obs = new Observable<number>(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);


  subscriber.complete();
});

In pratica abbiamo eseguito il costruttore passando una funzione che accetta un subscriber su cui vengono chiamati il metodo next per emettere valori e poi il metodo complete per chiudere lo stream.

Confrontiamo questo costruttore con quello di una promise:

const pro = new Promise<number>(resolve => resolve(1));

In pratica, un observable è come una promise in grado di emettere in modo asincrono un numero indefinito di valori prima di completare. Perciò una promise è in definitiva un caso particolare di observable; questo spiega perché gli sviluppatori di Angular hanno scelto di usare gli observable, adottando un’API uniforme per le attività asincrone e per i flussi di eventi.

Nella realtà concreta il costruttore Observable non si usa praticamente mai: la libreria mette a disposizione una incredibile quantità di factory functions pronte per generare observable a partire da eventi del DOM, timeout, interval e ogni sorta di API asincrona o basata su eventi di JavaScript.

Qui non ci interesseremo di vedere queste factory ma solo gli operatori utili per svolgere le operazioni più frequenti nel contesto di Angular.

Se un observable funziona come una promise con valori multipli, possiamo pensarlo come un array asincrono, possiamo cioè immaginare che i valori emessi nel tempo da un flusso di eventi siano come gli elementi iterati da un array o da una struttura analoga. In effetti, esiste un’altro costrutto JavaScript più vicino agli array e molto simile ad un observable: le generator functions.

Vediamone una:

const gen = function* () {
  yield 1;
  yield 2;
  yield 3;

  return;
}

for (const n of gen()) console.log(n);

La generator function di questo esempio ha un comportamento analogo a quello del primo observable che abbiamo creato: yield equivale a subscriber.next e return a subscriber.complete; l’API per sfogliare i valori emessi dal generatore è, in questo caso, sincrona e integrata nella sintassi di iterazione di JavaScript, la stessa che usiamo per gli array.

Quest’ultima differenza è cruciale per spiegare l’esigenza che sta dietro a RxJS: infatti in JavaScript esistono già costrutti sintattici utili a scandagliare diversi valori, ma il punto è che non esiste un modo per regolare alla fonte il tempismo con cui questi valori sono emessi.

La seguente tabella riassume perfettamente quanto detto finora:

Per esecuzione a trazione si intende che è il codice fruitore a “tirare fuori” (pull) il dato da un’espressione, come avviene chiamando funzioni sincrone oppure ciclando su un array;

per esecuzione a spinta invece si intende che è la fonte stessa del dato a determinare il tempismo con il quale il codice fruitore verrà eseguito, emettendo o “spingendo” (push) il nuovo valore come avviene con le promise native e, appunto, con gli observable.

Questa caratteristica degli observable di essere in grado di rappresentare tanti e tali casi d’uso rende la programmazione reattiva un vero e proprio stile di programmazione informatica, tale per cui ogni dato viene rappresentato come una fonte di eventi e ogni trasformazione dei dati viene rappresentata come un flusso di eventi derivato; il tutto è immutabile e tardivo (lazy), perciò solo in coda a tutte le trasformazioni troviamo una chiamata a subscribe che concretizza il tutto in un’esecuzione reale.

Siamo, dunque, pronti per vedere qualche esempio di trasformazione e qui diventerà chiaro come usare tutto questo nel contesto reale di un’applicazione Angular.

Precedentemente, parlando di routing abbiamo detto che il Router e la ActivatedRoute espongono un’API a eventi basata su proprietà di tipo Observable; come API alternativa avremmo potuto avere un metodo del tipo router.addEventListener analogo a quello usato per gli eventi del DOM, ma in quel modo questi eventi non sarebbero stati componibili così come lo sono se esposti attraverso un observable.

Prendiamo l’esempio di un componente che logga tutti gli eventi di navigazione e immaginiamo di voler appendere ad ogni evento una chiamata POST verso un’API che registri il comportamento della nostra app per fini statistici o diagnostici:

import { HttpClient } from '@angular/common/http';
import { Component, inject, OnInit } from '@angular/core';
import { Event, Router } from '@angular/router';
import { Observable } from 'rxjs';


@Component({
  selector: 'app-root',
  standalone: true,
  imports: [],
  template: 'Works',
})
export class AppComponent implements OnInit {
  router = inject(Router);
  http = inject(HttpClient);


  ngOnInit(): void {
    this.router.events.subscribe(event => this.trackEvent(event).subscribe()) // <-
  }


  trackEvent(event: Event): Observable<void> {
    return this.http.post<void>('/track/navigation', { event });
  }
}

Il codice in questo esempio è valido e funzionante, tuttavia chiamare più di una subscribe nello stesso stream è considerato una cattiva pratica, poiché c’è un modo più reattivo di fare questa cosa:

import { HttpClient } from '@angular/common/http';
import { Component, inject, OnInit } from '@angular/core';
import { Event, Router } from '@angular/router';
import { Observable, switchMap } from 'rxjs';


@Component({
  selector: 'app-root',
  standalone: true,
  imports: [],
  template: 'Works',
})
export class AppComponent implements OnInit {
  router = inject(Router);
  http = inject(HttpClient);


  ngOnInit(): void {
    this.router.events.pipe(
      switchMap(event => this.trackEvent(event)) // <-
    ).subscribe();
  }


  trackEvent(event: Event): Observable<void> {
    return this.http.post<void>('/track/navigation', { event });
  }
}

Qui abbiamo rimosso la subscribe interna e l’abbiamo sostituita con un pipe(switchMap(…)). Per spiegare il funzionamento di switchMap possiamo usare l’analogia tra Observable e Array, che da un punto di vista matematico hanno in comune veramente tanto.

Prendiamo il seguente esempio:

[1, 2, 3].map(n => [n * 2]).forEach(x => console.log(x)) // [2] [4] [6]
from([1, 2, 3]).pipe(map(x => from([x * 2]))).subscribe(x => console.log(x)); // ?

Per testare queste righe ci sarà sufficiente eseguirle nel costruttore di AppComponent.

Una nota sul metodo pipe: poiché RxJS è un sistema di estensioni, è implementato nella forma di un insieme di operatori da utilizzare in una chiamata al metodo pipe. Questa è semplicemente una sintassi alternativa, più flessibile, all’implementazione di metodi direttamente sul tipo Observable.

Dentro una chiamata a pipe è possibile concatenare più operatori che verranno semplicemente eseguiti in ordine. L’observable risultante dopo una chiamata a pipe è un nuovo observable ottenuto a partire dal precedente, e nessuno dei due observable inizierà a emettere dati finché non verrà chiamata la subscribe su di lui.

Qui abbiamo usato la factory from per estrarre i valori dagli array e farli emettere uno ad uno dall’observable risultante; in questo modo la somiglianza tra i due esempi dovrebbe risultare ancora più lampante.
In entrambi i casi il risultato non è quello che ci si aspetterebbe, perché stiamo mappando da un valore estratto a uno contenuto nel suo wrapper, in un caso array e nell’altro observable.

Se vogliamo appiattire (flatten) l’estrazione dei valori, dobbiamo usare un operatore leggermente diverso in entrambi i casi:

[1, 2, 3].flatMap(n => [n * 2]).forEach(x => console.log(x)); // 2 4 6
from([1, 2, 3]).pipe(switchMap(x => from([x * 2]))).subscribe(x => console.log(x)); // 2 4 6

Notiamo come siamo passati in un caso da map a flatMap, e nell’altro caso a switchMap. In entrambi i casi l’operazione che stiamo eseguendo è: mappa dal valore estratto al valore che si estrae dal wrapper ritornato.

In questo esempio emerge anche l’analogia tra forEach che concretizza la lettura degli elementi dell’array in modo da poter generare un effetto, e subscribe, che ha esattamente lo stesso ruolo. Nel caso delle promise il parallelo diventa strano poiché then funge da map, da flatMap/switchMap e anche da subscribe; come se non bastasse, la promise è precoce (eager), nel senso che si attiva anche se then non viene chiamato affatto.

Torniamo ad Angular e vediamo come possiamo usare RxJS per creare ed esporre la nostra API a eventi, costruendo una service class che fa da storage di una parte dello stato dell’applicazione, in modo che questo possa essere fruito dai componenti dell’intera app in modo reattivo.

Per questo esempio ci sbizzarriamo con una vera e propria web app che implementa una to-do list.

La nostra app mostrerà:

• un box di testo per filtrare le attività,
• la lista delle attività filtrate (o tutte se il filtro è vuoto), e per ognuna: 
  • la descrizione,
  • una checkbox,
  • un pulsante rimuovi;
• un riepilogo con le attività totali, quelle fatte e quelle da fare;
• un pulsante per aggiungere una nuova attività.

Useremo la programmazione reattiva per:

• esporre l’API dello storage in memory,
• replicare lo storage in memory sul localStorage,
• realizzare un filtro della lista.

Cominciamo a vedere che aspetto ha il nostro app component:

import { Component } from '@angular/core';
import { TodoListComponent } from './todo/todo-list/todo-list.component';
import { TodoSummaryComponent } from './todo/todo-summary/todo-summary.component';
import { TodoPersistComponent } from './todo/todo-persist/todo-persist.component';
import { TodoFindComponent } from './todo/todo-find/todo-find.component';


@Component({
  selector: 'app-root',
  standalone: true,
  imports: [
    TodoPersistComponent,
    TodoFindComponent,
    TodoListComponent,
    TodoSummaryComponent
  ],
  template: `
    <app-todo-persist />
    <app-todo-find />
    <app-todo-list />
    <app-todo-summary />
  `,
})
export class AppComponent { }

Benissimo: ci siamo limitati ad importare e dichiarare nel template 4 componenti.

Questi 4 componenti lavorano tutti su una dipendenza comune: TodoService;
nel suo modulo andiamo a dichiarare anche l’interfaccia dell’entità Todo e una semplice factory:

import { Injectable } from '@angular/core';
import { BehaviorSubject } from 'rxjs';


export interface Todo {
  label: string;
  done: boolean;
}


export const Todo = (label: string = ''): Todo => ({
  label,
  done: false
});


@Injectable({ providedIn: 'root' })
export class TodoService {
  #todos = new BehaviorSubject<Todo[]>([]);
  readonly changes = this.#todos.asObservable();


  get todos() { return this.#todos.value }
  set todos(todos: Todo[]) { this.#todos.next(todos) }


  notifyChanges(): void { this.#todos.next(this.todos) }
}

Qui abbiamo decisamente un po’ di cose da spiegare.

Iniziamo osservando che i nostri todo sono rappresentati da una proprietà privata che è un BehaviorSubject; questo tipo è una classe che estende Subject, che può essere usata sia come un Observable, sia come un Subscriber, cioè vi si possono chiamare sopra il metodo next e il metodo complete.

In aggiunta a Subject, BehaviorSubject prevede un valore iniziale ed espone sempre, anche in modo sincrono (tramite la proprietà value), l’ultimo valore emesso.

Poiché non vogliamo esporre direttamente il BehaviorSubject, esponiamo lo stream delle modifiche alla lista tramite changes, ricavato proprio da #todos. In questo modo abbiamo incapsulato l’API reattiva della nostra collezione.

A questo punto, incapsuliamo anche l’API sincrona con un getter e un setter.
Infine, completiamo l’API reattiva con il metodo notifyChanges, che non fa altro che riemettere l’ultima versione della lista, notificandone la modifica a chi è in ascolto su changes; questo metodo esiste specificamente per consentire la mutazione dell’array o di singoli todo sul posto (in-place) senza cioè dover riassegnare l’intera collezione.

Abbiamo detto che vogliamo anche persistere la nostra collezione replicandola su localStorage; lo facciamo con TodoPersistComponent:

import { afterNextRender, Component, inject, OnDestroy } from '@angular/core';
import { Todo, TodoService } from '../todo.service';
import { debounceTime, Observable, switchMap, tap } from 'rxjs';

@Component({
  selector: 'app-todo-persist',
  standalone: true,
  imports: [],
  template: '',
})
export class TodoPersistComponent implements OnDestroy {
  $ = inject(TodoService);

  #sub = this.#load().pipe(
    tap(todos => { this.$.todos = todos }),
    switchMap(() => this.$.changes),
    debounceTime(200)
  ).subscribe(todos => this.#save(todos));

  ngOnDestroy(): void { this.#sub.unsubscribe() }

  #save(todos: Todo[]): void {
    localStorage.setItem('todos', JSON.stringify(todos));
  }


  #load(): Observable<Todo[]> {
    return new Observable(subscriber => {
      afterNextRender(() => {
        const todos = localStorage.getItem('todos') ?? '[]';

        subscriber.next(JSON.parse(todos));
        subscriber.complete();
      })
    });
  }
}

Anche qui, un po’ di spiegazioni, a partire dalla più necessaria:
perché abbiamo implementato una funzionalità che non ha niente a che vedere con la UI in un componente con template vuoto?

La ragione è che questo è a tutti gli effetti il modo più pulito ed elegante per farlo: così facendo assumiamo il controllo sul ciclo di vita del componente attraverso il template del componente padre; possiamo, ad esempio, attivare e disattivare la feature in modo semplice e dichiarativo: @if (persist) { <app-todo-persist /> }.

Inoltre, normalmente i servizi servono a esporre dati e operazioni ai componenti che ne fruiscono, ma a noi qui non serve fruire di alcunché: vogliamo solo attivare un agente che poi opera in modo autonomo in reazione a degli eventi, e un componente è il posto ideale in cui fare questo tipo di cosa.

Come regola generale rispetto a RxJS, un servizio non dovrebbe fare sottoscrizioni a observable; quello è compito dei componenti, se possibile tramite la AsyncPipe. In questo caso non è possibile poiché non stiamo presentando dati ma agganciando un’operazione ad un evento.

Andando avanti, abbiamo iniettato TodoService nella proprietà $; la scelta di questo nome è intenzionale: vogliamo far risultare estremamente sintetico l’accesso agli attributi di TodoService, quasi come se li stessimo ereditando.

La proprietà #sub è molto interessante, e viene inizializzata appena costruiamo il componente:

• recupera i todo salvati a localStorage da #load sotto forma di observable (vedremo a breve perché questo è necessario);
• con tap intercetta il contenuto e inizializza la collezione in memory a bordo di TodoService che verrà esposta a tutti gli altri componenti;
• con switchMap, passa in ascolto sui changes, che a questo punto sono inizializzati con la collezione salvata;
• con debounceTime attende 200 millisecondi senza modifiche prima di proseguire;
• con subscribe applica l’azione desiderata, cioè ad ogni modifica chiama #save.

Il risultato della subscribe viene salvato in #sub, in modo che possiamo poi chiudere la sottoscrizione quando il componente viene distrutto.

Il metodo #save si spiega da solo, mentre invece su #load ci sono un po’ di cose da dire. Il motivo per cui questo metodo restituisce un Observable è lo stesso per cui il componente inizializza tutto in costruzione anziché usando ngOnInit: poiché Angular fa Server Side Rendering (SSR), non è sempre possibile fare operazioni su localStorage perché il componente viene inizializzato a backend; dunque, dobbiamo chiamare afterNextRender per assicurarci che il caricamento dei todo memorizzati avvenga lato client, e afterNextRender deve necessariamente essere chiamata nel costruttore del componente. Per ovviare a questo inconveniente, costruiamo un observable e lo usiamo come una promise.

Se anziché usare localStorage avessimo una REST API a back end, avremmo usato HttpClient.get, ritrovandoci con un observable del tutto simile a questo.

Andiamo ora a implementare TodoFindService:

import { inject, Injectable } from '@angular/core';
import { TodoService } from '../todo.service';
import { BehaviorSubject, combineLatest, map } from 'rxjs';


@Injectable({ providedIn: 'root' })
export class TodoFindService {
  $ = inject(TodoService);
  #filter = new BehaviorSubject('');


  readonly found = combineLatest([
    this.$.changes,
    this.#filter
  ]).pipe(
    map(([todos, filter]) => (
      todos.filter(t => t.label.includes(filter))
    ))
  );


  update(filter: string): void { this.#filter.next(filter) }
}

Anche in questo caso abbiamo TodoService come dipendenza.

Rappresentiamo ancora il nostro filtro come un BehaviorSubject e ricaviamo i todo filtrati (found) dai valori combinati di $.changes e #filter, usando combineLatest: questa factory crea un Observable che emette tuple di valori emessi da un elenco di stream; ogni volta che uno stream sorgente emette un valore, viene emessa una tupla contenente il valore nuovo insieme all’ultimo valore (da cui latest) emesso da tutti gli altri stream usati come sorgenti.

Quindi ogni volta che cambia la lista dei todo oppure il filtro, andiamo ad aggiornare la lista dei todo trovati.

Per finire, incapsuliamo l’aggiornamento del nostro filtro con il metodo update.

Il servizio appena creato ci serve per implementare TodoFindComponent:

import { Component, inject } from '@angular/core';
import { TodoFindService } from './todo-find.service';


@Component({
  selector: 'app-todo-find',
  standalone: true,
  imports: [],
  template: `
    <div style="margin: 8px 0;">
      <input type="text" placeholder="search..." (input)="update($event)">
    </div>
  `
})
export class TodoFindComponent {
  $ = inject(TodoFindService);


  update(event: Event): void {
    const target = event.currentTarget as HTMLInputElement;
    this.$.update(target.value);
  }
}

Questo componente è molto semplice e fa solo due cose:

• presenta il box in cui digitare il filtro,
• passa al servizio il filtro aggiornato.

Sarebbe divertente implementare un componente TodoFindPersistComponent che funziona come TodoPersistComponent ma opera sulla query string dell’URL anziché sul local storage. L’implementazione sarebbe analoga, perciò la lasciamo come esercizio.

Ora che abbiamo tutto il necessario per salvare nello storage, persistere e filtrare le nostre attività, vediamo di presentarle in una semplice lista;

se ne occupa TodoListComponent:

import { Component, inject } from '@angular/core';
import { Todo, TodoService } from '../todo.service';
import { TodoFindService } from '../todo-find/todo-find.service';
import { map } from 'rxjs';
import { AsyncPipe } from '@angular/common';

@Component({
  selector: 'app-todo-list',
  standalone: true,
  imports: [AsyncPipe],
  templateUrl: './todo-list.component.html'
})
export class TodoListComponent {
  $t = inject(TodoService);
  $f = inject(TodoFindService);

  add(): void {
    this.$t.todos = this.$t.todos.concat(Todo());
  }

  remove(todo: Todo): void {
    this.$t.todos = this.$t.todos.filter(t => t !== todo);
  }

  updateDone(todo: Todo, event: Event): void {
    const target = event.currentTarget as HTMLInputElement;
    todo.done = target.checked;

    this.$t.notifyChanges();
  }

  updateLabel(todo: Todo, event: Event): void {
    const target = event.currentTarget as HTMLInputElement;
    todo.label = target.value;

    this.$t.notifyChanges();
  }
}

Se abbiamo accolto la convenzione $ per le dipendenze singole, la presenza di $t e $f ci dice a colpo d’occhio che questo componente opera sia sulla lista totale, sia su quella filtrata. Qui è dove si compie il principio composition over inheritance.

Questo componente, infatti, integra una serie di responsabilità: presenta la lista filtrata ma opera su quella completa, e ne muta anche gli elementi individualmente.

Vediamo il suo template:

@for (t of $f.found | async; track $index) {
  <div style="margin: 8px 0; display: flex; gap: 4px;">
    <input
      type="checkbox"
      [checked]="t.done"
      (input)="updateDone(t, $event)"
    >
    <input
      type="text"
      placeholder="add label"
      [value]="t.label"
      (input)="updateLabel(t, $event)"
    >
    <button (click)="remove(t)">Remove</button>
  </div>
}

<button (click)="add()">Add</button>

Qui non c’è molto da dire, notiamo solo l’uso di un blocco @for insieme alla async pipe per presentare la lista filtrata; il resto è puro data/event binding.

Il nostro viaggio si conclude con l’implementazione del SummaryComponent:

import { Component, inject } from '@angular/core';
import { Todo, TodoService } from '../todo.service';
import { combineLatest, map } from 'rxjs';
import { AsyncPipe, JsonPipe } from '@angular/common';
import { TodoFindService } from '../todo-find/todo-find.service';


interface Summary {
  total: number;
  todo: number;
  done: number;
}


const Summary = (todos: Todo[]): Summary => ({
  total: todos.length,
  todo: todos.filter(t => !t.done).length,
  done: todos.filter(t => t.done).length
});


@Component({
  selector: 'app-todo-summary',
  standalone: true,
  imports: [AsyncPipe, JsonPipe],
  template: '<pre>{{summary | async | json}}</pre>'
})
export class TodoSummaryComponent {
  $t = inject(TodoService);
  $f = inject(TodoFindService);


  summary = combineLatest([
    this.$t.changes,
    this.$f.found
  ]).pipe(
    map(([todos, found]) => ({
      todos: Summary(todos),
      found: Summary(found)
    }))
  );
}

Anche qui componiamo la lista filtrata e quella completa in un oggetto Summary che rappresenta un riepilogo delle attività; questo riepilogo viene poi mostrato come JSON, ma potrebbe anche essere rappresentato con degli elementi grafici.

Al termine di questa sezione abbiamo verosimilmente collezionato un buon quantitativo di sgomento e confusione. La programmazione reattiva richiede un cambio di prospettiva per essere compresa a fondo; quando la si comprende, bisogna fare attenzione a non diventarne fanatici e cadere nel tranello di interpretare tutto come un enigma stimolante da risolvere: come ogni paradigma di programmazione, la programmazione reattiva è qui per semplificarci la vita, non per complicarcela.

Possiamo usare RxJS con successo ogni volta che abbiamo una o più fonti di dati soggette ad aggiornamenti (in questo caso le attività e il filtro) da combinare per ricavare dati derivati (come il nostro summary) oppure a cui agganciare reazioni (come la replica sul localStorage), col beneficio di poterne regolare molto comodamente il tempismo (usando gli operatori di RxJS come debounceTime). Per le normali logiche di flusso, la vecchia cara programmazione imperativa non passerà mai di moda, ed è giusto così.