The Observable Pattern in JavaScript

A Promise represents a single future value. An Observable represents a stream of values over time — zero, one, or many values, delivered synchronously or asynchronously. While JavaScript does not have a native Observable class yet (the TC39 proposal is at Stage 1), the pattern is straightforward to implement and is the foundation behind libraries like RxJS.

Observables vs. Promises

Feature	Promise	Observable
Values	Single	Multiple
Execution	Eager (runs immediately)	Lazy (runs on subscribe)
Cancellation	Not built-in	Supported via unsubscribe
Composition	.then() chains	Operators (map, filter, merge)

A Promise starts executing as soon as you create it. An Observable does nothing until someone subscribes.

Building a minimal Observable

An Observable is a function that accepts an observer (an object with next, error, and complete methods) and returns an unsubscribe function.

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

  subscribe(observer) {
    // Normalize: allow passing a plain function as shorthand
    const obs = typeof observer === "function"
      ? { next: observer, error: () => {}, complete: () => {} }
      : { next: observer.next || (() => {}),
          error: observer.error || (() => {}),
          complete: observer.complete || (() => {}) };

    const teardown = this._subscribe(obs);

    return {
      unsubscribe: typeof teardown === "function" ? teardown : () => {},
    };
  }
}

Creating observables

// From an array (synchronous)
function fromArray(arr) {
  return new Observable((observer) => {
    for (const item of arr) {
      observer.next(item);
    }
    observer.complete();
  });
}

// From a DOM event (asynchronous, ongoing)
function fromEvent(element, eventName) {
  return new Observable((observer) => {
    const handler = (e) => observer.next(e);
    element.addEventListener(eventName, handler);

    // Return cleanup function
    return () => element.removeEventListener(eventName, handler);
  });
}

// From a timer interval
function interval(ms) {
  return new Observable((observer) => {
    let count = 0;
    const id = setInterval(() => observer.next(count++), ms);
    return () => clearInterval(id);
  });
}

Subscribing and unsubscribing

const clicks = fromEvent(document, "click");

const subscription = clicks.subscribe({
  next: (e) => console.log("Clicked at", e.clientX, e.clientY),
});

// Later: stop listening
setTimeout(() => subscription.unsubscribe(), 5000);

The key benefit: the Observable encapsulates both the setup and the teardown. The subscriber does not need to know how cleanup works.

Adding operators

Operators transform one Observable into another. They are methods on the Observable class that return new Observables.

map

Observable.prototype.map = function (transformFn) {
  const source = this;
  return new Observable((observer) => {
    return source.subscribe({
      next: (value) => observer.next(transformFn(value)),
      error: (err) => observer.error(err),
      complete: () => observer.complete(),
    }).unsubscribe;
  });
};

filter

Observable.prototype.filter = function (predicateFn) {
  const source = this;
  return new Observable((observer) => {
    return source.subscribe({
      next: (value) => {
        if (predicateFn(value)) observer.next(value);
      },
      error: (err) => observer.error(err),
      complete: () => observer.complete(),
    }).unsubscribe;
  });
};

take

Observable.prototype.take = function (count) {
  const source = this;
  return new Observable((observer) => {
    let taken = 0;
    const sub = source.subscribe({
      next: (value) => {
        observer.next(value);
        taken++;
        if (taken >= count) {
          observer.complete();
          sub.unsubscribe();
        }
      },
      error: (err) => observer.error(err),
      complete: () => observer.complete(),
    });
    return () => sub.unsubscribe();
  });
};

Chaining operators

const numbers = fromArray([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);

numbers
  .filter((n) => n % 2 === 0)
  .map((n) => n * 10)
  .take(3)
  .subscribe({
    next: (v) => console.log(v),   // 20, 40, 60
    complete: () => console.log("Done"),
  });

Combining observables

merge

Merge interleaves values from multiple observables into a single stream.

function merge(...observables) {
  return new Observable((observer) => {
    let completedCount = 0;
    const subscriptions = observables.map((obs) =>
      obs.subscribe({
        next: (value) => observer.next(value),
        error: (err) => observer.error(err),
        complete: () => {
          completedCount++;
          if (completedCount === observables.length) {
            observer.complete();
          }
        },
      })
    );

    return () => subscriptions.forEach((s) => s.unsubscribe());
  });
}

const clicks = fromEvent(document, "click").map(() => "click");
const keys = fromEvent(document, "keydown").map((e) => `key: ${e.key}`);

merge(clicks, keys).subscribe((v) => console.log(v));
// Logs "click" or "key: a" etc. as events occur

Practical example: debounced search

This example combines several operators to build a real-time search input.

Observable.prototype.debounce = function (ms) {
  const source = this;
  return new Observable((observer) => {
    let timeoutId;
    const sub = source.subscribe({
      next: (value) => {
        clearTimeout(timeoutId);
        timeoutId = setTimeout(() => observer.next(value), ms);
      },
      error: (err) => observer.error(err),
      complete: () => observer.complete(),
    });
    return () => {
      clearTimeout(timeoutId);
      sub.unsubscribe();
    };
  });
};

Observable.prototype.distinctUntilChanged = function () {
  const source = this;
  return new Observable((observer) => {
    let last;
    let hasLast = false;
    return source.subscribe({
      next: (value) => {
        if (!hasLast || value !== last) {
          last = value;
          hasLast = true;
          observer.next(value);
        }
      },
      error: (err) => observer.error(err),
      complete: () => observer.complete(),
    }).unsubscribe;
  });
};

// Wire it up
const searchInput = document.querySelector("#search");

fromEvent(searchInput, "input")
  .map((e) => e.target.value.trim())
  .filter((query) => query.length >= 2)
  .debounce(300)
  .distinctUntilChanged()
  .subscribe(async (query) => {
    const response = await fetch(`/api/search?q=${encodeURIComponent(query)}`);
    const results = await response.json();
    renderResults(results);
  });

This pipeline avoids redundant API calls by debouncing keystrokes, filtering short queries, and skipping duplicate values.

The EventTarget-based approach

If you prefer using built-in browser APIs, EventTarget can serve as a lightweight observable.

class DataStream extends EventTarget {
  emit(value) {
    this.dispatchEvent(new CustomEvent("data", { detail: value }));
  }

  pipe(transformFn) {
    const output = new DataStream();
    this.addEventListener("data", (e) => {
      const result = transformFn(e.detail);
      if (result !== undefined) output.emit(result);
    });
    return output;
  }
}

const stream = new DataStream();

stream
  .pipe((v) => (v > 5 ? v : undefined))
  .pipe((v) => v * 2)
  .addEventListener("data", (e) => console.log(e.detail));

stream.emit(3);  // Filtered out
stream.emit(7);  // Logs 14
stream.emit(10); // Logs 20

When to use Observables

Observables are well-suited for:

• DOM event streams — clicks, keyboard input, scroll, resize
• WebSocket messages — continuous data from a server
• Polling — periodic fetch calls that return values over time
• Animations — frame-by-frame updates

For single async values, Promises remain simpler and more idiomatic. For complex event processing with many operators, consider a battle-tested library like RxJS rather than rolling your own.

Summary

The Observable pattern provides a unified model for working with streams of data:

• An Observable is lazy — it only runs when subscribed to.
• Observers receive values via next, errors via error, and a completion signal via complete.
• Operators like map, filter, debounce, and merge compose into declarative data pipelines.
• Unsubscription handles cleanup automatically, preventing memory leaks.
• The pattern works for synchronous data, DOM events, timers, and any async source.

Understanding Observables gives you a mental model for reactive programming that transfers directly to RxJS, Angular, and other reactive frameworks.