Understanding RxJS Observables from the Inside: The Contract, Laziness, Teardown and Your Own 40-Line Observable

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

Predict First

The code below comes from Bookstore, the online bookstore used as the shared example across this site's series: an Observable that emits Dune's stock level. Before you run it, write down which lines appear in the console and in which order.

import { Observable } from "rxjs";

const duneStock$ = new Observable<number>((subscriber) => {
  console.log("producer started");
  subscriber.next(12);
  subscriber.next(11);
  subscriber.complete();
  subscriber.next(10);
});

console.log("before subscribe");
duneStock$.subscribe((stock) => console.log("A got", stock));
duneStock$.subscribe((stock) => console.log("B got", stock));
console.log("after subscribe");

Three questions: how many times is "producer started" printed? Does the value 10 show up? Does "after subscribe" come before or after the stock values? Most people learned Observables as "something asynchronous", so they say "after subscribe" comes first. The answer is a little further down.

In the Angular series post "Introduction to RxJS in Angular" we covered the difference between Observables and Promises, the first operators, a service built on BehaviorSubject and the async pipe. This series picks up from there: the goal is to give a reasoned answer to "which operator, and why". The first post looks inside the Observable: its contract, its laziness, teardown and, to show that RxJS is not magic, a 40-line Observable of our own.

Running the Examples

The code for this post lives in the series snapshot under src/01/. The project is not an Angular app but a plain Node.js project; RxJS is easier to see on its own, away from Angular. Node.js 24 can run .ts files directly by stripping their types, so there is no build step:

npm install
node src/01/01-predict.ts

before subscribe
producer started
A got 12
A got 11
producer started
B got 12
B got 11
after subscribe

That answers the prediction. "producer started" was printed twice, because every subscribe call runs the function you passed to new Observable from the start. 10 never arrived, because a value sent after complete() breaks the contract. "after subscribe" came last because nothing in this producer is asynchronous: the values were delivered synchronously, inside the subscribe call. An Observable does not have to be asynchronous; the producer decides that.

Push and Pull: Where the Observable Fits

In JavaScript there are four basic relationships between the side that produces a value (the producer) and the side that uses it (the consumer). With pull, the consumer says "give it to me now"; with push, the producer says "I'll send it when it's ready". Then there is the number of values: one, or many?

Let's put all four side by side with the same data:

import { Observable } from "rxjs";

// 1) Function: pull, a single value
function readStock(): number {
  return 12;
}
console.log("function:", readStock());

// 2) Iterator: pull, many values; the consumer calls next()
function* stockHistory() {
  yield 12;
  yield 11;
  yield 10;
}
const it = stockHistory();
console.log("iterator:", it.next().value, it.next().value);

// 3) Promise: push, a single value; starts working immediately
const stockPromise = new Promise<number>((resolve) => {
  console.log("promise executor runs now");
  setTimeout(() => resolve(12), 10);
});

// 4) Observable: push, many values; nothing runs until subscribe
const stock$ = new Observable<number>((subscriber) => {
  console.log("observable producer runs now");
  subscriber.next(12);
  setTimeout(() => {
    subscriber.next(11);
    subscriber.complete();
  }, 20);
});

console.log("--- subscribing ---");
stockPromise.then((v) => console.log("promise:", v));
stock$.subscribe({
  next: (v) => console.log("observable:", v),
  complete: () => console.log("observable: complete"),
});

function: 12
iterator: 12 11
promise executor runs now
--- subscribing ---
observable producer runs now
observable: 12
promise: 12
observable: 11
observable: complete

Look at two lines. "promise executor runs now" appeared before "--- subscribing ---": the Promise started working before anyone called then. The Observable's producer only ran on subscribe, sent its first value (12) synchronously and the second one asynchronously 20 ms later. The same Observable can produce values both synchronously and asynchronously; a Promise cannot, its then callback always runs one microtask later (the JavaScript series post "The JavaScript Event Loop in Depth" explains that ordering).

The comparison that explains Observables best is this: an Observable is a function that, instead of returning values, sends them to a callback. Subscribing is calling that function. Call a function twice and its body runs twice; subscribe to an Observable twice and its producer runs twice.

