Build Your Own Observable

You've probably seen RxJS mentioned in every Angular tutorial and a fair number of React ones too. The pitch is always "reactive streams", "compose async", "powerful operators". But Observables have a reputation for being hard to grasp because most explanations start with the abstraction rather than the mechanics.

Here's the thing: an Observable is just a function. Everything else — operators, subscriptions, teardown — is built on that one idea. Building one from scratch, with two operators and three creation helpers, takes about 120 lines. When you're done you'll understand why unsubscribing actually stops work, how operators chain without nesting callbacks, and what "lazy" really means.

What We're Building

An Observable wraps a producer function. When you subscribe, the producer runs and starts pushing values to an observer. The observer receives values via next(), gets notified of failures via error(), and hears when the stream is done via complete(). Unsubscribing tears the producer down.

Rendering diagram...

Data flows one direction: producer → observer. The subscription is the return ticket that lets you cancel.

Step 1: The Observable Core

The minimum viable Observable is almost embarrassingly simple.

Observable class with subscribe()

// observable.js
 
class Observable {
  constructor(subscribeFn) {
    this._subscribeFn = subscribeFn;
  }
 
  subscribe(observerOrNext, errorFn, completeFn) {
    // Accept either an observer object or three separate functions
    const observer =
      typeof observerOrNext === 'function'
        ? {
            next: observerOrNext,
            error: errorFn || (() => {}),
            complete: completeFn || (() => {}),
          }
        : {
            next: observerOrNext?.next?.bind(observerOrNext) || (() => {}),
            error: observerOrNext?.error?.bind(observerOrNext) || (() => {}),
            complete: observerOrNext?.complete?.bind(observerOrNext) || (() => {}),
          };
 
    // Run the producer, passing in the observer
    const teardown = this._subscribeFn(observer);
 
    return new Subscription(teardown);
  }
}

Two things worth pointing out:

