Reading Marble Diagrams and the Core Operators: map, filter, scan, take and distinctUntilChanged

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

Predict First

On the catalog page of Bookstore, the online bookstore used as the shared example across this site's series, every click on "Apply" emits the current filter. To avoid going back to the API when the same filter arrives twice in a row, we added distinctUntilChanged:

import { distinctUntilChanged } from "rxjs";
import { showMarbles } from "../../tools/marble.ts";

// the catalog page emits the current filter every time the user clicks "Apply"
showMarbles("distinctUntilChanged()", (cold) =>
  cold("-a--b--c--|", {
    a: { category: "classics" },
    b: { category: "classics" },
    c: { category: "fantasy" },
  }).pipe(distinctUntilChanged()),
);

The user picked "classics" twice in a row, then "fantasy". How many values get through distinctUntilChanged, and how many requests reach the API? If you cannot read the cold("-a--b--c--|") line yet, that is fine; it is the first thing this post teaches.

In the Angular series post "Introduction to RxJS in Angular" we used map, filter and debounceTime in a search box. This post goes through the core operators one by one, edge cases included: map, scan, reduce and pairwise for transformation; filter, take, takeWhile, skip, first, last, distinctUntilChanged and distinctUntilKeyChanged for filtering. We show each one's behavior with a marble diagram.

Reading Marble Diagrams

A marble diagram is a notation that shows what an Observable does over time on a single line. The RxJS documentation, its test tools and the rest of this series all use it:

• - one unit of time (a frame) in which nothing happens.
• a, b, c… a value (next); what each letter stands for is given next to it.
• | complete, # error.
• (b|) everything inside the parentheses happens in the same frame: here the value b, immediately followed by complete.

I did not draw the diagrams in this post by hand. The showMarbles function in the snapshot's tools/marble.ts plays the source marble you give it on RxJS's TestScheduler in virtual time (one character is one frame, that is, one virtual millisecond), records the operator's output and writes it out as a marble, then checks that output with the TestScheduler's own expectObservable. So every "out" line below is RxJS's real behavior; had it been wrong, the script would have failed. Using marbles as a test tool, that is, writing tests with TestScheduler, is the topic of this series' marble testing post. The first example:

import { map, take } from "rxjs";
import { showMarbles } from "../../tools/marble.ts";

// price in TRY -> price in kuruş (minor units)
showMarbles("map((price) => price * 100)", (cold) =>
  cold("-a--b--c--|", { a: 450, b: 320, c: 390 }).pipe(map((price) => price * 100)),
);

// the stock service fails after two values: # is the error
showMarbles("map((stock) => stock < 3)", (cold) =>
  cold("-a--b--#", { a: 12, b: 2 }, new Error("stock service down")).pipe(map((stock) => stock < 3)),
);

// take(2): the second value and complete land on the same frame
showMarbles("take(2)", (cold) => cold("-a--b--c--|", { a: "Dune", b: "Emma", c: "Middlemarch" }).pipe(take(2)));

map((price) => price * 100)
  in : -a--b--c--|      a=450 b=320 c=390
  out: -a--b--c--|      a=45000 b=32000 c=39000
map((stock) => stock < 3)
  in : -a--b--#         a=12 b=2
  out: -a--b--#         a=false b=true #=Error
take(2)
  in : -a--b--c--|      a=Dune b=Emma c=Middlemarch
  out: -a--(b|)         a=Dune b=Emma

Read the diagram from left to right. With map, every value comes out in the same frame, transformed; the timing does not change. When the source errors, map errors at the same moment: operators pass errors downstream by default. take(2) has something interesting: complete arrives together with the second value ((b|)). take completes the moment it has counted enough values, without waiting for the source, and unsubscribes from the source; c is never even produced.

An Operator Is a Function That Takes an Observable and Returns One

In the first post we wrote map ourselves for MiniObservable. RxJS is no different: map(fn) returns not an Observable but a function. RxJS calls it a pipeable operator, of type OperatorFunction<In, Out>:

import { map, of } from "rxjs";

