Combining Streams in RxJS: combineLatest, withLatestFrom, forkJoin, zip, merge and concat

Version note: The examples in this post were run with RxJS 7.8.2, TypeScript 6.0 and Node.js 24; the time-based examples ran in virtual time with RxJS's TestScheduler.

Predict First

The catalog page of Bookstore, the online bookstore used as the shared example across this site's series, has three filters: category, sort order and price range. We turn all three into a single query with combineLatest. At 100 ms the user picks "classics" as the category, at 300 ms "price-asc" as the sort order; they never touch the price range slider:

import { combineLatest, NEVER, timer, type Observable } from "rxjs";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";
import type { PriceRange } from "./catalog-api.ts";

runInVirtualTime("catalog filters", () => {
  const category$ = emitAt([100, "classics"]);
  const sort$ = emitAt([300, "price-asc"]);
  const priceRange$: Observable<PriceRange> = NEVER; // the user never touches the price slider

  combineLatest({ category: category$, sort: sort$, priceRange: priceRange$ }).subscribe((filters) =>
    log(`query: ${JSON.stringify(filters)}`),
  );
  timer(1000).subscribe(() => log("one second later"));
});

How many "query" lines do you see within one second? One, two, or none?

In the previous post we produced a new stream from every value of a single stream. In this post we combine streams that are independent of each other: filters, buttons, several API calls. RxJS has seven core tools for this: combineLatest, withLatestFrom, forkJoin, zip, merge, concat and race. They all look like "combine two streams", but each answers a different question. The examples use the virtual-time helpers (runInVirtualTime, emitAt, log) and the fake HTTP call (fakeRequest) from the previous post.

Every Combinator Answers One Question

Seeing the marbles side by side with the same two sources is the quickest way to explain the difference. The first source emits letters (for example, category changes), the second digits (for example, page changes); values in the output such as a1 are the values of the two sources that were paired at that moment:

import { combineLatest, concat, map, merge, race, withLatestFrom, zip } from "rxjs";
import { showMarbles } from "../../tools/marble.ts";

// two sources: letters (e.g. category changes) and digits (e.g. page changes)
const letters = "-a----b---c--|";
const digits = "---1-----2----|";
const pair = ([x, y]: [string, string]) => x + y;

showMarbles("combineLatest", (cold) => combineLatest([cold<string>(letters), cold<string>(digits)]).pipe(map(pair)));
showMarbles("withLatestFrom", (cold) => cold<string>(letters).pipe(withLatestFrom(cold<string>(digits)), map(pair)));
showMarbles("zip", (cold) => zip([cold<string>(letters), cold<string>(digits)]).pipe(map(pair)));
showMarbles("merge", (cold) => merge(cold<string>(letters), cold<string>(digits)));
showMarbles("concat", (cold) => concat(cold<string>("-a--b|"), cold<string>("-1--2|")));
showMarbles("race", (cold) => race(cold<string>("---a--b|"), cold<string>("-1--2---|")));

combineLatest
  in : -a----b---c--|   
  in : ---1-----2----|  
  out: ---a--b--cd---|  a=a1 b=b1 c=b2 d=c2
withLatestFrom
  in : -a----b---c--|   
  in : ---1-----2----|  
  out: ------a---b--|   a=b1 b=c2
zip
  in : -a----b---c--|   
  in : ---1-----2----|  
  out: ---a-----b----|  a=a1 b=b2
merge
  in : -a----b---c--|   
  in : ---1-----2----|  
  out: -a-1--b--2c---|  
concat
  in : -a--b|           
  in : -1--2|           
  out: -a--b-1--2|      
race
  in : ---a--b|         
  in : -1--2---|        
  out: -1--2---|        

The first three lines are the ones people mix up most. combineLatest gave its first value only at frame 3, when the digit source emitted for the first time; the a that came before produced nothing on its own. After that it emitted the current pair on every new letter or digit (a1, b1, b2, c2). withLatestFrom emitted only when a letter arrived, and silently swallowed the a that came before the digit source had emitted; the digits themselves trigger nothing. zip paired values by position: the first letter with the first digit (a1), the second letter with the second digit (b2); c's partner never came.

combineLatest and startWith: Catalog Filters

The answer to the prediction:

--- catalog filters
 1000 ms  one second later

No query at all. combineLatest emits nothing until it has seen at least one value from every source; because the price range never emitted, the list never loads, whatever the user picks. No error, no warning: the page just stays empty. In real applications, form controls and route parameters are very often exactly this kind of source, one that emits nothing until the user touches it.

