RxJS Creation Functions and Cold vs Hot Observables: of, from, defer, timer, fromEvent and iif

Version note: The examples in this post were run with RxJS 7.8.2, TypeScript 6.0, Node.js 24 and Vite 8.3; the browser example ran in the Chromium build of Playwright 1.65.

Predict First

The price service of Bookstore, the online bookstore used as the shared example across this site's series, sometimes answers 503. We use the mock API from the first post; the ?fail=2 parameter makes it answer 503 to the first two requests for that book. We set up three Observables and subscribe to two of them with retry(2):

import { defer, from, retry } from "rxjs";

const apiUrl = "http://localhost:5180/api";

interface PriceInfo {
  isbn: string;
  price: number;
  currency: string;
}

// fail = N: the mock API answers 503 to the first N requests for that book
async function fetchPrice(isbn: string, fail = 0): Promise<PriceInfo> {
  const response = await fetch(`${apiUrl}/books/${isbn}/price?fail=${fail}`);
  if (!response.ok) throw new Error(`HTTP ${response.status}`);
  return (await response.json()) as PriceInfo;
}

const emmaPrice$ = from(fetchPrice("9780141439587")); // nobody subscribes to this one
const dunePrice$ = from(fetchPrice("9780441172719", 2));
const hobbitPrice$ = defer(() => fetchPrice("9780547928227", 2));

dunePrice$.pipe(retry(2)).subscribe({
  next: (p) => console.log("Dune:", p.price, p.currency),
  error: (err: Error) => console.log("Dune failed:", err.message),
});
hobbitPrice$.pipe(retry(2)).subscribe({
  next: (p) => console.log("The Hobbit:", p.price, p.currency),
  error: (err: Error) => console.log("The Hobbit failed:", err.message),
});

How many requests in total will you see in the server log? Does a request go out for Emma? Can both Dune and The Hobbit get their price after two 503s? All you need to know is that retry(2) means "on error, subscribe to the source again, at most twice"; the operator's details are in this series' error handling post.

In the first post we built Observables by hand with new Observable and saw that every subscribe runs the producer from the start. This post looks at RxJS's ready-made creation functions: of, from, range, defer, iif, EMPTY, NEVER, throwError, timer, interval, fromEvent and fromFetch. Knowing when each of them runs is also the basis of the "cold" and "hot" Observable distinction.

Observables from Values: of, from and range

When the value is already at hand you use of or from. The difference looks small, but it trips people up often:

import { from, of, range } from "rxjs";
import { books } from "../../data/books.ts";

const log = (label: string) => (value: unknown) => console.log(label, value);

of("Dune", "Emma").subscribe(log("of:"));
of(["Dune", "Emma"]).subscribe(log("of(array):")); // a single value: the array itself
from(["Dune", "Emma"]).subscribe(log("from(array):"));

// any iterable works: Map, Set, string, generator
const stockByTitle = new Map(books.slice(0, 2).map((b) => [b.title, b.stock] as const));
from(stockByTitle).subscribe(log("from(Map):"));

function* catalogPages() {
  yield "page 1";
  yield "page 2";
}
from(catalogPages()).subscribe(log("from(generator):"));

range(1, 3).subscribe(log("range:"));

// Promise and async iterable: asynchronous even when the value is ready
from(Promise.resolve(770)).subscribe(log("from(Promise):"));

async function* orderFeed() {
  yield 1041;
  yield 1042;
}
from(orderFeed()).subscribe(log("from(async iterable):"));

console.log("--- end of synchronous code ---");

of: Dune
of: Emma
of(array): [ 'Dune', 'Emma' ]
from(array): Dune
from(array): Emma
from(Map): [ 'Dune', 12 ]
from(Map): [ 'Emma', 8 ]
from(generator): page 1
from(generator): page 2
range: 1
range: 2
range: 3
--- end of synchronous code ---
from(Promise): 770
from(async iterable): 1041
from(async iterable): 1042

data/books.ts is the shared sample data we also use in the JavaScript and TypeScript series: six books (Dune 450 TRY, Emma 320 TRY…), four customers (Ada, Alan, Grace, Linus) and the order statuses. of emits its arguments as they are: give it an array and you get a single value, the array itself. from unpacks its argument: it emits the elements of an array, Map, Set, string, generator or any other iterable one by one. range(start, count) is a shortcut for consecutive numbers.