The Observable Contract: next, Then At Most One error or complete

The RxJS documentation sums up the rules of an Observable as a regular expression: next*(error|complete)?. That is zero or more next calls, followed by at most one error or one complete. Once either arrives, that execution is over and nothing sent afterwards reaches the consumer. Let's try it with an order status stream:

import { Observable } from "rxjs";

const orderStatus$ = new Observable<string>((subscriber) => {
  subscriber.next("Pending");
  subscriber.next("Paid");
  subscriber.error(new Error("payment provider timeout"));
  subscriber.next("Shipped");
  subscriber.complete();
  console.log("producer: closed =", subscriber.closed);
  return () => console.log("teardown");
});

orderStatus$.subscribe({
  next: (status) => console.log("next:", status),
  error: (err: Error) => console.log("error:", err.message),
  complete: () => console.log("complete"),
});

next: Pending
next: Paid
error: payment provider timeout
producer: closed = true
teardown

The "Shipped" after the error and the complete() were silently dropped, and no exception was thrown. RxJS enforces this through the Subscriber object it hands to the producer: the moment error or complete is called, closed becomes true and later calls do nothing. A producer can also check subscriber.closed and stop doing useless work; in a long loop that check pays off.

The last line matters too: the teardown ran as soon as the producer function returned. Because the execution had already finished inside the producer, RxJS calls the returned teardown the moment it receives it. So "teardown runs on unsubscribe" is an incomplete rule; the accurate one is "teardown runs when the execution ends": on error, complete or unsubscribe, whichever comes first.

Laziness: Every subscribe Runs the Producer Again

The most expensive consequence of laziness shows up with HTTP. The snapshot's api/server.ts is a small stand-in for the Bookstore API: it logs every request with a number and answers after 200 ms. Let's write an Observable that fetches stock; we start fetch inside the producer and cancel it in the teardown with an AbortController:

import { Observable } from "rxjs";

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

interface StockInfo {
  isbn: string;
  stock: number;
}

function getStock(isbn: string): Observable<StockInfo> {
  return new Observable<StockInfo>((subscriber) => {
    const controller = new AbortController();
    console.log(`GET /books/${isbn}/stock`);

    fetch(`${apiUrl}/books/${isbn}/stock`, { signal: controller.signal })
      .then((response) => response.json() as Promise<StockInfo>)
      .then((info) => {
        subscriber.next(info);
        subscriber.complete();
      })
      .catch((err) => subscriber.error(err));

    // teardown: runs after unsubscribe or complete
    return () => controller.abort();
  });
}

const duneStock$ = getStock("9780441172719");

// two places subscribe to the same Observable
duneStock$.subscribe((info) => console.log("badge:", info.stock));
duneStock$.subscribe((info) => console.log("cart warning:", info.stock < 3));

// a third subscriber gives up before the response arrives
const subscription = duneStock$.subscribe((info) => console.log("never printed", info));
setTimeout(() => subscription.unsubscribe(), 50);

Run node api/server.ts in one terminal and node src/01/04-stock-request.ts in another, and the client prints:

GET /books/9780441172719/stock
GET /books/9780441172719/stock
GET /books/9780441172719/stock
badge: 12
cart warning: false

The server log tells the same story:

[api] listening on http://localhost:5180
[api] #1 GET /api/books/9780441172719/stock
[api] #2 GET /api/books/9780441172719/stock
[api] #3 GET /api/books/9780441172719/stock
[api] #3 client aborted
[api] #1 200
[api] #2 200

There is a single duneStock$ in the code, yet three requests reached the server. The third one was unsubscribed after 50 ms, the teardown called abort(), and the server saw the connection close. This is not a bug, it is the definition of an Observable: duneStock$ is not a request but a recipe for making one. Angular's HttpClient works the same way; the Angular documentation states plainly that HttpClient Observables are cold and that subscribing to the same Observable several times triggers several backend requests. Binding two async pipes in a template to the same Observable therefore means two requests. Sharing one request among many subscribers is the topic of this series' multicasting post (share, shareReplay); we look at the cold and hot distinction in the next post.