The fix is to give every filter a start value. The startWith operator puts a value in front of the source that is emitted synchronously on subscribe. The snapshot's src/05/catalog-api.ts mimics Bookstore's catalog endpoint: it filters the six books of the shared sample data by category, price and sort order, and answers after 200 ms. We combine the filters with the dictionary signature; the result is a single object with named fields:

import { combineLatest, distinctUntilChanged, map, startWith, switchMap } from "rxjs";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { getBooks, toQueryString, type CatalogQuery, type PriceRange, type SortOrder } from "./catalog-api.ts";

runInVirtualTime("combineLatest + startWith", () => {
  // what the user does on the catalog page, and when
  const category$ = emitAt<CatalogQuery["category"]>([100, "classics"]);
  const priceRange$ = emitAt<PriceRange>([500, { min: 0, max: 350 }]);
  const sort$ = emitAt<SortOrder>([900, "price-asc"]);

  const query$ = combineLatest({
    category: category$.pipe(startWith<CatalogQuery["category"]>("all")),
    priceRange: priceRange$.pipe(startWith({ min: 0, max: 1000 })),
    sort: sort$.pipe(startWith<SortOrder>("title")),
  });

  query$
    .pipe(
      distinctUntilChanged((a, b) => toQueryString(a) === toQueryString(b)),
      switchMap((query) => getBooks(query)),
      map((titles) => titles.join(", ")),
    )
    .subscribe((list) => log(`screen: ${list}`));
});

--- combineLatest + startWith
    0 ms  -> GET /books?category=all&min=0&max=1000&sort=title
  100 ms  x  GET /books?category=all&min=0&max=1000&sort=title cancelled
  100 ms  -> GET /books?category=classics&min=0&max=1000&sort=title
  300 ms  <- GET /books?category=classics&min=0&max=1000&sort=title
  300 ms  screen: Emma 320, Frankenstein 280, Middlemarch 380, Pride and Prejudice 300
  500 ms  -> GET /books?category=classics&min=0&max=350&sort=title
  700 ms  <- GET /books?category=classics&min=0&max=350&sort=title
  700 ms  screen: Emma 320, Frankenstein 280, Pride and Prejudice 300
  900 ms  -> GET /books?category=classics&min=0&max=350&sort=price-asc
 1100 ms  <- GET /books?category=classics&min=0&max=350&sort=price-asc
 1100 ms  screen: Frankenstein 280, Pride and Prejudice 300, Emma 320

As soon as the page opened (at 0 ms) the first query went out with the default filters; when the user changed the category at 100 ms, switchMap cancelled that request and sent a new one. The next two changes turned into one query each, and the final screen shows classics under 350 TRY, sorted by ascending price: Frankenstein 280, Pride and Prejudice 300, Emma 320. The chain is built from the pieces of the previous two posts: distinctUntilChanged compares by query string (the object trap from the third post), switchMap cancels the old query (the fourth post).

The dictionary signature (combineLatest({ category, priceRange, sort })) reads better than the array signature: you access values as filters.category instead of filters[0], and adding a new filter does not shift any positions. In RxJS 7 the rest-parameter signature such as combineLatest(a$, b$) is deprecated; the type definitions say it will be removed in the next major version. Pass an array or an object.

withLatestFrom: Trigger and Context

On Bookstore's institutional ordering screen, a user buying for a school, library or company fills in the quantity and the institution, then presses the "Request Quote" button. Here the request should go out only when the button is pressed, not on every field change. Let's compare the same streams with withLatestFrom and combineLatest:

import { combineLatest, map, startWith, withLatestFrom } from "rxjs";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";

runInVirtualTime("withLatestFrom: the click decides, the form only supplies context", () => {
  // the institutional order form: the buyer edits fields, then presses "Request Quote"
  const quantity$ = emitAt([100, 20], [200, 25], [600, 30]).pipe(startWith(10));
  const institution$ = emitAt([150, "Hillside School"]);
  const quoteClicks$ = emitAt([50, "click"], [400, "click"], [800, "click"]);

  quoteClicks$
    .pipe(
      withLatestFrom(quantity$, institution$),
      map(([, quantity, institution]) => `quote: ${quantity} x Dune for ${institution}`),
    )
    .subscribe((line) => log(line));
});

runInVirtualTime("combineLatest: every field change fires", () => {
  const quantity$ = emitAt([100, 20], [200, 25], [600, 30]).pipe(startWith(10));
  const institution$ = emitAt([150, "Hillside School"]);

  combineLatest([quantity$, institution$])
    .pipe(map(([quantity, institution]) => `quote: ${quantity} x Dune for ${institution}`))
    .subscribe((line) => log(line));
});

--- withLatestFrom: the click decides, the form only supplies context
  400 ms  quote: 25 x Dune for Hillside School
  800 ms  quote: 30 x Dune for Hillside School