The order of the output matters too. Arrays, iterables and range are fully synchronous: their values arrived before the "--- end of synchronous code ---" line. A Promise and an async iterable were delivered asynchronously even though the values were ready, because from reads them the way then and for await do. On top of these, from also accepts another Observable and a ReadableStream; RxJS calls all of these inputs ObservableInput, and that is the type the higher-order operators (such as switchMap) accept.

defer and iif: Leaving the Decision to Subscribe Time

Now the answer to the prediction. With the server started by node api/server.ts, running node src/02/01-predict.ts prints two lines on the client:

Dune failed: HTTP 503
The Hobbit: 390 TRY

The server log shows why:

[api] listening on http://localhost:5180
[api] #1 GET /api/books/9780141439587/price?fail=0
[api] #2 GET /api/books/9780441172719/price?fail=2
[api] #3 GET /api/books/9780547928227/price?fail=2
[api] #1 200
[api] #2 503
[api] #3 503
[api] #4 GET /api/books/9780547928227/price?fail=2
[api] #4 503
[api] #5 GET /api/books/9780547928227/price?fail=2
[api] #5 200

Three things happened. First, Emma's price was requested (#1), although nobody subscribed to emmaPrice$. The cause is not from: the fetchPrice(...) call runs before it is passed to from as an argument, and the work starts the moment the Promise is created. from(promise) is not lazy because the Promise you hand it is already running. Second, only one request went out for Dune (#2). retry(2) resubscribed twice, but each time it got the same, already rejected Promise; no new request reached the server and the error reached the user. Third, The Hobbit got three requests (#3, #4, #5). We gave defer not a Promise but a function that produces one; defer calls that function again on every subscribe, so every retry is a new HTTP request.

The rule is simple: when you need to wrap a function that returns a Promise in an Observable, write defer(() => fn()), not from(fn()). defer brings the first post's "every subscribe runs the producer again" behavior to the Promise world. The same idea applies to anything whose value has to be computed at subscribe time. iif is a special case of it: it leaves the choice between two Observables to subscribe time.

import { defer, iif, of } from "rxjs";

const session = { customer: "Grace", isInstitutional: false };

const retailPrice$ = of(450);
const institutionalPrice$ = of(405); // 10% institutional discount on Dune

// decided once, when this line runs
const ternaryPrice$ = session.isInstitutional ? institutionalPrice$ : retailPrice$;
// decided again on every subscribe
const iifPrice$ = iif(() => session.isInstitutional, institutionalPrice$, retailPrice$);

session.isInstitutional = true; // Grace switches to her school's account

ternaryPrice$.subscribe((p) => console.log("ternary:", p));
iifPrice$.subscribe((p) => console.log("iif:", p));

// defer runs its factory on every subscribe
let cartVersion = 0;
const cartSnapshot$ = defer(() => of(`cart v${++cartVersion}`));
cartSnapshot$.subscribe((v) => console.log("first subscriber:", v));
cartSnapshot$.subscribe((v) => console.log("second subscriber:", v));

ternary: 450
iif: 405
first subscriber: cart v1
second subscriber: cart v2

The ternary operator made its decision when the Observable was created; Grace saw the retail price even after switching to the institutional account. iif called its condition function at subscribe time and gave the right price. In Bookstore's institutional ordering flow the 10% discount depends on the user's current account; decisions like that have to be made at subscribe time, not while the Observable is being set up. iif does the same job as defer(() => cond() ? a$ : b$); with more than two options, defer on its own reads better.

EMPTY, NEVER and throwError

Three special Observables mean "nothing", "wait forever" and "error" in operator chains:

import { EMPTY, NEVER, throwError, type Observable } from "rxjs";

function watch(label: string, source$: Observable<unknown>) {
  source$.subscribe({
    next: (v) => console.log(label, "next", v),
    error: (err: Error) => console.log(label, "error:", err.message),
    complete: () => console.log(label, "complete"),
  });
}

watch("EMPTY:", EMPTY);
watch("NEVER:", NEVER); // prints nothing, ever
watch("throwError:", throwError(() => new Error("Order 1042 not found")));
console.log("done");