// an operator is a function: Observable in, Observable out
const toMinorUnits = map((price: number) => price * 100);
console.log(typeof toMinorUnits);

// calling it directly ...
toMinorUnits(of(450, 320)).subscribe((v) => console.log("direct call:", v));

// ... is exactly what pipe does for you
of(390).pipe(toMinorUnits).subscribe((v) => console.log("pipe:", v));

// one operator value, reused on two sources
const cart$ = of(450, 320);
const wishlist$ = of(280);
cart$.pipe(toMinorUnits).subscribe((v) => console.log("cart:", v));
wishlist$.pipe(toMinorUnits).subscribe((v) => console.log("wishlist:", v));

function
direct call: 45000
direct call: 32000
pipe: 39000
cart: 45000
cart: 32000
wishlist: 28000

source$.pipe(a, b, c) is the readable way of writing c(b(a(source$))). No operator changes its source; each returns a new Observable that subscribes to the source. That is why an operator value like toMinorUnits can be reused safely on several sources, and why you can write your own operators by combining existing ones (in this series' custom operator post).

Transformation: map, scan, reduce and pairwise

In Bookstore's cart the user adds Dune (450 TRY) and Emma (320 TRY), removes Dune and adds The Hobbit (390 TRY). Let's treat the cart changes as a stream of amounts in TRY and keep a running subtotal:

import { pairwise, reduce, scan } from "rxjs";
import { showMarbles } from "../../tools/marble.ts";

// cart changes in TRY: +Dune, +Emma, -Dune, +The Hobbit
const cartChanges = { a: 450, b: 320, c: -450, d: 390 };

showMarbles("scan((total, change) => total + change, 0)", (cold) =>
  cold("-a--b--c--d--|", cartChanges).pipe(scan((total, change) => total + change, 0)),
);

showMarbles("reduce((total, change) => total + change, 0)", (cold) =>
  cold("-a--b--c--d--|", cartChanges).pipe(reduce((total, change) => total + change, 0)),
);

// Dune's stock level; pairwise gives [previous, current]
showMarbles("pairwise()", (cold) => cold("-a--b--c--|", { a: 12, b: 11, c: 9 }).pipe(pairwise()));

scan((total, change) => total + change, 0)
  in : -a--b--c--d--|   a=450 b=320 c=-450 d=390
  out: -a--b--c--d--|   a=450 b=770 c=320 d=710
reduce((total, change) => total + change, 0)
  in : -a--b--c--d--|   a=450 b=320 c=-450 d=390
  out: -------------(a|) a=710
pairwise()
  in : -a--b--c--|      a=12 b=11 c=9
  out: ----a--b--|      a=[ 12, 11 ] b=[ 11, 9 ]

scan and reduce take the same accumulator function but answer different questions. scan emits the total so far on every change: 450, 770, 320, 710. That is exactly the subtotal on the cart screen; scan is the standard way to keep running state inside a stream. reduce gives a single result only when the source completes. On a stream that never completes, like the cart, reduce emits nothing at all; the user never sees a total. Use reduce on streams that really end, such as collecting every page of a paged API.

pairwise pairs every value with the previous one; the first value on its own produces nothing. When the stock drops from 12 to 11 and then to 9, you get [12, 11] and [11, 9]. "Did the stock just drop below 3?" cannot be answered from a single value, you need the previous one; to show the warning only once, when the threshold is crossed, pairwise followed by a filter is enough.

Filtering: filter, takeWhile and skip

filter looks like the simplest operator, but on the TypeScript side it has something to teach. Let the cart events be a union type:

import { filter, map, skip, takeWhile } from "rxjs";
import { showMarbles } from "../../tools/marble.ts";

type AddToCart = { type: "add"; title: string; price: number };
type RemoveFromCart = { type: "remove"; title: string };
type CartEvent = AddToCart | RemoveFromCart;

const events: Record<string, CartEvent> = {
  a: { type: "add", title: "Dune", price: 450 },
  b: { type: "remove", title: "Dune" },
  c: { type: "add", title: "Bookmark", price: 0 }, // a free gift added by a campaign
  d: { type: "add", title: "The Hobbit", price: 390 },
};

// an explicit type guard: after this filter, TypeScript knows every value is AddToCart
const isPaidAdd = (e: CartEvent): e is AddToCart => e.type === "add" && e.price > 0;

showMarbles("filter(isPaidAdd), map((e) => e.price)", (cold) =>
  cold("-a--b--c--d--|", events).pipe(
    filter(isPaidAdd),
    map((e) => e.price),
  ),
);

// Middlemarch's stock while orders come in
const stock = { a: 5, b: 3, c: 2, d: 4 };
showMarbles("takeWhile((s) => s >= 3)", (cold) =>
  cold("-a--b--c--d--|", stock).pipe(takeWhile((s) => s >= 3)),
);
showMarbles("takeWhile((s) => s >= 3, true)", (cold) =>
  cold("-a--b--c--d--|", stock).pipe(takeWhile((s) => s >= 3, true)),
);
showMarbles("skip(1)", (cold) => cold("-a--b--c--d--|", stock).pipe(skip(1)));

filter(isPaidAdd), map((e) => e.price)
  in : -a--b--c--d--|   a={ type: 'add', title: 'Dune', price: 450 } b={ type: 'remove', title: 'Dune' } c={ type: 'add', title: 'Bookmark', price: 0 } d={ type: 'add', title: 'The Hobbit', price: 390 }
  out: -a--------b--|   a=450 b=390
takeWhile((s) => s >= 3)
  in : -a--b--c--d--|   a=5 b=3 c=2 d=4
  out: -a--b--|         a=5 b=3
takeWhile((s) => s >= 3, true)
  in : -a--b--c--d--|   a=5 b=3 c=2 d=4
  out: -a--b--(c|)      a=5 b=3 c=2
skip(1)
  in : -a--b--c--d--|   a=5 b=3 c=2 d=4
  out: ----a--b--c--|   a=3 b=2 c=4

The return type of isPaidAdd is e is AddToCart; RxJS's filter overload recognizes this type guard, and in the following map the type of e becomes AddToCart. For simple checks this is no longer needed: since TypeScript 5.5 the compiler infers the type predicate of an arrow function such as filter((e) => e.type === "add") on its own, and narrowing worked when I tried it in TypeScript 6.0. But once the condition gets more complex, the inference stops. Write the same filter without an explicit guard:

filter((e) => e.type === "add" && e.price > 0),
map((e) => e.price),

src/03/05-filter.ts(21,18): error TS2339: Property 'price' does not exist on type 'CartEvent'.

The compiler cannot say that the value is not an AddToCart when the condition returns false (the free gift is an AddToCart too), so it does not infer a predicate. The fix is a named, explicit type guard, as above.

takeWhile completes the first time the condition is false and does not emit that value: when the stock dropped to 2, the stream ended and the 4 after it was never seen. Pass inclusive: true as the second parameter and the value that broke the condition is emitted too; that version is handier when you do not want to lose the "stock fell below the threshold" information. Do not confuse takeWhile with filter: filter skips a value that does not match and carries on, takeWhile ends the stream. skip(n) drops the first n values; it is useful with sources that hand you their current value as soon as you subscribe, such as the BehaviorSubject we will see in the Subject post, when you only care about later changes.

first and last: the EmptyError Trap

first and last take a single value from the stream and complete. They are easy to use, but you need to know what they do on an empty stream:

import { EMPTY, first, last, of } from "rxjs";
import { showMarbles } from "../../tools/marble.ts";

const prices = { a: 450, b: 320, c: 390 };

showMarbles("first()", (cold) => cold("-a--b--c--|", prices).pipe(first()));
showMarbles("first((p) => p < 400)", (cold) => cold("-a--b--c--|", prices).pipe(first((p) => p < 400)));
showMarbles("last()", (cold) => cold("-a--b--c--|", prices).pipe(last()));
showMarbles("first() on an empty stream", (cold) => cold("----|").pipe(first()));

// no book under 100 TRY: the error message, and the default value that avoids it
of(450, 320, 390).pipe(first((p) => p < 100)).subscribe({
  error: (err: Error) => console.log(`${err.name}: ${err.message}`),
});
EMPTY.pipe(first(undefined, "no books")).subscribe((v) => console.log("with default:", v));

first()
  in : -a--b--c--|      a=450 b=320 c=390
  out: -(a|)            a=450
first((p) => p < 400)
  in : -a--b--c--|      a=450 b=320 c=390
  out: ----(a|)         a=320
last()
  in : -a--b--c--|      a=450 b=320 c=390
  out: ----------(a|)   a=390
first() on an empty stream
  in : ----|            
  out: ----#             #=EmptyError
EmptyError: no elements in sequence
with default: no books

first() completes as soon as it gets the first value and unsubscribes from the source; it is similar to take(1). last() has to wait for the source to end, because it can only know which value was last when complete arrives. The real difference shows on an empty stream: if the source completes without emitting (or no value matches the predicate), first and last error with EmptyError. take(1) simply completes in the same situation. If a query in Bookstore such as "the first discounted book in this category" can come back empty, first throws an error, and if nobody catches it, it is reported as an uncaught error as in the first post. Either pass a default value as the second parameter, or use take(1) if an empty result is normal for you.

distinctUntilChanged and distinctUntilKeyChanged

The answer to the prediction:

distinctUntilChanged()
  in : -a--b--c--|      a={ category: 'classics' } b={ category: 'classics' } c={ category: 'fantasy' }
  out: -a--b--c--|      a={ category: 'classics' } b={ category: 'classics' } c={ category: 'fantasy' }

All three values got through, so three requests reach the API. distinctUntilChanged compares with === by default; a and b have the same content but are different objects, so they count as "different". Because forms and route parameters produce a new object every time, this happens a lot in real code. With primitive values the operator works as expected:

import { distinctUntilChanged, distinctUntilKeyChanged, map } from "rxjs";
import { showMarbles } from "../../tools/marble.ts";

// search box: the same term again should not send a new request
showMarbles("distinctUntilChanged() on search terms", (cold) =>
  cold("-a--a--b--b--a--|", { a: "dune", b: "emma" }).pipe(distinctUntilChanged()),
);

type CatalogFilter = { category: string; maxPrice: number };
const filters: Record<string, CatalogFilter> = {
  a: { category: "classics", maxPrice: 400 },
  b: { category: "classics", maxPrice: 400 }, // same content, new object
  c: { category: "classics", maxPrice: 300 },
};

// 1) compare the fields yourself
showMarbles("distinctUntilChanged(sameFilter)", (cold) =>
  cold("-a--b--c--|", filters).pipe(
    distinctUntilChanged((prev, curr) => prev.category === curr.category && prev.maxPrice === curr.maxPrice),
  ),
);

// 2) only one field matters
showMarbles("distinctUntilKeyChanged('category')", (cold) =>
  cold("-a--b--c--|", filters).pipe(distinctUntilKeyChanged("category")),
);

// 3) turn the filter into a primitive first, e.g. the query string it produces
showMarbles("map(toQuery), distinctUntilChanged()", (cold) =>
  cold("-a--b--c--|", filters).pipe(
    map((f) => `category=${f.category}&maxPrice=${f.maxPrice}`),
    distinctUntilChanged(),
  ),
);

distinctUntilChanged() on search terms
  in : -a--a--b--b--a--| a=dune b=emma
  out: -a-----b-----a--| a=dune b=emma
distinctUntilChanged(sameFilter)
  in : -a--b--c--|      a={ category: 'classics', maxPrice: 400 } b={ category: 'classics', maxPrice: 400 } c={ category: 'classics', maxPrice: 300 }
  out: -a-----b--|      a={ category: 'classics', maxPrice: 400 } b={ category: 'classics', maxPrice: 300 }
distinctUntilKeyChanged('category')
  in : -a--b--c--|      a={ category: 'classics', maxPrice: 400 } b={ category: 'classics', maxPrice: 400 } c={ category: 'classics', maxPrice: 300 }
  out: -a--------|      a={ category: 'classics', maxPrice: 400 }
map(toQuery), distinctUntilChanged()
  in : -a--b--c--|      a={ category: 'classics', maxPrice: 400 } b={ category: 'classics', maxPrice: 400 } c={ category: 'classics', maxPrice: 300 }
  out: -a-----b--|      a=category=classics&maxPrice=400 b=category=classics&maxPrice=300

Three of the five search terms got through. Note that the final "dune" passed although it equals the first "dune", because distinctUntilChanged only compares with the previous value, not with the whole history. There is a separate operator (distinct) for "have we ever searched for this term?", but it keeps a Set that grows without limit, so use it carefully on long-lived streams.

For objects there are three options. A comparator is the most explicit, but it is easy to forget to update it when a new field is added to the filter: then changing the new field sends no request. distinctUntilKeyChanged('category') looks at a single field; in the example nothing got through even though maxPrice changed from 400 to 300, so the list would not update when the user changed the price filter. That is the wrong choice here. The third way is the most robust in my view: turn the filter into the query string that goes to the API first, and compare that primitive. Since that string is what determines the request anyway, two filters that produce the same request count as equal, and the comparison stays correct when a new field is added.

In Production

Most of these operators cost next to nothing, but a wrong choice is paid for in network traffic and user experience. On Bookstore's catalog page, distinctUntilChanged not working on objects means the same query goes out again on every "Apply" click. Measuring it is simple: in the DevTools Network tab, count consecutive requests with the same query string, or log with a tap that counts requests in the service layer. If the number is higher than the number of real filter changes the user made, the comparison is wrong.

Watch the size of the state you keep with scan, too. A single number such as the cart total is no problem; but if you collect every value into an array with scan((all, item) => [...all, item], []), on a stream that never completes that array grows as long as the page stays open and is copied in full on every new value. If you really need the whole history, set an upper limit; usually the last n values are enough.

Finally, operators that say "I expect exactly this many values", such as first, last and single, throw an error when the expectation is not met. In production, wiring the first post's config.onUnhandledError setting to a logging system so that these errors are visible speeds up answering "why did the list come back empty?".

Common Pitfalls

• Using first() on a stream that can be empty. If the source completes without emitting you get EmptyError; pass a default value or use take(1).
• Assuming distinctUntilChanged compares object contents. The default comparison is ===; for objects pass a comparator or turn them into a primitive first (a query string, an id).
• Watching a single field with distinctUntilKeyChanged and missing changes to the other fields.
• Using reduce or last on a stream that never completes and waiting for a value. Neither emits anything until complete; use scan for running state.
• Assuming takeWhile emits the value that broke the condition. By default it does not; if you need that value too, use inclusive: true.
• Expecting TypeScript to narrow the type in a complex filter condition. If the condition looks at more than one field, write an explicit type guard (e is AddToCart).

A Small Experiment

★☆☆ Extend the cart example in 04-transform.ts with a chain that emits "free shipping unlocked" once, the first time the total goes above 1000 TRY. scan, map and first are enough. Then change the stream of changes so that the total never goes above 1000, and see in the marble output what first does.

★★☆ Using pairwise, write an operator chain that emits a single warning when Dune's stock has just dropped below 3 (previous value 3 or more, new value below 3). Try it with the source "-a--b--c--d--e--|" and { a: 5, b: 2, c: 1, d: 4, e: 2 }; how many warnings do you expect?

★★★ Add a sort: "price-asc" | "price-desc" field to the catalog filter. Show with the marble output why the comparator-based distinctUntilChanged solution misses the new field, then confirm that the query string solution works correctly without any change. What happens if the order of the parameters in the query string changes?

Every operator in this post turned one value into another. In a real application a value often starts a request: a search term turns into an HTTP call, a "Place Order" click into a POST. Then you get an Observable inside an Observable, and the question "what happens if a new one arrives before the previous request finishes?" comes up. The next post answers that question for switchMap, mergeMap, concatMap and exhaustMap.