--- combineLatest: every field change fires
  150 ms  quote: 20 x Dune for Hillside School
  200 ms  quote: 25 x Dune for Hillside School
  600 ms  quote: 30 x Dune for Hillside School

With withLatestFrom a quote was requested only on clicks, each time with the current quantity: 25 at 400 ms, 30 at 800 ms. combineLatest produced a quote on every field change; that is the right behavior for a search list, but for a quote request it is needless traffic and perhaps a request accidentally sent on the institution's behalf.

Something is missing from the output: the click at 50 ms produced nothing. At that moment institution$ had not emitted yet, and withLatestFrom silently drops the trigger if one of its context sources has not emitted. For the user that means "I pressed the button and nothing happened". Give the context sources a meaningful start value with startWith, or disable the button until the required fields are filled in. The division of labor between the two operators can be summed up like this: in combineLatest all sources are equal and any of them triggers; in withLatestFrom one source triggers and the others only supply context.

forkJoin: Once, When Everything Is Done

The order details page needs three pieces of information: the order, the customer and the shipment status. The three requests are independent of each other, so they should go out in parallel, not one after another. forkJoin subscribes to all of them at once and, when they have all completed, gives you the last value of each in a single emission. For order 1042 from the shared sample data (Ada, Dune and Emma):

import { EMPTY, forkJoin, interval, map, timer } from "rxjs";
import { customers, orders } from "../../data/books.ts";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { fakeRequest } from "../04/fake-api.ts";

const order = orders[0]; // order 1042: Ada, Dune + Emma

runInVirtualTime("forkJoin: three calls in parallel, one result", () => {
  forkJoin({
    order: fakeRequest(`GET /orders/${order.id}`, 300, () => order),
    customer: fakeRequest(`GET /customers/${order.customerId}`, 100, () => customers[0]),
    shipment: fakeRequest(`GET /orders/${order.id}/shipment`, 200, () => ({ status: "Shipped", trackingNo: "TRK-1042" })),
  }).subscribe(({ order, customer, shipment }) =>
    log(`order ${order.id} for ${customer.name}: ${order.items.length} items, ${order.status}, shipment ${shipment.status}`),
  );
});

runInVirtualTime("forkJoin with a source that never completes", () => {
  const liveStock$ = interval(1000).pipe(map((i) => 12 - i)); // stock updates, never completes
  forkJoin({
    order: fakeRequest(`GET /orders/${order.id}`, 300, () => order),
    stock: liveStock$,
  }).subscribe(() => log("never printed"));
  timer(5000).subscribe(() => log("still waiting after 5 seconds"));
});

runInVirtualTime("forkJoin edge cases", () => {
  forkJoin([]).subscribe({
    next: (v) => log(`[] next ${JSON.stringify(v)}`),
    complete: () => log("[] complete"),
  });
  forkJoin({ order: fakeRequest(`GET /orders/${order.id}`, 300, () => order), coupons: EMPTY }).subscribe({
    next: () => log("never printed"),
    complete: () => log("{ order, coupons: EMPTY } complete without a value"),
  });
});

--- forkJoin: three calls in parallel, one result
    0 ms  -> GET /orders/1042
    0 ms  -> GET /customers/1
    0 ms  -> GET /orders/1042/shipment
  100 ms  <- GET /customers/1
  200 ms  <- GET /orders/1042/shipment
  300 ms  <- GET /orders/1042
  300 ms  order 1042 for Ada: 2 items, Paid, shipment Shipped
--- forkJoin with a source that never completes
    0 ms  -> GET /orders/1042
  300 ms  <- GET /orders/1042
 5000 ms  still waiting after 5 seconds
--- forkJoin edge cases
    0 ms  [] complete
    0 ms  -> GET /orders/1042
    0 ms  { order, coupons: EMPTY } complete without a value
    0 ms  x  GET /orders/1042 cancelled

In the first scenario the three requests went out together at 0 ms, and the result arrived after the slowest one (300 ms); one after another they would have taken 600 ms. forkJoin is the counterpart of Promise.all in the Promise world, and it is designed for sources such as HTTP that emit one value and complete.

The other two scenarios show the limits of that design. A source that never completes (a live stock stream, a form's valueChanges, a Subject) makes forkJoin wait forever: the order arrived at 300 ms, but there is still no result 5 seconds later. That is also why this virtual-time example relies on runInVirtualTime stopping the virtual clock at 60 seconds; otherwise interval would run forever. With an empty array, forkJoin([]) completes immediately without emitting; for a list such as "fetch the stock of the selected products", the screen may wait for a result and never update when the list is empty. If a source completes without emitting (EMPTY, or a stream in which no value gets through a filter), forkJoin also completes without emitting and cancels the other requests: the x on the last line shows the order request was wasted. An error from one source likewise fails the whole forkJoin and cancels the others; if you want to continue with partial results, you need to add error handling to each source (in this series' error handling post).