EMPTY: complete
throwError: error: Order 1042 not found
done

EMPTY completes right away without emitting; inside a switchMap it is the shortest way to say "don't send a request in this case". NEVER neither emits nor ends; it is handy in tests or for "wait until the user does something", but hand it to something that waits for completion (such as forkJoin) and that waits forever too. throwError sends error as soon as you subscribe.

Watch out with throwError: in RxJS 7 the correct usage is to pass a function that creates the error, throwError(() => new Error(...)). The old throwError(err) signature still compiles, but RxJS's type definitions mark it @deprecated and say it will be removed in the next major version; your editor strikes the call through as well. The factory signature has two benefits: the error is created when it actually happens, so the stack trace points to the right place, and every subscriber gets its own error object.

Time-Based Sources: timer and interval

There are two creation functions that produce values over time. We cover the time operators (debounceTime, throttleTime…) and schedulers in detail in this series' post on time; here we only look at when the first value arrives:

import { interval, take, timer } from "rxjs";

const start = performance.now();
const at = () => `${Math.round((performance.now() - start) / 100) * 100} ms`;

// timer(due): one value (0) after the delay, then complete
timer(300).subscribe({
  next: (v) => console.log(at(), "timer(300):", v),
  complete: () => console.log(at(), "timer(300): complete"),
});

// timer(0, period): first value right away, then every period
timer(0, 200).pipe(take(3)).subscribe((v) => console.log(at(), "timer(0, 200):", v));

// interval(period): first value only after one period
interval(200).pipe(take(3)).subscribe((v) => console.log(at(), "interval(200):", v));

0 ms timer(0, 200): 0
200 ms interval(200): 0
200 ms timer(0, 200): 1
300 ms timer(300): 0
300 ms timer(300): complete
400 ms interval(200): 1
400 ms timer(0, 200): 2
600 ms interval(200): 2

I printed the times rounded to 100 ms; the output was identical across three runs. The first value of interval(200) came at 200 ms, that of timer(0, 200) immediately. When Bookstore polls an order's status, this difference matters: if you want the first request as soon as the page opens you write timer(0, 5000), while interval(5000) makes the user wait 5 seconds for the first answer. timer also accepts a Date; then the value arrives at that moment (for example, when a campaign starts). Neither completes (except single-argument timer), so without take or an unsubscribe they run forever.

DOM and HTTP: fromEvent and fromFetch

When a new book is added in Bookstore's admin screen, a cover image is chosen and previewed right away. The snapshot's web/ folder contains this as a small page served by Vite (npm run web):

<!doctype html>
<html lang="en">
  <head>
    <meta charset="utf-8">
    <title>Bookstore - cover preview</title>
  </head>
  <body>
    <label for="cover">Cover image</label>
    <input id="cover" type="file" accept="image/*">
    <p id="info">No cover selected</p>
    <img id="preview" alt="Cover preview" width="160" height="240">
    <script type="module" src="./cover-preview.ts"></script>
  </body>
</html>

import { filter, fromEvent, map } from "rxjs";

const input = document.querySelector<HTMLInputElement>("#cover")!;
const info = document.querySelector<HTMLParagraphElement>("#info")!;
const preview = document.querySelector<HTMLImageElement>("#preview")!;

type CoverChoice = { ok: true; file: File } | { ok: false; reason: string };

const coverChoice$ = fromEvent(input, "change").pipe(
  map(() => input.files?.[0]),
  filter((file): file is File => file !== undefined),
  map((file): CoverChoice =>
    file.type.startsWith("image/")
      ? { ok: true, file }
      : { ok: false, reason: `${file.name} is not an image` },
  ),
);

let previewUrl: string | undefined;
coverChoice$.subscribe((choice) => {
  if (!choice.ok) {
    info.textContent = choice.reason;
    return;
  }
  if (previewUrl) URL.revokeObjectURL(previewUrl);
  previewUrl = URL.createObjectURL(choice.file);
  preview.src = previewUrl;
  info.textContent = `${choice.file.name} (${choice.file.size} bytes)`;
});