Teardown: What Gets Cleaned Up When the Execution Ends?

Real producers open resources: timers, event listeners, WebSockets, HTTP connections. The teardown function is the one correct place to close them. Let's write a producer that simulates stock changes: every 100 ms the stock drops by one as if an order came in, and the stream completes at zero.

import { Observable } from "rxjs";

// A fake producer: Dune's stock drops by one every 100 ms (as if orders came in)
function stockChanges(title: string, start: number): Observable<number> {
  return new Observable<number>((subscriber) => {
    let stock = start;
    console.log(`[${title}] timer started`);
    const id = setInterval(() => {
      stock -= 1;
      subscriber.next(stock);
      if (stock === 0) subscriber.complete();
    }, 100);

    return () => {
      clearInterval(id);
      console.log(`[${title}] timer cleared`);
    };
  });
}

// 1) the consumer stops early: teardown runs on unsubscribe
const sub = stockChanges("Dune", 12).subscribe((s) => console.log("Dune stock:", s));
setTimeout(() => sub.unsubscribe(), 350);

// 2) the producer completes: teardown runs without any unsubscribe call
setTimeout(() => {
  stockChanges("Middlemarch", 3).subscribe({
    next: (s) => console.log("Middlemarch stock:", s),
    complete: () => console.log("Middlemarch: sold out"),
  });
}, 500);

[Dune] timer started
Dune stock: 11
Dune stock: 10
Dune stock: 9
[Dune] timer cleared
[Middlemarch] timer started
Middlemarch stock: 2
Middlemarch stock: 1
Middlemarch stock: 0
Middlemarch: sold out
[Middlemarch] timer cleared

The interval was cleared in both cases. In the second, nobody called unsubscribe; complete() ended the execution, so RxJS ran the teardown itself. That is why unsubscribing from an Observable that completes (such as a single-response HTTP request) is not mandatory, while unsubscribing from one that never completes (an interval, events, a WebSocket) is.

And if no teardown is returned at all? Here is the same producer without one:

import { Observable } from "rxjs";

let ticks = 0;

const leakyStock$ = new Observable<number>((subscriber) => {
  setInterval(() => {
    ticks++;
    subscriber.next(ticks);
  }, 100);
  // no teardown returned: nobody can stop this interval
});

const sub = leakyStock$.subscribe((n) => console.log("value", n));
setTimeout(() => {
  sub.unsubscribe();
  console.log("unsubscribed at tick", ticks);
}, 250);
setTimeout(() => {
  console.log("ticks one second later:", ticks);
  process.exit(0);
}, 1250);

value 1
value 2
unsubscribed at tick 2
ticks one second later: 12

No values were printed after unsubscribe, which gives the impression that the code works. But the counter went from 2 to 12: the interval keeps running in the background, and its subscriber.next calls go nowhere on a closed subscriber. That is also why the last line has process.exit(0); without it the Node process never exits. In an Angular component this is a timer that keeps running after you leave the page, plus one more every time you come back.

Your Own Observable in 40 Lines

Everything we have seen so far (laziness, the contract, teardown, pipe) we can write ourselves. The class below is not RxJS's code but a small model of its idea; the Observable part is under 40 lines, and with map and filter the file has 46:

export interface Observer<T> {
  next: (value: T) => void;
  error: (err: unknown) => void;
  complete: () => void;
}
type Teardown = (() => void) | void;
type Operator<A, B> = (source: MiniObservable<A>) => MiniObservable<B>;

export class MiniObservable<T> {
  readonly producer: (subscriber: Observer<T>) => Teardown;
  constructor(producer: (subscriber: Observer<T>) => Teardown) {
    this.producer = producer;
  }

  subscribe(partial: Partial<Observer<T>>): { unsubscribe: () => void } {
    let closed = false;
    let teardown: Teardown = undefined;
    const close = () => {
      if (closed) return;
      closed = true;
      teardown?.();
    };
    const subscriber: Observer<T> = {
      next: (v) => { if (!closed) partial.next?.(v); },
      error: (e) => { if (!closed) { partial.error?.(e); close(); } },
      complete: () => { if (!closed) { partial.complete?.(); close(); } },
    };
    try {
      teardown = this.producer(subscriber);
    } catch (err) {
      subscriber.error(err);
    }
    if (closed) teardown?.(); // the producer finished synchronously
    return { unsubscribe: close };
  }

