Solution

You'll build a small class around one idea: an observable is its subscribe function. Everything — of, from, map, filter, cancellation — is a thin layer over calling that function with an observer and handing back a teardown.

The problem, in plain English

A promise gives you one future value. An observable gives you a stream: zero, one, or many values pushed over time, then a terminal complete or error. You consume it by subscribing with an observer, and you stop by unsubscribing. Two properties make it powerful: it's lazy (the producer doesn't run until you subscribe) and cancellable (unsubscribing tears the producer down, so an interval or listener is cleaned up). You're building a minimal version.

The mental model

The constructor just stores a subscribeFn. subscribe(observer) calls it, passing a safe observer that (a) tolerates a missing next/error/complete, (b) refuses to emit after a terminal or unsubscribe, and (c) returns a subscription whose unsubscribe runs the teardown that subscribeFn returned. of/from are constructors that emit then complete. map/filter are new observables whose subscribeFn subscribes to the source and forwards transformed/filtered values.

A first attempt

The naive version calls the producer but forgets cancellation:

class ObservableNaive {
  constructor(subscribeFn) { this.subscribeFn = subscribeFn; }
  subscribe(onNext) {
    this.subscribeFn({ next: onNext });
    // returns nothing — no way to stop!
  }
}

For a finite source this seems fine, but it's missing the half of the model that makes observables useful. There's no unsubscribe, so a source like setInterval(() => next(n++), 1000) runs forever — you've leaked a timer with no way to clear it. It also delivers values after the consumer would want to stop, and doesn't guard complete/error. Cancellation and the terminal guard aren't extras; they're the point.

The working solution

class Observable {
  constructor(subscribeFn) {
    this._subscribeFn = subscribeFn;
  }

  subscribe(observer) {
    // Normalize: a bare function is the `next` handler.
    const o = typeof observer === 'function' ? { next: observer } : observer || {};

    let closed = false;
    let teardown;
    const cleanup = () => {
      if (typeof teardown === 'function') teardown();
    };

    // A safe observer: no emissions after a terminal / unsubscribe.
    const safe = {
      next: (v) => { if (!closed && o.next) o.next(v); },
      error: (e) => {
        if (closed) return;
        closed = true;
        if (o.error) o.error(e);
        cleanup();
      },
      complete: () => {
        if (closed) return;
        closed = true;
        if (o.complete) o.complete();
        cleanup();
      },
    };

    teardown = this._subscribeFn(safe); // run the producer

    return {
      unsubscribe: () => {
        if (closed) return;
        closed = true;
        cleanup();
      },
    };
  }

  map(fn) {
    // A new observable that subscribes to the source and transforms values.
    return new Observable((observer) => {
      const sub = this.subscribe({
        next: (v) => observer.next(fn(v)),
        error: (e) => observer.error(e),
        complete: () => observer.complete(),
      });
      return () => sub.unsubscribe(); // teardown chains upstream
    });
  }

  filter(predicate) {
    return new Observable((observer) => {
      const sub = this.subscribe({
        next: (v) => { if (predicate(v)) observer.next(v); },
        error: (e) => observer.error(e),
        complete: () => observer.complete(),
      });
      return () => sub.unsubscribe();
    });
  }

  static of(...values) {
    return new Observable((observer) => {
      for (const v of values) observer.next(v);
      observer.complete();
    });
  }

  static from(iterable) {
    return new Observable((observer) => {
      for (const v of iterable) observer.next(v);
      observer.complete();
    });
  }
}

module.exports = { Observable };

subscribe is the core. It wraps the caller's observer in a safe one gated by a closed flag, so once error, complete, or unsubscribe flips it, no further next gets through — and a terminal auto-runs cleanup. The producer runs when we call this._subscribeFn(safe), and whatever it returns becomes the teardown; the returned subscription's unsubscribe flips closed and runs it. Because subscribeFn runs inside subscribe, every subscriber triggers a fresh run — that's the "cold" behavior. of/from are producers that push values then complete. map/filter return a new Observable whose producer subscribes to this, transforming (map) or gating (filter) each value before forwarding it, and whose teardown unsubscribes upstream — so cancellation propagates all the way back through the pipeline.

Walking through one call

Take Observable.of(1, 2).map((x) => x * 10).subscribe((v) => log(v)):

• map builds a new Observable whose producer, when subscribed, will subscribe to of(1, 2) with a transforming observer.
• the outer subscribe wraps log as safe.next and runs the map-observable's producer.
• that producer subscribes to of(1, 2) — which, being cold, immediately runs its producer: observer.next(1), observer.next(2), observer.complete().
  • next(1) → map's next → observer.next(1 * 10) → outer safe.next(10) → log(10).
  • next(2) → log(20).
  • complete() → propagates through map to the outer observer; closed flips.
• result — 10, then 20, then completion. If the source were an interval instead, calling sub.unsubscribe() would set closed and run the teardown chain down to the clearInterval.

Each value flowed source → map → subscriber, transformed on the way, and the whole thing was lazy — nothing ran until the final subscribe.

Gotchas

• No teardown = a leak — without returning unsubscribe (and honoring the producer's teardown), an interval- or listener-based source runs forever. Cancellation is half the model.
• Guard the terminal — after complete/error/unsubscribe, next must be a no-op. A closed flag in the safe observer enforces "at most one terminal, nothing after."
• Cold, not hot — run subscribeFn inside subscribe so each subscriber is independent. Running it once in the constructor and sharing makes it hot (a different, shared semantics).
• Operators return new observables — map/filter must not mutate or subscribe to the source at creation time. They build a new Observable that subscribes when it is subscribed.

Going further

• More operators — take(n), merge, switchMap, debounceTime all follow the same shape: a new observable that subscribes to the source and manages what it forwards. switchMap (cancel the previous inner subscription on each new value) is the one worth studying.
• Hot observables & Subjects — a Subject is both an observable and an observer, multicasting one run to many subscribers — the bridge from cold to hot, and how event buses are built.
• The TC39 proposal / RxJS — this mirrors the Observable proposal and the heart of RxJS. Backpressure (a consumer that can't keep up) is the hard problem those libraries and pull-based streams tackle next.