I opened the page in Chromium with Playwright and gave the input first a 69-byte PNG, then a text file. The #info paragraph went from "No cover selected" to "dune-cover.png (69 bytes)" and then to "dune.txt is not an image"; the preview image's src got a blob: URL. fromEvent adds an event listener to the element on subscribe and removes it on unsubscribe; RxJS writes the teardown for you. We also applied the rule from the first post: the file check lives in map, and subscribe only writes to the screen. accept="image/*" is only a hint for the file picker; users can still choose another type of file, which is why the check in code is needed.

For HTTP, RxJS has its own fromFetch (imported from rxjs/fetch). It is like a ready-made version of the getStock we wrote by hand in the first post: lazy, a new request on every subscribe, cancelled with an AbortController on unsubscribe. But there is one difference:

import { throwError } from "rxjs";
import { fromFetch } from "rxjs/fetch";

const apiUrl = "http://localhost:5180/api";

// 1) fromFetch is lazy: the request starts on subscribe
const dune$ = fromFetch(`${apiUrl}/books/9780441172719`, {
  selector: (response) => response.json(),
});
dune$.subscribe((book) => console.log("1) book:", book.title, book.price));

// 2) a 404 is not an error for fetch: response.json() fails on the empty body
const missing$ = fromFetch(`${apiUrl}/books/9780000000000`, {
  selector: (response) => response.json(),
});
missing$.subscribe({
  next: (book) => console.log("2) book:", book),
  error: (err: Error) => console.log("2) error:", err.name, "-", err.message),
});

// 3) check response.ok yourself and turn HTTP errors into error notifications
const checked$ = fromFetch(`${apiUrl}/books/9780000000000`, {
  selector: (response) =>
    response.ok ? response.json() : throwError(() => new Error(`HTTP ${response.status}`)),
});
checked$.subscribe({ error: (err: Error) => console.log("3) error:", err.message) });

// 4) unsubscribe before the response: the request is aborted
const sub = fromFetch(`${apiUrl}/books`).subscribe(() => console.log("4) never printed"));
setTimeout(() => sub.unsubscribe(), 50);

1) book: Dune 450
2) error: SyntaxError - Unexpected end of JSON input
3) error: HTTP 404

[api] #1 GET /api/books/9780441172719
[api] #2 GET /api/books/9780000000000
[api] #3 GET /api/books/9780000000000
[api] #4 GET /api/books
[api] #4 client aborted
[api] #1 200
[api] #2 404
[api] #3 404

Fetch does not treat responses such as 404 and 500 as errors; it rejects only on a network failure. fromFetch follows the same rule. In the second example the real problem (the book does not exist) hid behind a JSON parse error; someone reading the log would look in the wrong place, thinking "the API returns broken JSON". You have to check response.ok inside the selector, as in the third example. Without a selector, fromFetch emits the Response object and completes at once; if you read the body afterwards, that read can no longer be cancelled. So always read the body inside the selector. In an Angular project we use HttpClient for HTTP anyway; fromFetch is useful in code outside Angular or in small scripts.

Cold and Hot Observables

What we have seen so far splits into two groups. In a cold Observable the producer starts on subscribe and every subscriber has its own producer. In a hot Observable the producer runs independently of subscribers; subscribers attach to a source that is already flowing. Cold first:

import { interval, take } from "rxjs";

// cold: every subscriber gets its own interval, starting from 0
const cold$ = interval(100).pipe(take(4));

cold$.subscribe((v) => console.log("cold A:", v));
setTimeout(() => cold$.subscribe((v) => console.log("cold B:", v)), 250);

cold A: 0
cold A: 1
cold A: 2
cold B: 0
cold A: 3
cold B: 1
cold B: 2
cold B: 3

B arrived 250 ms late but missed nothing: its own counter started from 0. Now a hot source. The warehouse fires an event when stock is replenished; the events happen whether or not anyone listens. Node.js also has EventTarget and CustomEvent, so the example runs without a browser:

import { fromEvent, map } from "rxjs";

// hot: the warehouse fires stock events whether anyone listens or not
const warehouse = new EventTarget();
const restocked$ = fromEvent(warehouse, "restocked").pipe(
  map((event) => (event as CustomEvent<string>).detail),
);