  pipe(...operators: Operator<any, any>[]): MiniObservable<any> {
    return operators.reduce((source, op) => op(source), this as MiniObservable<any>);
  }
}

export const map = <A, B>(fn: (v: A) => B): Operator<A, B> => (source) =>
  new MiniObservable<B>((sub) => source.subscribe({ ...sub, next: (v) => sub.next(fn(v)) }).unsubscribe);

export const filter = <A>(fn: (v: A) => boolean): Operator<A, A> => (source) =>
  new MiniObservable<A>((sub) => source.subscribe({ ...sub, next: (v) => fn(v) && sub.next(v) }).unsubscribe);

Let's match the pieces one by one. The constructor only stores the producer function; it does not run it: that is all laziness is. Every subscribe call sets up a fresh closed flag and a fresh subscriber object, then runs the producer: that is why every subscribe gets its own execution. The closed checks are the contract itself: after error or complete, no call reaches the consumer. close runs the teardown once; if the producer finished synchronously, the teardown is called as soon as its return value is received (the reason for the "teardown" line we saw with RxJS). An exception thrown by the producer is turned into an error notification by the try/catch.

Operators are simpler than you might think, too. map(fn) produces a function that takes an Observable and returns a new one. When you subscribe to the new Observable, it subscribes to the source, passes every value through fn downstream and returns the source's unsubscribe as its own teardown. That is how an unsubscribe at the bottom of the chain reaches the producer at the top. pipe is just a reduce that applies these functions in order. Let's try it:

import { MiniObservable, map, filter } from "./mini-observable.ts";

const stock$ = new MiniObservable<number>((subscriber) => {
  console.log("producer started");
  let stock = 5;
  const id = setInterval(() => {
    stock -= 1;
    subscriber.next(stock);
    if (stock === 0) subscriber.complete();
  }, 100);
  return () => {
    clearInterval(id);
    console.log("teardown");
  };
});

const lowStock$ = stock$.pipe(
  filter((s: number) => s < 3),
  map((s: number) => `only ${s} left`),
);

console.log("nothing has run yet");
lowStock$.subscribe({
  next: (msg) => console.log(msg),
  complete: () => console.log("complete"),
});

nothing has run yet
producer started
only 2 left
only 1 left
only 0 left
complete
teardown

Running the prediction code from the start of the post with MiniObservable (08-mini-predict.ts in the snapshot) gives the same order as RxJS: "producer started" twice, no 10, "after subscribe" last. An exception thrown in a producer lands in the error callback as well:

before subscribe
producer started
A got 12
A got 11
teardown
producer started
B got 12
B got 11
teardown
after subscribe
error: stock service unavailable

What RxJS adds to this model is clear, too: a Subscription object that collects many teardowns, more than a hundred well-tested operators, schedulers that manage time, type-safe pipe overloads (ours falls back to any) and reporting of uncaught errors. The last one is the topic of the next section.

In Production

Say Bookstore's API returned null as the price of a book on the cart page. We wrapped the producer in try/catch and wrote an error callback as well:

import { Observable } from "rxjs";

const cart$ = new Observable<{ title: string; price: number | null }>((subscriber) => {
  try {
    subscriber.next({ title: "Dune", price: 450 });
    subscriber.next({ title: "Emma", price: null }); // price missing in the API response
    subscriber.next({ title: "The Hobbit", price: 390 });
    subscriber.complete();
  } catch (err) {
    console.log("producer caught:", err); // never runs
    subscriber.error(err);
  }
});

console.log("before subscribe");
cart$.subscribe({
  next: (item) => console.log(item.title, item.price!.toFixed(2)),
  error: (err) => console.log("error callback:", err), // never runs
  complete: () => console.log("complete"),
});
console.log("after subscribe");

before subscribe
Dune 450.00
The Hobbit 390.00
complete
after subscribe
C:\Users\user\bookstore-rxjs\node_modules\rxjs\dist\cjs\internal\util\reportUnhandledError.js:13
            throw err;
            ^

