Angular 43 🅰️ Combining Streams — merge, concat, combineLatest
A single stream is a sequence of values over time. Real applications rarely have just one. A component needs the current user, the current route, the current filter, and the current data, and it needs to react when any of them changes. The combining operators are how multiple streams are brought together into one. They fall into two families: the flattening combiners — merge, concat, race — which interleave or sequence whole streams, and the combining operators — combineLatest, zip, withLatestFrom, forkJoin — which combine the latest values from several streams into a single value. The choice depends on whether the streams are peers that should be interleaved, or inputs that should be combined into a snapshot. This chapter covers each operator, the mental model that makes it predictable, and the Angular scenarios where each is the right choice.
Key point: merge interleaves streams as their values arrive, and concat runs them one after another. combineLatest emits a combined value whenever any source emits, using the latest value from each — it waits for all sources to emit at least once. zip pairs values by index, emitting only when every source has produced the next value. withLatestFrom lets one source drive the emission and reads the latest value from the others as secondary inputs. forkJoin waits for all sources to complete and emits a single array of their last values, which makes it the RxJS equivalent of Promise.all. The choice depends on the emission semantics the consumer needs.
Why combining streams matters
A component’s view model is almost never a single stream. It is the current user, the current items, the current filter, the loading state — each a stream, and the view is a function of all of them. The combining operators are how the view model is built.
The two families. The flattening combiners treat whole streams as the unit: merge runs them concurrently and interleaves their values, concat runs them one after another, race picks the first to emit and ignores the rest. The combining operators treat the streams as inputs to a snapshot: combineLatest produces a tuple of the latest values, zip pairs by index, withLatestFrom reads the secondaries when the primary emits, and forkJoin collects the last values after all complete.
Why the distinction matters. merge of two streams produces a stream of individual values from either — the identity of each value’s source is lost unless the values carry it. combineLatest of the same two streams produces a stream of tuples — each emission is a snapshot of both. The consumer’s expectation determines which is correct.
Why the operators compose. A view model is often built from several combining operators: combineLatest to combine the user and the filter, switchMap to fetch the data when either changes, forkJoin to load several independent pieces, merge to combine events. The operators are the vocabulary, and the view model is the composition.
Why the subscription model is important. Each combining operator subscribes to its sources when it is subscribed to. A combineLatest of three streams subscribes to all three when the outer subscription starts, and unsubscribes from all three when it ends. The subscription is the lifecycle, and the combining operator is the fan-in point.
Why
combineLatestis the most used combining operator. It matches the common case: a component depends on several inputs, and the view is a function of the latest values of each. Whenever any input changes, the view updates. The operator is the natural expression of that dependency.
merge — interleave
merge subscribes to all source Observables and emits values from each as they arrive. The result is a single stream that interleaves the sources’ values.
import { merge, interval } from 'rxjs';
import { map } from 'rxjs/operators';
const fast$ = interval(100).pipe(map((i) => `fast ${i}`));
const slow$ = interval(300).pipe(map((i) => `slow ${i}`));
merge(fast$, slow$).subscribe((v) => console.log(v));
// fast 0
// fast 1
// fast 2
// slow 0
// fast 3
// ...The values from fast$ and slow$ are interleaved in the order they arrive. Both sources run concurrently, and the merged stream emits whenever either one does.
Why the order is not guaranteed. The sources emit at their own pace, so the merged stream’s order depends on timing. This is correct for events that are independent — clicks and keystrokes, messages from different channels — and wrong when the order matters.
Why merge is useful for events. A component that reacts to several event sources — a form change, a button click, a WebSocket message — can merge them into a single stream and handle each event with the same logic. The identity of the source is carried in the value if it is needed.
The concurrency argument. merge accepts a concurrency argument that limits the number of subscriptions.
merge(source$, 3);The limit is the same as the one on mergeMap — at most three inner subscriptions at once. The rest are queued. This is used to limit the resource consumption of the merged stream.
Why merge is the simplest combiner. It has no notion of pairing or ordering. It is the union of the streams’ values, and the consumer handles each value as it arrives. It is the right tool when the sources are independent and the consumer does not need to combine their values.
concat — sequence
concat subscribes to the first source and emits its values until it completes, then subscribes to the next, and so on. The sources run one after another, and the order is preserved.
import { concat, of } from 'rxjs';
const first$ = of(1, 2, 3);
const second$ = of(4, 5, 6);
concat(first$, second$).subscribe((v) => console.log(v));
// 1
// 2
// 3
// 4
// 5
// 6The values from first$ are all emitted before any from second$. The sequence is deterministic: the first source’s values, then the second’s, in order.
Why the sequence matters. Operations that must run in order — an animation then a navigation, a load then a save, a setup then a request — need the sequence. concat guarantees that the second source is not subscribed until the first completes.
The difference from concatMap. concatMap maps each outer value to an inner Observable and runs them in order. concat takes a fixed sequence of Observables and runs them in order. The two are related: concat is the static version, concatMap is the dynamic version where the inner streams depend on the outer values.
Why concat is rarely the right tool for HTTP. A set of HTTP requests that should run in order is usually expressed as a concatMap over a list, because the list is a stream. concat is used when the sequence is known statically — a fixed set of streams that run in a known order.
Why concat is used for animations. An animation is a stream that emits over time and completes. A sequence of animations is a concat of them, and the navigation happens after the last completes. This is a common pattern in Angular’s animation callbacks.
Why concat waits for completion. The operator does not move to the next source until the current one completes. An infinite stream — an interval, a WebSocket — blocks the rest of the sequence forever. This is a design property, not a bug, and it means concat is for finite streams.
combineLatest — the snapshot
combineLatest subscribes to all sources, waits for each to emit at least once, and then emits a tuple of the latest values whenever any source emits.
import { combineLatest, BehaviorSubject } from 'rxjs';
const user$ = new BehaviorSubject<User | null>(null);
const items$ = new BehaviorSubject<Item[]>([]);
const filter$ = new BehaviorSubject<string>('');
combineLatest([user$, items$, filter$]).subscribe(([user, items, filter]) => {
console.log(user, items.length, filter);
});
// after all emit at least once:
// [user, items, filter]
// [newUser, items, filter] when user changes
// [user, newItems, filter] when items changeThe combined stream emits whenever any of the three emits, using the latest value from each. The first emission happens only after all three have emitted at least once.
Why the “all emit once” rule is important. Before any of the sources emits, there is no value for that source, so there is no complete tuple. The operator waits. Once all have emitted, every subsequent emission from any source produces a new tuple.
Why the initial value matters. A BehaviorSubject emits its initial value immediately, so a combineLatest of BehaviorSubjects emits as soon as all are subscribed. A plain Subject that has not emitted blocks the combined stream — the operator waits for it. This is a common source of “why does my combined stream not emit?” — one of the sources has not emitted.
Why the tuple is destructured. The array form combineLatest([a$, b$, c$]) is destructured in the subscriber. The array form is preferred over the deprecated variadic form combineLatest(a$, b$, c$) because the array is easier to read and the types are clearer.
Why a projection function helps. The operator accepts a projection function that receives the values and returns the combined value.
combineLatest([user$, items$, filter$]).pipe(
map(([user, items, filter]) => ({ user, items: filterItems(items, filter) })),
);The projection is where the view model is built. The combineLatest provides the inputs, and the map produces the shape the template needs.
Why combineLatest is used for view models. A component’s view model is a function of its inputs. The combineLatest of the inputs, followed by a map to the view model, is the pattern. The template subscribes to the view model stream with the async pipe, and the view updates whenever any input changes.
Why
combineLatestshould not be used for triggering side effects. The operator emits on every change from any source. If the projection function triggers a side effect — an HTTP call, a save — the side effect fires on every change, which is often not what is wanted. For side effects,switchMapafter thecombineLatestis the pattern: the combined values drive the request, and the request is canceled when the inputs change.
zip — pair by index
zip combines the sources by index. It emits a tuple when every source has produced its next value, pairing them by position.
import { zip, of } from 'rxjs';
const names$ = of('Alice', 'Bob', 'Carol');
const ages$ = of(30, 25, 35);
zip(names$, ages$).subscribe(([name, age]) => console.log(name, age));
// Alice 30
// Bob 25
// Carol 35The first values are paired, then the second, and so on. The operator emits when every source has produced its next value, and it buffers the values from faster sources until the slower ones catch up.
Why zip is rarely used. The pairing-by-index semantics is unusual. It is correct when the sources are naturally parallel — the second value from one source is meant to go with the second value from another — but this is uncommon in practice.
Why buffering matters. A faster source’s values accumulate in a buffer until the slower source produces its matching value. For an infinite source with a mismatched rate, the buffer grows unbounded, which is a memory concern. This is why zip is not the tool for combining a fast stream with a slow one.
Why zip is used for ordered pairs. A common case is a stream of requests and a stream of responses where the pairing is by order. The zip pairs them correctly, and the buffer holds the values until the match is found. For most Angular code, combineLatest or withLatestFrom is the better tool.
Why the deprecation of the result-selector form matters. zip accepted a result-selector function as its last argument, which projected the tuple. The form is deprecated in favor of zip(...).pipe(map(...)). The same deprecation applies to combineLatest and forkJoin. The map form is clearer and composes with the rest of the pipeline.
withLatestFrom — the driver and the passengers
withLatestFrom lets one source drive the emissions and reads the latest value from the others as secondary inputs. The combined stream emits only when the primary source emits, and only if the secondaries have emitted at least once.
import { fromEvent } from 'rxjs';
import { withLatestFrom, map } from 'rxjs/operators';
const clicks$ = fromEvent(document, 'click');
const user$ = this.auth.currentUser$;
const filter$ = this.filter$;
clicks$.pipe(
withLatestFrom(user$, filter$),
map(([click, user, filter]) => ({
x: (click as MouseEvent).clientX,
userId: user?.id,
filter,
})),
).subscribe((event) => console.log(event));The click drives the emission. When the user clicks, the current user and filter are read and combined into the event. Clicks before the user and filter emit are dropped, because the secondaries have not produced a value yet.
Why the primary is the trigger. The semantics is “when the primary emits, snapshot the latest of the others.” This matches the common case: an event happens, and the current state is needed to handle it. The event is the trigger, and the state is the context.
Why withLatestFrom is different from combineLatest. combineLatest emits when any source emits. withLatestFrom emits only when the primary emits. The direction of the trigger is the difference, and it changes the behavior.
Why the secondaries must have emitted. The operator waits for each secondary to emit at least once. Until then, the primary’s emissions are dropped. This is the same “all emit once” rule as combineLatest, applied to the secondaries.
Why withLatestFrom is used for event handlers. A click handler needs the current state, not a reaction to the state’s changes. The click is the trigger, and the state is read at the moment of the click. The withLatestFrom is the exact expression of that.
Why the operator is used after a user action. Any action that needs the current state — a save button, a form submit, a navigation — is a candidate for withLatestFrom. The action is the primary, and the state is the secondary. The operator reads the state at the moment of the action.
forkJoin — the Promise.all
forkJoin subscribes to all sources, waits for each to complete, and emits a single array of their last values. It is the RxJS equivalent of Promise.all.
import { forkJoin, of } from 'rxjs';
import { delay } from 'rxjs/operators';
forkJoin([
this.http.get<User>('/api/user'),
this.http.get<Item[]>('/api/items'),
this.http.get<Settings>('/api/settings'),
]).subscribe(([user, items, settings]) => {
console.log(user, items, settings);
});The three requests run concurrently. When all three complete, the combined stream emits a single array with their results. The subscription completes after the emission.
Why forkJoin is for concurrent loads. The operator is the right tool for loading several independent resources and producing a combined result. The requests run in parallel, and the result is available when all complete.
Why the sources must complete. forkJoin waits for completion, not for emission. An infinite stream never completes, so forkJoin never emits. This is why the operator is used with HTTP requests (which complete) and not with event streams (which do not).
Why an empty source completes immediately. If the array is empty, forkJoin([]) completes immediately without emitting. If any source completes without emitting, the result is the same — the operator completes without emitting. This is a subtle behavior that surprises developers coming from Promise.all, which resolves with an empty array.
Why forkJoin replaces combineLatest for one-shot loads. combineLatest emits whenever any source emits, which means it can emit multiple times. forkJoin emits once, when all are done. For a one-shot load, the single emission is correct, and the operator completes, which releases the subscription.
Why error handling is all-or-nothing. If any source errors, the combined stream errors. The other sources are not canceled, but their results are discarded. This is the Promise.all behavior, and it means forkJoin is for the case where all the results are needed together.
Why forkJoin is used for parallel initialization. Loading a component’s initial data — the user, the settings, the permissions — is a forkJoin. The requests run in parallel, and the component renders when all are done. The pattern is common in route resolvers and in component initialization.
Why the completion requirement is the main gotcha. A
forkJoinof HTTP requests works because HTTP completes. AforkJoinof aBehaviorSubjectnever emits because the Subject never completes. The operator is for finite streams, and the consumer must know that.
Choosing the right operator
The combining operators are distinguished by their emission semantics. The choice depends on what the consumer needs.
| Operator | Emits when | Uses | Completes |
|---|---|---|---|
merge | Any source emits | Each value | When all complete |
concat | Sequentially | Each value in order | When all complete |
combineLatest | Any source emits (after all have) | Tuple of latest | When all complete |
zip | All sources emit the next value | Tuple by index | When any completes |
withLatestFrom | Primary emits (after all have) | Tuple of latest | When primary completes |
forkJoin | All complete | Array of last values | After the single emission |
Why the table is the chapter. The table is the answer to the recurring question of which operator to use. The rest of the chapter is the reasoning behind each row. A reader who knows the table and the semantics behind each row will make the right choice in most cases.
Why the wrong choice is subtle. A combineLatest used where forkJoin is needed emits multiple times instead of once. A withLatestFrom used where combineLatest is needed drops emissions that do not come from the primary. A merge used where combineLatest is needed loses the pairing. The bugs are not in the common path; they are in the edges.
Why the operators compose. A real view model often combines several operators: a combineLatest of the inputs, a switchMap to fetch, a forkJoin to load the independent pieces, and a map to produce the shape. The composition is the view model, and the choice of each operator is the design decision.
Why the subscription lifecycle matters. Each combining operator subscribes to its sources when it is subscribed to and unsubscribes when it is unsubscribed. The takeUntilDestroyed from Angular 39 is the standard cleanup, and it applies to the combined stream as a whole.
Why the initial values matter. combineLatest and withLatestFrom wait for all sources to emit. A source that is a plain Subject and has not emitted blocks the combined stream. Using BehaviorSubject with an initial value is the standard fix, and it ensures the combined stream emits promptly.
Complete Example Session
import { Component, DestroyRef, inject, signal } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { FormControl, ReactiveFormsModule } from '@angular/forms';
import { BehaviorSubject, Subject, merge, concat, combineLatest, zip, forkJoin, of, fromEvent } from 'rxjs';
import {
map, switchMap, withLatestFrom, startWith, debounceTime, distinctUntilChanged,
catchError, takeUntilDestroyed, scan, shareReplay,
} from 'rxjs/operators';
// ============================================
// PART 1: MERGE — EVENT SOURCES
// ============================================
const clicks$ = fromEvent(document, 'click');
const keys$ = fromEvent<KeyboardEvent>(document, 'keydown');
const anyInput$ = merge(
clicks$.pipe(map(() => 'click')),
keys$.pipe(map((e) => `key: ${e.key}`)),
);
anyInput$.subscribe((v) => console.log(v));
// ============================================
// PART 2: CONCAT — SEQUENCE
// ============================================
const step1$ = of('init');
const step2$ = of('load');
const step3$ = of('ready');
concat(step1$, step2$, step3$).subscribe((v) => console.log(v));
// init
// load
// ready
// ============================================
// PART 3: COMBINELATEST — VIEW MODEL
// ============================================
@Component({
selector: 'app-view',
standalone: true,
imports: [ReactiveFormsModule],
template: `
@if (vm$ | async; as vm) {
<div>{{ vm.count }} items for {{ vm.user }}</div>
}
`,
})
export class ViewComponent {
private readonly user$ = new BehaviorSubject('alice');
private readonly items$ = new BehaviorSubject<Item[]>([]);
private readonly filter$ = new FormControl('', { nonNullable: true });
readonly vm$ = combineLatest([
this.user$,
this.items$,
this.filter$.valueChanges.pipe(startWith('')),
]).pipe(
map(([user, items, filter]) => ({
user,
count: items.filter((i) => i.name.includes(filter)).length,
})),
);
}
// ============================================
// PART 4: ZIP — PAIRING
// ============================================
const names$ = of('Alice', 'Bob', 'Carol');
const ages$ = of(30, 25, 35);
zip(names$, ages$).pipe(
map(([name, age]) => ({ name, age })),
).subscribe((pair) => console.log(pair));
// ============================================
// PART 5: WITHLATESTFROM — ACTION WITH STATE
// ============================================
@Component({
selector: 'app-save',
standalone: true,
template: `<button (click)="onSave()">Save</button>`,
})
export class SaveComponent {
private readonly http = inject(HttpClient);
private readonly destroyRef = inject(DestroyRef);
private readonly saveClicks$ = new Subject<void>();
private readonly currentUser$ = new BehaviorSubject<User | null>(null);
private readonly formValue$ = new BehaviorSubject<FormValue | null>(null);
constructor() {
this.saveClicks$.pipe(
withLatestFrom(this.currentUser$, this.formValue$),
switchMap(([_, user, value]) =>
this.http.post('/api/save', { userId: user?.id, value }),
),
takeUntilDestroyed(this.destroyRef),
).subscribe();
}
onSave(): void {
this.saveClicks$.next();
}
}
// ============================================
// PART 6: FORKJOIN — PARALLEL LOAD
// ============================================
@Injectable({ providedIn: 'root' })
export class InitService {
private readonly http = inject(HttpClient);
loadAll() {
return forkJoin({
user: this.http.get<User>('/api/user'),
items: this.http.get<Item[]>('/api/items'),
settings: this.http.get<Settings>('/api/settings'),
});
}
}
// ============================================
// PART 7: COMBINELATEST + SWITCHMAP — SEARCH
// ============================================
@Component({
selector: 'app-search',
standalone: true,
imports: [ReactiveFormsModule],
template: `<input [formControl]="search" />`,
})
export class SearchComponent {
private readonly http = inject(HttpClient);
private readonly destroyRef = inject(DestroyRef);
readonly search = new FormControl('', { nonNullable: true });
private readonly user$ = new BehaviorSubject<User | null>(null);
private readonly category$ = new BehaviorSubject<string>('all');
readonly results$ = combineLatest([
this.search.valueChanges.pipe(
debounceTime(300),
distinctUntilChanged(),
startWith(''),
),
this.category$,
]).pipe(
switchMap(([term, category]) =>
this.http.get<Result[]>(`/api/search?q=${term}&cat=${category}`),
),
catchError(() => of([])),
takeUntilDestroyed(this.destroyRef),
);
}
// ============================================
// PART 8: SCAN WITH COMBINELATEST — STATE
// ============================================
interface State {
items: Item[];
loading: boolean;
}
@Component({ selector: 'app-store', standalone: true, template: `` })
export class StoreComponent {
private readonly http = inject(HttpClient);
private readonly destroyRef = inject(DestroyRef);
private readonly refresh$ = new Subject<void>();
private readonly query$ = new BehaviorSubject<string>('');
readonly state$ = combineLatest([
this.query$,
this.refresh$.pipe(startWith(void 0)),
]).pipe(
switchMap(([query]) =>
this.http.get<Item[]>(`/api/items?q=${query}`).pipe(
map((items) => ({ items, loading: false })),
catchError(() => of({ items: [], loading: false })),
startWith({ items: [] as Item[], loading: true }),
),
),
scan((state, update) => ({ ...state, ...update }), { items: [], loading: false }),
shareReplay({ bufferSize: 1, refCount: true }),
takeUntilDestroyed(this.destroyRef),
);
}
// ============================================
// PART 9: WHAT NOT TO DO
// ============================================
// Don't use combineLatest for a one-shot load
// It emits multiple times; forkJoin emits once.
// Don't use forkJoin with infinite streams
// It never emits because they never complete.
// Don't use zip with mismatched rates
// The faster source buffers unbounded.
// Don't use withLatestFrom when the secondary should trigger
// It only emits when the primary emits.
// Don't forget the startWith for optional inputs
// combineLatest waits for all to emit.
// Don't trigger side effects in the projection
// Every emission fires the side effect.The nine parts cover each operator, the view model, the search, the store, and the anti-patterns.
Quick Reference
Combining Operators
| Operator | Emits when | Result | Completes |
|---|---|---|---|
merge | Any source emits | Each value | When all complete |
concat | Sequentially | Each value in order | When all complete |
combineLatest | Any source emits (after all) | Tuple of latest | When all complete |
zip | All emit the next value | Tuple by index | When any completes |
withLatestFrom | Primary emits (after all) | Tuple of latest | When primary completes |
forkJoin | All complete | Array of last values | After the single emission |
Input Requirements
| Operator | Sources must emit | Sources must complete |
|---|---|---|
merge | No | No |
concat | No | Yes (to advance) |
combineLatest | All, at least once | No |
zip | All, per pair | No |
withLatestFrom | All secondaries, at least once | No |
forkJoin | Yes | Yes |
Common Patterns
| Pattern | Operator |
|---|---|
| View model from inputs | combineLatest + map |
| Search with filters | combineLatest + switchMap |
| Parallel load | forkJoin |
| Action with current state | withLatestFrom |
| Event interleaving | merge |
| Sequential steps | concat |
| State accumulation | combineLatest + scan |
Angular Scenarios
| Scenario | Operator |
|---|---|
| User + items in view | combineLatest |
| Initial data load | forkJoin |
| Save with user context | withLatestFrom |
| Multiple event sources | merge |
| Multi-step flow | concat |
| Search with filter | combineLatest + switchMap |
| Store with refresh | combineLatest + scan |
Best Practices
✅ Do This:
// Use combineLatest for a view model
combineLatest([user$, items$]).pipe(map(([user, items]) => ({ user, items }))) // ✅
// Use forkJoin for a parallel one-shot load
forkJoin({ user: http.get('/api/user'), items: http.get('/api/items') }) // ✅
// Use withLatestFrom for an action with state
saveClicks$.pipe(withLatestFrom(user$, form$)) // ✅
// Use merge for independent event sources
merge(clicks$, keys$) // ✅
// Use concat for sequential steps
concat(step1$, step2$, step3$) // ✅
// Add startWith for optional inputs
input$.pipe(startWith('')) // ✅
// Put switchMap after combineLatest for side effects
combineLatest([...]).pipe(switchMap(([...]) => http.get(...))) // ✅
// Use takeUntilDestroyed for cleanup
.pipe(takeUntilDestroyed(this.destroyRef)) // ✅❌ Don’t Do This:
// Don't use combineLatest for a one-shot load
combineLatest([http.get('/a'), http.get('/b')]) // emits multiple times // ⚠️
// Don't use forkJoin with infinite streams
forkJoin([interval(1000), http.get('/api')]) // never emits // ⚠️
// Don't use zip with mismatched rates
zip(fast$, slow$) // fast$ buffers unbounded // ⚠️
// Don't use withLatestFrom when the secondary should trigger
withLatestFrom(filter$) // filter changes do not emit // ⚠️
// Don't forget startWith for optional inputs
combineLatest([optional$]) // waits for optional$ to emit // ⚠️
// Don't trigger side effects in the projection
combineLatest([...]).pipe(map(([...]) => http.get(...))) // wrong // ⚠️
// Don't forget to clean up
subscribe() // without takeUntilDestroyed // ⚠️Common Pitfalls
| Pitfall | Problem | Solution |
|---|---|---|
combineLatest for one-shot | Multiple emissions | forkJoin |
forkJoin with infinite | Never emits | Finite streams only |
zip with mismatched rates | Unbounded buffer | Use combineLatest |
withLatestFrom as combiner | Secondary ignored | Use combineLatest |
Missing startWith | No emission | Add startWith |
Side effect in map | Fires on every change | switchMap after |
| Missing cleanup | Leak | takeUntilDestroyed |
| Wrong operator for the semantics | Subtle bug | Check the table |
Real-World Examples
1. View model
combineLatest([user$, items$]).pipe(map(([user, items]) => ({ user, items })))2. Parallel load
forkJoin({ user: http.get('/api/user'), items: http.get('/api/items') })3. Action with state
save$.pipe(withLatestFrom(user$, form$))4. Multiple events
merge(clicks$, keys$)5. Sequential steps
concat(load$, process$, save$)6. Search with filter
combineLatest([search$, category$]).pipe(switchMap(([q, c]) => http.get(...)))7. Initial values
input$.pipe(startWith(''))8. Store with scan
combineLatest([query$, refresh$]).pipe(switchMap(...), scan(...), shareReplay(1))9. Pair by index
zip(names$, ages$)10. Full view model
combineLatest([user$, items$, filter$]).pipe(
map(([user, items, filter]) => ({ user, items: filterItems(items, filter) })),
takeUntilDestroyed(this.destroyRef),
)Visual: The Two Families
┌──────────────────────────────────────────────────────────┐
│ FLATTENING COMBINERS │
│ │
│ merge: A ──► values │
│ B ──► values │
│ ─────────────► interleaved │
│ │
│ concat: A ──► values │
│ then B ──► values │
│ ─────────────► sequential │
│ │
│ race: A or B, whichever emits first │
│ ─────────────► one source wins │
│ │
├──────────────────────────────────────────────────────────┤
│ COMBINING OPERATORS │
│ │
│ combineLatest: A, B ──► [a, b] on every emission │
│ zip: A, B ──► [a0, b0], [a1, b1], ... │
│ withLatestFrom: A ──► [a, b] on A's emission │
│ forkJoin: A, B ──► [a, b] when both complete │
│ │
└──────────────────────────────────────────────────────────┘Visual: combineLatest
┌──────────────────────────────────────────────────────────┐
│ A: ──a1──────a2──────────a3────► │
│ B: ──────b1────────b2────────────► │
│ │
│ combineLatest([A, B]) │
│ waits for both to emit at least once │
│ emits when any emits, using the latest of each │
│ │
│ ──[a1,b1]──[a2,b1]──[a2,b2]──[a3,b2]──► │
│ ▲ ▲ ▲ ▲ │
│ │ │ │ │ │
│ b1 a2 b2 a3 │
│ │
│ Each emission is a snapshot of both. │
│ │
└──────────────────────────────────────────────────────────┘Visual: withLatestFrom
┌──────────────────────────────────────────────────────────┐
│ PRIMARY (clicks): ──c1──────c2──────────c3────► │
│ SECONDARY (user): ──────u1────────u2─────────► │
│ │
│ clicks$.pipe(withLatestFrom(user$)) │
│ emits only when the primary emits │
│ reads the latest of the secondary │
│ │
│ ───────[c1]──[c2,u1]──[c3,u2]──► │
│ ▲ │
│ │ │
│ c1 is dropped because user has not emitted │
│ │
│ The primary drives; the secondary is context. │
│ │
└──────────────────────────────────────────────────────────┘Visual: forkJoin
┌──────────────────────────────────────────────────────────┐
│ A: ──a1──a2──a3──| │
│ B: ──b1──| │
│ C: ──c1──c2──c3──c4──| │
│ │
│ forkJoin([A, B, C]) │
│ waits for ALL to complete │
│ emits the LAST value of each │
│ then completes │
│ │
│ ──────────────────[a3, b1, c4]──| │
│ │
│ One emission, on completion. │
│ │
└──────────────────────────────────────────────────────────┘Visual: Choosing the Operator
┌──────────────────────────────────────────────────────────┐
│ What does the consumer need? │
│ │ │
│ ├── Each value from any source │
│ │ └── merge │
│ │ │
│ ├── Values in sequence │
│ │ └── concat │
│ │ │
│ ├── A snapshot whenever any input changes │
│ │ └── combineLatest │
│ │ │
│ ├── A snapshot when a specific source emits │
│ │ └── withLatestFrom │
│ │ │
│ ├── Pairs by index │
│ │ └── zip │
│ │ │
│ └── One combined result when all complete │
│ └── forkJoin │
│ │
└──────────────────────────────────────────────────────────┘Visual: Search Pipeline
┌──────────────────────────────────────────────────────────┐
│ search.valueChanges │
│ │ debounceTime(300) │
│ │ distinctUntilChanged() │
│ │ startWith('') │
│ ▼ │
│ combineLatest([search$, category$]) │
│ │ emits when search or category changes │
│ ▼ │
│ switchMap(([term, cat]) => http.get(...)) │
│ │ cancels the previous request │
│ ▼ │
│ catchError(() => of([])) │
│ │ recovers from errors │
│ ▼ │
│ takeUntilDestroyed(destroyRef) │
│ │ cleans up on destroy │
│ ▼ │
│ results │
│ │
└──────────────────────────────────────────────────────────┘Summary
| Operator | Family | Emits | Use |
|---|---|---|---|
merge | Flattening | Each value | Independent events |
concat | Flattening | In order | Sequential steps |
combineLatest | Combining | On any change | View model |
zip | Combining | By index | Ordered pairs |
withLatestFrom | Combining | On primary | Action with state |
forkJoin | Combining | On all complete | Parallel load |
Key takeaways:
- Combining operators fall into two families — the flattening combiners that interleave or sequence whole streams, and the combining operators that snapshot the latest values
mergeinterleaves values from independent sources — the order is not guaranteed, which is correct for events and wrong for orderingconcatruns streams one after another — the sequence is deterministic, and each source must complete before the next beginscombineLatestemits a tuple of the latest values whenever any source emits — it waits for all to emit at least once, which is whyBehaviorSubjectandstartWithare commonzippairs values by index — the faster source buffers, which is a memory concern with mismatched rateswithLatestFromlets one source drive and reads the others — the primary’s emissions trigger the combined emission, and the secondaries are contextforkJoinisPromise.all— it waits for all sources to complete and emits once, which makes it right for parallel loads and wrong for infinite streams- The choice depends on the emission semantics — the table of operators and their emission rules is the answer to the recurring question
combineLatestplusswitchMapis the search pattern — the combined inputs drive the request, and the request is canceled when the inputs changetakeUntilDestroyedapplies to the combined stream — the cleanup is at the end of the pipeline, and it releases all the source subscriptions at once
Remember: Combining streams is how a view model is built. merge and concat flatten whole streams; combineLatest, zip, withLatestFrom, and forkJoin combine their latest values. The emission semantics is the design decision, and the table of operators and their rules is the reference. Use combineLatest for a view model, forkJoin for a parallel load, withLatestFrom for an action with state, merge for independent events, and concat for sequential steps. The wrong choice produces a bug that does not throw, which is why the semantics matters.
Stop using slow, ad-bloated tool sites! 🤮
🔎 Search “KandZ Tools” on Google to use many professional utilities for free.
KandZ.me is the ultimate minimalist hub for:
✅ Finance (Mortgage, Interest, Inflation)
✅ Tech (Base64, JSON, Dev Suite, IP)
✅ Health (BMI, BMR, TDEE)
✅ Productivity (Timer, Workspace, QR)
⚡️ Fast & Private
🔒 No data leaves your device
💎 100% Free
🔗 Use it now: https://tools.kandz.me
🔖 Bookmark it—you’ll need it later!