["Dune", "Emma", "The Hobbit", "Middlemarch"].forEach((title, i) => {
  setTimeout(() => {
    console.log(`warehouse: ${title} restocked`);
    warehouse.dispatchEvent(new CustomEvent("restocked", { detail: title }));
  }, 100 * (i + 1));
});

restocked$.subscribe((title) => console.log("  hot A:", title));
setTimeout(() => restocked$.subscribe((title) => console.log("  hot B:", title)), 250);

warehouse: Dune restocked
  hot A: Dune
warehouse: Emma restocked
  hot A: Emma
warehouse: The Hobbit restocked
  hot A: The Hobbit
  hot B: The Hobbit
warehouse: Middlemarch restocked
  hot A: Middlemarch
  hot B: Middlemarch

B missed Dune and Emma; those events happened before B subscribed and will not come again. Technically fromEvent itself adds a new listener on every subscribe, so its mechanism works like a cold Observable; but the source it wraps (DOM events, the warehouse) is hot. In practice the question to ask is: "Can a late subscriber get the values it missed, and do two subscribers cause the same work to happen twice?"

Turning a cold Observable into a hot one, that is, sharing a single producer among many subscribers, is the topic of this series' multicasting post; Subjects are RxJS's counterpart of hot sources and get their own post.

In Production

The choice of creation function shows up directly in network traffic. If you wrap an old Promise-based service for price, stock and reviews on Bookstore's product page, use defer; from(service.getPrice()) starts a request before the page has even rendered, possibly one that will never be used, and makes retry useless. We saw that in the server log of the prediction example: Emma's request went out without anyone asking, and Dune's retry never reached the server.

When measuring, check two things in the DevTools Network tab: a request's "Initiator" column (which line of code started it) and whether a retry really sends a new request to the same URL. A failure whose retry never reaches the server shows up in the log as "a single 503" and is easy to miss. On the fromEvent side, the thing to measure is the listener count: the getEventListeners(element) helper in the Chromium DevTools console shows the listeners left behind by subscriptions that were never unsubscribed. In Angular, forgetting to unsubscribe when a component is destroyed leaves one more listener behind on every page visit.

Timed sources (interval, timer(0, n)) raise the same question: who unsubscribes, and when? Any creation function that never completes and has no defined end (take, takeUntil, the component being destroyed) is a leak.

Common Pitfalls

• Writing of(fetch(url)) or from(fetch(url)). The request starts when the Observable is created, not on subscribe; retrying or resubscribing sends no new request. Use defer(() => fetch(url)) or fromFetch(url) instead.
• Mixing up of(array) and from(array). of emits the array as a single value; if you want each element as its own value, use from.
• Using the deprecated throwError(err) signature. Write throwError(() => new Error(...)); the error is created when it actually happens.
• Expecting fromFetch to error on 404 and 500. Fetch rejects only on network failures; check response.ok inside the selector.
• Choosing an Observable with a ternary operator and expecting the decision to happen at subscribe time. That decision is made once, when the line runs; for state that can change, use iif or defer.
• Assuming the first value comes immediately when polling with interval. It comes after one period; to start right away, use timer(0, period).

A Small Experiment

★☆☆ Rewrite dunePrice$ in 01-predict.ts with defer and run it with fail=2. How many requests for Dune do you see in the server log? Then change it to fail=3; explain why the result changes with retry(2).

★★☆ In the warehouse example in 08-hot.ts, what has to change so that B can also see the events it missed? First try wrapping restocked$ in your own new Observable that keeps the last event; then note in which situation this approach gives a wrong result (for example, when the warehouse sells out of a book again). We will see the ready-made solution in the Subject and multicasting posts.

★★★ Add a rule to the cover preview page that shows a warning instead of a preview when the chosen file is larger than 2 MB. Add the rule as a new case of the CoverChoice type inside map, and leave the subscribe side as code that only writes to the screen. Try it by choosing a fake 3 MB file with Playwright or in a browser.

We can now produce Observables; the next question is how to transform them. In the next post we learn to read marble diagrams and look at the core operators such as map, filter, scan, take and distinctUntilChanged: we will keep a running subtotal of Bookstore's cart with scan and skip a needless request when the same filter value arrives again.