zip, merge, concat and race

Back to the last four lines of the marbles. merge puts the values of several sources into one stream in the order they arrive; it does no pairing. If Bookstore's "added to cart" event comes both from the catalog list and from the book details page, merging the two into one stream with merge is enough to update the cart counter. merge completes when all sources have completed; if one of them errors, the whole thing ends.

concat runs its sources one after another: it subscribes to the second source only when the first completes (in the marble, 1 and 2 came after b, starting from the beginning of their own timeline). The typical use is showing a cached value first and the fresh one after: concat(cachedBooks$, fetchBooks$). The thing to watch is that if the first source never completes, the second never starts.

race picks the source that emits first and unsubscribes from the others: in the marble the digit source emitted at frame 1, so the letters never appeared. It is used to ask two sources for the same data (for example, two warehouse services) and take the faster one, or to race a request against a timeout; for timeouts, though, RxJS's own timeout operator is clearer (in this series' post on time).

zip is the one you need least. It pairs values by position, so if one source is fast, its values wait in memory until their partners arrive. Cases where you really mean "first with first, second with second" are rare: request and response pairs sent in the same order, or animation steps. Zipping user events is almost always a mistake; if the two streams run at different speeds, the buffer grows without limit.

In Production

Production problems with combinators usually show up as "nothing happens", not as errors: when combineLatest waits for a source, when forkJoin waits for a source that never completes, or when withLatestFrom swallows a trigger, the console shows no trace. That is why watching each source on its own helps: during development, temporarily adding a tap({ next, complete }) to the end of each source to see which one never emits or never completes answers "why does the list not load?" in minutes. Turning such debugging helpers into a custom operator is the topic of this series' custom operator post.

On the network side, the difference between forkJoin and combineLatest can be measured. On the order details page you should see the three requests start at the same time in the Network tab (stacked in the waterfall); if they start one after another like a staircase, there is a nested subscribe or a concatMap chain somewhere. Collecting the order details into a single endpoint on Bookstore's ASP.NET Core side, instead of three separate ones, is an option too; forkJoin on the client does not reduce the number of requests per page, it only brings the waiting time down to that of the slowest request.

Finally, if you combine two streams derived from the same source with combineLatest, every change of the source updates one and then the other, so for a moment an inconsistent pair is emitted in between. If that intermediate value turns into an HTTP request, it is one more needless request; keeping the filters in a single state object, or comparing the query string with distinctUntilChanged, cuts down these repeats.

Common Pitfalls

• Giving combineLatest a source that has not emitted yet. Nothing comes out until every source has emitted at least once; give form and filter sources a start value with startWith.
• Giving forkJoin a source that never completes (a live stream, valueChanges, a Subject). The result never comes; if one value is enough, add take(1) to the source.
• Passing an empty array to forkJoin and waiting for a result. forkJoin([]) completes without emitting; check for an empty list beforehand or add defaultIfEmpty([]).
• Forgetting that a source that completes without emitting (EMPTY, a filter that lets nothing through) ends the whole forkJoin without a value.
• Not knowing that withLatestFrom silently drops a trigger that arrives before its context sources have emitted.
• Pairing user events with zip. The fast side piles up in memory and values get paired wrongly.
• Using the rest-parameter signatures combineLatest(a$, b$) and forkJoin(a$, b$). They are deprecated in RxJS 7; pass an array or an object.

A Small Experiment

★☆☆ In 02-catalog-filters.ts, remove startWith from priceRange$. At which millisecond does the first query go out, and what does the user see on screen until then? Then put startWith back and move the category change from 100 ms to 50 ms; does the number of cancelled requests change?

★★☆ In 03-with-latest-from.ts, stop the click at 50 ms from getting lost: first by giving institution$ a start value, then, without a start value, with a solution that holds clicks back until an institution is chosen. In the second solution, what happens if the user clicks twice in that time?

★★★ Change the order details in 04-fork-join.ts so that when the shipment service fails, the page still shows the order and customer information, with "unknown" for the shipment. Explain which behavior of forkJoin you work around and at which level you catch the error. Hint: we have not covered the error handling operators yet, but you can guess that catchError has to sit in each source's own pipe.

In this post the timing was always set by when the user clicked or how long the API took to answer. In the next post we manage time ourselves: we look at the difference between debounceTime, throttleTime, auditTime and timeout, at RxJS's schedulers, and at polling an order's status in a way that stops while the page is hidden.