subscribe() accepts either an observer object { next, error, complete } or three separate functions. Both are common in real code.
The producer runs synchronously when you call subscribe(). The Observable itself is lazy (the producer doesn't run until subscribe is called), but once you subscribe, it starts immediately.

⚠️

Lazy also means re-executed. The producer function runs once per subscribe() call, not once per Observable. Wrap a fetch in one of these, subscribe twice, and you've made two HTTP requests. This is what "cold" means, and it's the single most common surprise for people coming from Promises, which run once and cache their result. See hot vs cold at the end.

Step 2: Subscription and Teardown

Without teardown, Observables holding resources — timers, event listeners, WebSocket connections — would leak. The Subscription gives you unsubscribe().

Subscription with unsubscribe and safe observer

class Subscription {
  #closed = false;
  #teardownFn;
 
  constructor(teardownFn) {
    // Producers that complete synchronously return nothing — that's fine
    this.#teardownFn = typeof teardownFn === 'function' ? teardownFn : () => {};
  }
 
  unsubscribe() {
    if (!this.#closed) {
      this.#closed = true;
      this.#teardownFn();
    }
  }
 
  get closed() {
    return this.#closed;
  }
}

The #closed flag is load-bearing. Without it, calling unsubscribe() twice would double-clear a timer or double-remove a listener.

Now update subscribe() to do two things: stop delivering values once the stream ends, and run the teardown on every exit path. That second part is the one it's tempting to get wrong — if you only run teardown from unsubscribe(), then a producer that finishes on its own never gets cleaned up, and the timer or socket it opened stays open forever.

subscribe(observerOrNext, errorFn, completeFn) {
  let closed = false;
  let teardown;
  let teardownDone = false;
  let producerReturned = false;
 
  // Runs the producer's cleanup exactly once, however the stream ended
  const runTeardown = () => {
    if (teardownDone || !producerReturned) return;
    teardownDone = true;
    if (typeof teardown === 'function') teardown();
  };
 
  const rawObserver =
    typeof observerOrNext === 'function'
      ? { next: observerOrNext, error: errorFn, complete: completeFn }
      : observerOrNext ?? {};
 
  // Wrap observer to stop delivery once the stream ends or unsubscribes
  const safeObserver = {
    next(value) {
      if (!closed) rawObserver.next?.(value);
    },
    error(err) {
      if (!closed) {
        closed = true;
        rawObserver.error?.(err);
        runTeardown();
      }
    },
    complete() {
      if (!closed) {
        closed = true;
        rawObserver.complete?.();
        runTeardown();
      }
    },
  };
 
  teardown = this._subscribeFn(safeObserver);
  producerReturned = true;
  if (closed) runTeardown();   // producer terminated synchronously
 
  return new Subscription(() => {
    closed = true;
    runTeardown();
  });
}

The safeObserver wrapper enforces the Observable contract: once error or complete fires, or once you unsubscribe, no more values get through. The producer might keep running internally — you can't always stop it mid-flight — but the observer is shielded.

The producerReturned flag exists for an ordering problem that's easy to miss. A synchronous producer can call complete() before this._subscribeFn(...) has returned, which means teardown isn't assigned yet. So runTeardown() no-ops in that window, and the line right after the producer call retries it. Drop that flag and of(1, 2, 3) silently skips its own cleanup.

Step 3: Operators

Operators are where Observables earn their keep. An operator is a function that takes an Observable and returns a new Observable. The new Observable subscribes to the source and transforms values before forwarding them downstream.

Rendering diagram...

map() and filter() operators

function map(transformFn) {
  return function (sourceObservable) {
    return new Observable((observer) => {
      const subscription = sourceObservable.subscribe({
        next(value) {
          try {
            observer.next(transformFn(value));
          } catch (err) {
            observer.error(err);
          }
        },
        error(err) { observer.error(err); },
        complete() { observer.complete(); },
      });
      return () => subscription.unsubscribe();
    });
  };
}
 
function filter(predicateFn) {
  return function (sourceObservable) {
    return new Observable((observer) => {
      const subscription = sourceObservable.subscribe({
        next(value) {
          try {
            if (predicateFn(value)) observer.next(value);
          } catch (err) {
            observer.error(err);
          }
        },
        error(err) { observer.error(err); },
        complete() { observer.complete(); },
      });
      return () => subscription.unsubscribe();
    });
  };
}

The try/catch around the transform is essential. If transformFn throws (bad input, type mismatch), we route it to error() rather than letting it escape as an unhandled exception. Errors travel through the stream — they don't leak out of it.

Step 4: pipe()

Without pipe(), chaining operators reads inside-out:

filter(x => x > 0)(map(x => x * 2)(source$))

pipe() inverts this into natural left-to-right reading:

source$.pipe(
  map(x => x * 2),
  filter(x => x > 0)
)

Add pipe() to Observable

// Add to the Observable class
pipe(...operators) {
  return operators.reduce((obs, operatorFn) => operatorFn(obs), this);
}

reduce threads this through each operator left-to-right. Each operator receives the Observable produced by the previous one. The final result is a new composed Observable — nothing executes yet. Subscribe triggers the whole chain.

Step 5: Creation Utilities

Writing new Observable(fn) every time gets repetitive. Three helpers cover the most common patterns.

of(), fromEvent(), and interval()

// Emit a fixed set of values then complete
function of(...values) {
  return new Observable((observer) => {
    for (const value of values) {
      observer.next(value);
    }
    observer.complete();
    // Synchronous producer — nothing to tear down
  });
}
 
// Wrap a DOM or Node.js EventEmitter event
function fromEvent(target, eventName) {
  return new Observable((observer) => {
    const handler = (event) => observer.next(event);
 
    if (typeof target.addEventListener === 'function') {
      target.addEventListener(eventName, handler);
      return () => target.removeEventListener(eventName, handler);
    } else if (typeof target.on === 'function') {
      target.on(eventName, handler);
      return () => target.off(eventName, handler);
    }
  });
}
 
// Emit an incrementing counter on a timer
function interval(ms) {
  return new Observable((observer) => {
    let count = 0;
    const id = setInterval(() => observer.next(count++), ms);
    return () => clearInterval(id);
  });
}

fromEvent makes the teardown story concrete: the event listener removal lives right next to the registration. When you unsubscribe, the Subscription calls that cleanup function. This pattern generalises to anything — WebSockets, fetch with AbortController, child processes.

Full Implementation

Everything in one file:

// observable.js
 
class Subscription {
  #closed = false;
  #teardownFn;
 
  constructor(teardownFn) {
    this.#teardownFn = typeof teardownFn === 'function' ? teardownFn : () => {};
  }
 
  unsubscribe() {
    if (!this.#closed) {
      this.#closed = true;
      this.#teardownFn();
    }
  }
 
  get closed() { return this.#closed; }
}
 
class Observable {
  constructor(subscribeFn) {
    this._subscribeFn = subscribeFn;
  }
 
  subscribe(observerOrNext, errorFn, completeFn) {
    let closed = false;
    let teardown;
    let teardownDone = false;
    let producerReturned = false;
 
    const runTeardown = () => {
      if (teardownDone || !producerReturned) return;
      teardownDone = true;
      if (typeof teardown === 'function') teardown();
    };
 
    const rawObserver =
      typeof observerOrNext === 'function'
        ? { next: observerOrNext, error: errorFn, complete: completeFn }
        : observerOrNext ?? {};
 
    const safeObserver = {
      next(value) {
        if (!closed) rawObserver.next?.(value);
      },
      error(err) {
        if (!closed) { closed = true; rawObserver.error?.(err); runTeardown(); }
      },
      complete() {
        if (!closed) { closed = true; rawObserver.complete?.(); runTeardown(); }
      },
    };
 
    teardown = this._subscribeFn(safeObserver);
    producerReturned = true;
    if (closed) runTeardown();
 
    return new Subscription(() => {
      closed = true;
      runTeardown();
    });
  }
 
  pipe(...operators) {
    return operators.reduce((obs, op) => op(obs), this);
  }
}
 
function map(fn) {
  return (source) =>
    new Observable((observer) => {
      const sub = source.subscribe({
        next(v) { try { observer.next(fn(v)); } catch (e) { observer.error(e); } },
        error(e) { observer.error(e); },
        complete() { observer.complete(); },
      });
      return () => sub.unsubscribe();
    });
}
 
function filter(fn) {
  return (source) =>
    new Observable((observer) => {
      const sub = source.subscribe({
        next(v) { try { if (fn(v)) observer.next(v); } catch (e) { observer.error(e); } },
        error(e) { observer.error(e); },
        complete() { observer.complete(); },
      });
      return () => sub.unsubscribe();
    });
}
 
function of(...values) {
  return new Observable((observer) => {
    for (const v of values) observer.next(v);
    observer.complete();
  });
}
 
function fromEvent(target, eventName) {
  return new Observable((observer) => {
    const handler = (e) => observer.next(e);
    if (typeof target.addEventListener === 'function') {
      target.addEventListener(eventName, handler);
      return () => target.removeEventListener(eventName, handler);
    }
    target.on(eventName, handler);
    return () => target.off(eventName, handler);
  });
}
 
function interval(ms) {
  return new Observable((observer) => {
    let count = 0;
    const id = setInterval(() => observer.next(count++), ms);
    return () => clearInterval(id);
  });
}
 
module.exports = { Observable, Subscription, map, filter, of, fromEvent, interval };

Testing It Out

Run the demo

// demo.js
const { Observable, of, fromEvent, interval, map, filter } = require('./observable');
const { EventEmitter } = require('events');
 
// --- of() with pipe ---
of(1, 2, 3, 4, 5)
  .pipe(
    filter(x => x % 2 === 0),
    map(x => x * 10)
  )
  .subscribe({
    next: (v) => console.log('even*10:', v),   // 20, 40
    complete: () => console.log('done'),
  });
 
// --- interval() with unsubscribe ---
console.log('\nCounting for 350ms:');
const sub = interval(100).subscribe({
  next: (n) => console.log('tick:', n),        // 0, 1, 2
});
setTimeout(() => {
  sub.unsubscribe();
  console.log('stopped');
}, 350);
 
// --- fromEvent() with Node.js EventEmitter ---
const emitter = new EventEmitter();
const click$ = fromEvent(emitter, 'click').pipe(
  map(e => e.x * 2)
);
 
const clickSub = click$.subscribe({
  next: (v) => console.log('click x2:', v),
});
 
emitter.emit('click', { x: 5 });    // click x2: 10
emitter.emit('click', { x: 8 });    // click x2: 16
clickSub.unsubscribe();
emitter.emit('click', { x: 99 });   // (nothing — unsubscribed)
 
// --- error stops the stream ---
new Observable((observer) => {
  observer.next(1);
  observer.next(2);
  observer.error(new Error('stream exploded'));
  observer.next(3);  // never delivered
}).subscribe({
  next: (v) => console.log('value:', v),             // 1, 2
  error: (e) => console.log('caught:', e.message),   // caught: stream exploded
});
 
// --- lazy evaluation ---
const lazy$ = new Observable((observer) => {
  console.log('producer started');
  observer.next(42);
  observer.complete();
});
 
console.log('before subscribe');
lazy$.subscribe({ next: (v) => console.log('got:', v) });
// Output order: "before subscribe" → "producer started" → "got: 42"
// The producer function doesn't run until subscribe() is called.

Run with node demo.js. No dependencies.

What We Skipped

This covers the mechanics that make Observables work, but RxJS is a much larger surface:

Hot vs cold. Our implementation is "cold" — each subscriber gets an independent execution of the producer. A "hot" Observable shares a single execution across all subscribers (think a live WebSocket stream). Subject bridges the two.
Subject. A Subject is simultaneously an Observable and an observer — you can call next() on it directly, and multiple subscribers share the stream. The foundation for multicast patterns.
Higher-order operators. mergeMap, switchMap, concatMap handle Observables that emit other Observables. This is where most real-world async complexity lives — racing requests, cancelling inflight calls, sequencing dependent streams.
Error recovery. catchError, retry, retryWhen let you recover mid-stream without terminating. Our implementation just closes the stream on error.
Schedulers. RxJS lets you control when next calls are dispatched — synchronously, on microtask, on animation frame. Useful for testing and for controlling backpressure.
Unhandled errors. Call error() on a subscriber that passed no error handler and ours drops it on the floor — no throw, no log, nothing. RxJS reports it to a global handler instead, which is the behaviour you want in production. A console.error in the else branch is the cheap version.
Unsubscribing mid-emission. For a synchronous producer like of(1, 2, 3) you can't cancel from inside next(), because subscribe() hasn't returned the Subscription yet. RxJS avoids this by handing the producer a Subscriber that is the subscription, so closed is readable before the producer starts. Our two-object split can't express that.
A native Observable. The old TC39 Observable proposal has been dormant since 2019. The live effort is the WICG/WHATWG Observable API, which hangs when() off EventTarget — a web-platform API rather than a language primitive, but with the same subscribe/next/error/complete shape.

If you want to compare against the real thing, RxJS 7's Observable class in src/internal/Observable.ts is 498 lines, though roughly 180 of those are actual code and the rest is JSDoc. The extra bulk over what we built is mostly Subscriber, operator plumbing, and the error-reporting path.

Build Your Own Observable

Build Your Own Observable

What We're Building

Step 1: The Observable Core

Observable class with subscribe()

Step 2: Subscription and Teardown

Subscription with unsubscribe and safe observer

Step 3: Operators

map() and filter() operators

Step 4: pipe()

Add pipe() to Observable

Step 5: Creation Utilities

of(), fromEvent(), and interval()

Full Implementation

Testing It Out

Run the demo

What We Skipped

Comments (0)

Related Articles