Combiner Observables : combineLatest, forkJoin

Introduction

En Angular et RxJS, vous devez souvent combiner plusieurs Observables : filtres qui changent, chargements parallèles, événements multiples. RxJS offre plusieurs stratégies puissantes selon votre besoin.

combineLatest() — Émettre à chaque changement

combineLatest() combine les valeurs les plus récentes de chaque Observable. Il émet chaque fois que n'importe quel Observable émet une nouvelle valeur :

// Exemple : Filtrer les produits avec 3 critères indépendants
import { combineLatest } from 'rxjs';

@Component({...})
export class ProductListComponent {
    products: Product[] = [];

    constructor(private filterService: FilterService, private api: ApiService) {
        combineLatest([
            this.filterService.category$,  // Catégorie sélectionnée
            this.filterService.search$,     // Texte de recherche
            this.filterService.page$        // Page actuelle
        ]).subscribe(([category, search, page]) => {
            // Appelé chaque fois qu'UN des trois Observables change
            this.loadProducts(category, search, page);
        });
    }

    private loadProducts(category: string, search: string, page: number) {
        this.api.getProducts(category, search, page)
            .subscribe(products => this.products = products);
    }
}

forkJoin() — Attendre la fin de tous

forkJoin() attend que TOUS les Observables se terminent, puis émet une seule fois avec les dernières valeurs :

// Exemple : Charger tous les données au démarrage d'une page
import { forkJoin } from 'rxjs';

@Component({...})
export class DashboardComponent implements OnInit {
    users: User[] = [];
    posts: Post[] = [];
    comments: Comment[] = [];
    isLoading = true;

    constructor(private api: ApiService) {}

    ngOnInit() {
        forkJoin({
            users: this.api.getUsers(),
            posts: this.api.getPosts(),
            comments: this.api.getComments()
        }).subscribe({
            next: ({ users, posts, comments }) => {
                // Tous les appels sont terminés
                this.users = users;
                this.posts = posts;
                this.comments = comments;
                this.isLoading = false;
            },
            error: (err) => {
                console.error('Une requête a échoué :', err);
                this.isLoading = false;
            }
        });
    }
}

merge() — Fusionner les émissions

merge() combine les émissions de plusieurs Observables sans attendre. Chaque émission passe au travers :

// Exemple : Déclencher un refresh depuis plusieurs sources
import { merge } from 'rxjs';
import { switchMap, startWith } from 'rxjs/operators';

@Component({...})
export class DataComponent {
    data: any;

    constructor(private api: ApiService, private route: ActivatedRoute) {
        // Refresh au démarrage, quand le bouton est cliqué, ou quand les params changent
        merge(
            of(null),                                    // Émettre immédiatement
            this.refreshButton.click$,                  // Clic sur le bouton refresh
            this.route.params.pipe(skip(1)),           // Route change (skip le premier)
            this.autoRefresh$.pipe(throttleTime(5000)) // Auto-refresh tous les 5s
        ).pipe(
            switchMap(() => {
                console.log('Refresh déclenché');
                return this.api.getData();
            })
        ).subscribe(data => {
            this.data = data;
        });
    }
}

Tableau comparatif

Conclusion

Combiner les Observables est un art. Demandez-vous :