TypeError: Cannot read properties of null (reading 'toFixed')
    at Object.next (file:///C:/Users/user/bookstore-rxjs/src/01/09-lost-error.ts:17:55)

The Emma line silently disappeared, the stream carried on with The Hobbit and completed; neither the producer's catch nor the error callback ran. RxJS 7 does not send an exception thrown in the consumer's next callback back to the producer; it catches it and rethrows it on the next macrotask. In Node that uncaught exception killed the process. When I ran the same code in Chromium, the page stayed alive and the console showed a single uncaught error:

[console.log] before subscribe
[console.log] Dune 450.00
[console.log] The Hobbit 390.00
[console.log] complete
[console.log] after subscribe
[pageerror] Cannot read properties of null (reading 'toFixed')
[console.log] page still alive

For the user the result is a missing book in the cart, a wrong total and an error that only exists in the console. Two practical rules follow. First, keep business logic out of the subscribe callback; do transformations in operators such as map. An exception thrown inside an operator becomes a proper error notification and can be caught with the error handling operators (the topic of this series' error handling post). Second, log the errors nobody catches in one central place with RxJS's config.onUnhandledError setting:

import { config, Observable } from "rxjs";

// one central place for errors that no subscriber handles
config.onUnhandledError = (err) => {
  console.log("[log] unhandled RxJS error:", (err as Error).message);
};

const price$ = new Observable<number>((subscriber) => {
  subscriber.error(new Error("price service returned 503"));
});

price$.subscribe((price) => console.log("price", price)); // no error callback
price$.subscribe({
  next: (price) => console.log("price", price),
  error: (err: Error) => console.log("handled here:", err.message),
});
console.log("after subscribe");

handled here: price service returned 503
after subscribe
[log] unhandled RxJS error: price service returned 503

The error of the first subscription, which has no error callback, was not lost, but it was still reported asynchronously (after "after subscribe"). In production, wiring this hook into Bookstore's logging makes the errors that "only live in the console" visible. On the measurement side, two numbers are worth tracking: how many requests go to the same URL when a page opens, in the browser's Network tab (a side effect of laziness), and whether memory and the number of timers grow on a page that stays open for a long time (a missing teardown).

Common Pitfalls

• Thinking of an Observable as "a request that runs once". Every subscribe runs the producer again; subscribing to the same HTTP Observable from two places means two requests.
• Assuming an Observable is always asynchronous. If the producer calls next synchronously, the values are delivered before subscribe returns; code that relies on "the line after subscribe runs first" breaks silently.
• Opening a resource inside new Observable without returning a teardown. No values on screen after unsubscribe does not mean the resource is closed; the interval, listener or connection stays open.
• Calling next after complete or error and expecting the value to arrive. By contract it is all dropped, without an error; checking subscriber.closed in the producer avoids useless work.
• Putting business logic in the subscribe callback. An exception thrown there does not reach the error callback; it is rethrown asynchronously while the rest of the stream continues.

A Small Experiment

★☆☆ Call getStock from 04-stock-request.ts three times for Emma (9780141439587) and subscribe to only one of the three Observables. How many requests do you see in the server log? Then subscribe to that one Observable once more inside a setTimeout one second later; predict how many requests will go out, then check the log.

★★☆ Add a take(n) operator to MiniObservable: it should pass on the first n values, then send complete and unsubscribe from the source. Try it with an interval producer like stockChanges and confirm that the "teardown" line appears right after complete. Hint: a synchronous source can reach n before its subscription has even been returned.

★★★ Change getStock so that it sends an error notification when the server answers 503 (check response.ok). Then set up config.onUnhandledError so that it also logs which ISBN the error was for. Note how you carry that information on the error object, and why it should be added in the producer rather than in the subscribe callback.

In this post we always built Observables by hand with new Observable. In real code you rarely do that; RxJS's creation functions such as of, from, defer, timer and fromEvent do the same job ready-made. The next post covers those functions and the difference between "cold" and "hot" Observables; using the server log, we will also measure why from(promise) is not lazy and why retrying a request needs defer.