Advanced RxJS Patterns for Angular: Combine, Share and Flatten

You know map, filter, switchMap and debounceTime. Advanced Angular work needs a few more patterns: choosing the right flattening operator, combining sources, sharing results and keeping streams leak-free.

What you will learn

• When to use switchMap, concatMap, mergeMap and exhaustMap
• How to combine streams with combineLatest, forkJoin and withLatestFrom
• How to share work with shareReplay and subjects, and avoid leaks

Choosing a flattening operator

All four start an inner Observable (like an HTTP call) for each outer value. They differ in what happens when a new outer value arrives while an inner one is still running:

• `switchMap`: Behaviour: Cancel the old, start the new; Use for: Search, route params, latest data wins
• `concatMap`: Behaviour: Queue and run one after another; Use for: Saves that must keep order
• `mergeMap`: Behaviour: Run all at the same time; Use for: Independent parallel work
• `exhaustMap`: Behaviour: Ignore new until current finishes; Use for: Submit buttons, prevent double clicks

// src/app/rxjs/operators-demo.ts
import { Subject, of, delay } from 'rxjs';
import { concatMap, exhaustMap, mergeMap, switchMap } from 'rxjs/operators';

const clicks = new Subject<number>();
const fakeSave = (n: number) => of(`saved ${n}`).pipe(delay(500));

// Double-clicking "Save" sends only one request:
clicks.pipe(exhaustMap(fakeSave)).subscribe(console.log);

// Order-sensitive updates:
clicks.pipe(concatMap(fakeSave)).subscribe(console.log);

// Only the latest matters:
clicks.pipe(switchMap(fakeSave)).subscribe(console.log);

// Everything in parallel:
clicks.pipe(mergeMap(fakeSave)).subscribe(console.log);

A rule of thumb: switchMap for reads, exhaustMap or concatMap for writes. Using switchMap for a save can cancel the first save on the second click.

Combining streams

// src/app/rxjs/combine-demo.ts
import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { HttpClient } from '@angular/common/http';
import { combineLatest, forkJoin, map, startWith, Subject, withLatestFrom } from 'rxjs';

interface User { id: number; name: string }
interface Post { id: number; userId: number; title: string }

@Component({
  selector: 'app-combine-demo',
  template: `
    <button type="button" (click)="refresh$.next()">Refresh</button>
    @for (row of rows(); track row.id) {
      <p>{{ row.title }} by {{ row.author }}</p>
    }
  `,
})
export class CombineDemo {
  private readonly http = inject(HttpClient);
  protected readonly refresh$ = new Subject<void>();

  // forkJoin: wait for several requests, emit once when ALL complete.
  private readonly data$ = forkJoin({
    users: this.http.get<User[]>('https://jsonplaceholder.typicode.com/users'),
    posts: this.http.get<Post[]>('https://jsonplaceholder.typicode.com/posts?_limit=5'),
  });

  // combineLatest: emit whenever ANY input changes (after all emitted once).
  private readonly view$ = combineLatest([this.data$, this.refresh$.pipe(startWith(undefined))]).pipe(
    map(([{ users, posts }]) =>
      posts.map((p) => ({ id: p.id, title: p.title, author: users.find((u) => u.id === p.userId)?.name ?? 'Unknown' })),
    ),
  );

  protected readonly rows = toSignal(this.view$, { initialValue: [] });

  // withLatestFrom: take the latest value of another stream when this one emits.
  readonly example$ = this.refresh$.pipe(withLatestFrom(this.data$));
}

• forkJoin is like Promise.all, and works with Observables that complete, such as HTTP.
• combineLatest keeps emitting when any input changes. It never completes if an input does not.
• withLatestFrom reads the latest value from a second stream but only emits when the first one does.

Sharing results

By default each subscriber to an Observable triggers its own request. Use shareReplay to run it once and replay the last value to later subscribers:

// src/app/rxjs/shared-config.ts
import { Injectable, inject } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { Observable, shareReplay } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class SharedConfig {
  private readonly http = inject(HttpClient);

  readonly users$: Observable<unknown[]> = this.http
    .get<unknown[]>('https://jsonplaceholder.typicode.com/users')
    .pipe(shareReplay({ bufferSize: 1, refCount: true }));
}

Without shareReplay, three components subscribing would send three requests. With refCount: true the source is released when everyone unsubscribes, which avoids a long-lived subscription.

Subjects as event buses

• Subject has no initial value, and is good for events such as the refresh$ above.
• BehaviorSubject always holds a current value and replays it. It was the old way to build stores. Today, signals do that job better inside components and services.
• Expose a Subject as an Observable (subject.asObservable()) so outsiders cannot push values.

Avoid memory leaks

A subscription that is never closed keeps running after the component is gone. Fix it with one of these:

// src/app/rxjs/leak-safe.ts
import { Component, DestroyRef, inject } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { interval } from 'rxjs';

@Component({ selector: 'app-leak-safe', template: `<p>Open the console.</p>` })
export class LeakSafe {
  private readonly destroyRef = inject(DestroyRef);

  constructor() {
    // In a constructor or field initializer you can call it with no argument:
    interval(1000).pipe(takeUntilDestroyed()).subscribe((n) => console.log('tick', n));
  }

  later(): void {
    // Outside an injection context pass the DestroyRef explicitly:
    interval(1000).pipe(takeUntilDestroyed(this.destroyRef)).subscribe();
  }
}

Better still, avoid manual subscribe in components: use toSignal() or the async pipe, which clean up automatically. HTTP calls complete by themselves, so they do not leak, but long-lived streams (interval, fromEvent, store streams) must be closed.

Try it yourself

1. Add a save button with exhaustMap and press it rapidly. How many requests are sent?
2. Replace it with switchMap. What changes?
3. Make a service stream shared with shareReplay and count requests in the Network tab.

Quick check

1. Which operator prevents double submits?
2. What is the difference between combineLatest and forkJoin?
3. How can you stop a long-lived subscription when a component is destroyed?

Next, take control of navigation in advanced routing. See the RxJS documentation for the full operator list.