Observable

This is the explainer for the Observable API proposal for more ergonomic and composable event handling.

Introduction

EventTarget integration

This proposal adds an .on() method to EventTarget that becomes a better addEventListener(); specifically it returns a new Observable that adds a new event listener to the target when its subscribe() method is called. The Observable calls the subscriber's next() handler with each event.

Observables turn event handling, filtering, and termination, into an explicit, declarative flow that's easier to understand and compose than today's imperative version, which often requires nested calls to addEventListener() and hard-to-follow callback chains.

Example 1

// Filtering and mapping:
element.on('click')
  .filter(e => e.target.matches('.foo'))
  .map(e => ({x: e.clientX, y: e.clientY }))
  .subscribe({next: handleClickAtPoint});

Example 2

// Automatic, declarative unsubscription via the takeUntil method:
element.on('mousemove')
  .takeUntil(document.on('mouseup'))
  .subscribe({next: e => … });

// Since reduce and some other terminators return promises, they also play
// well with async functions:
await element.on('mousemove')
  .takeUntil(element.on('mouseup'))
  .reduce((soFar, e) => …);

// Imperative
const controller = new AbortController();
element.addEventListener('mousemove', e => {
  element.addEventListener('mouseup', e => controller.abort());
  console.log(e);
}, {signal});

Example 3

Tracking all link clicks within a container (example):

container.on('click').filter(e => e.target.closest('a')).subscribe({next: e => {
  // …
}});

Example 4

Find the maximum Y coordinate while the mouse is held down (example):

const maxY = await element.on('mousemove')
                          .takeUntil(element.on('mouseup'))
                          .map(e => e.clientY)
                          .reduce((soFar, y) => Math.max(soFar, y), 0);

Example 5

Multiplexing a WebSocket, such that a subscription message is send on connection, and an unsubscription message is send to the server when the user unsubscribes.

const socket = new WebSocket('wss://example.com');

function multiplex({ startMsg, stopMsg, match }) {
  if (socket.readyState !== WebSocket.OPEN) {
    return socket
      .on('open')
      .flatMap(() => multiplex({ startMsg, stopMsg, match }));
  } else {
    socket.send(JSON.stringify(startMsg));
    return socket
      .on('message')
      .filter(match)
      .takeUntil(socket.on('close'))
      .takeUntil(socket.on('error'))
      .map((e) => JSON.parse(e.data))
      .finally(() => {
        socket.send(JSON.stringify(stopMsg));
      });
  }
}

function streamStock(ticker) {
  return multiplex({
    startMsg: { ticker, type: 'sub' },
    stopMsg: { ticker, type: 'unsub' },
    match: (data) => data.ticker === ticker,
  });
}

const googTrades = streamStock('GOOG');
const nflxTrades = streamStock('NFLX');

const googController = new AbortController();
const googSubscription = googTrades.subscribe({next: updateView, signal: googController.signal});
const nflxSubscription = nflxTrades.subscribe({next: updateView, ...});

// And the stream can disconnect later, which
// automatically sends the unsubscription message
// to the server.
googController.abort();

// Imperative
function multiplex({ startMsg, stopMsg, match }) {
  const start = (callback) => {
    const teardowns = [];

    if (socket.readyState !== WebSocket.OPEN) {
      const openHandler = () => start({ startMsg, stopMsg, match })(callback);
      socket.addEventListener('open', openHandler);
      teardowns.push(() => {
        socket.removeEventListener('open', openHandler);
      });
    } else {
      socket.send(JSON.stringify(startMsg));
      const messageHandler = (e) => {
        const data = JSON.parse(e.data);
        if (match(data)) {
          callback(data);
        }
      };
      socket.addEventListener('message', messageHandler);
      teardowns.push(() => {
        socket.send(JSON.stringify(stopMsg));
        socket.removeEventListener('message', messageHandler);
      });
    }

    const finalize = () => {
      teardowns.forEach((t) => t());
    };

    socket.addEventListener('close', finalize);
    teardowns.push(() => socket.removeEventListener('close', finalize));
    socket.addEventListener('error', finalize);
    teardowns.push(() => socket.removeEventListener('error', finalize));

    return finalize;
  };

  return start;
}

function streamStock(ticker) {
  return multiplex({
    startMsg: { ticker, type: 'sub' },
    stopMsg: { ticker, type: 'unsub' },
    match: (data) => data.ticker === ticker,
  });
}

const googTrades = streamStock('GOOG');
const nflxTrades = streamStock('NFLX');

const unsubGoogTrades = googTrades(updateView);
const unsubNflxTrades = nflxTrades(updateView);

// And the stream can disconnect later, which
// automatically sends the unsubscription message
// to the server.
unsubGoogTrades();

Example 6

Here we're leveraging observables to match a secret code, which is a pattern of keys the user might hit while using an app:

const pattern = [
  'ArrowUp',
  'ArrowUp',
  'ArrowDown',
  'ArrowDown',
  'ArrowLeft',
  'ArrowRight',
  'ArrowLeft',
  'ArrowRight',
  'b',
  'a',
  'b',
  'a',
  'Enter',
];

const keys = document.on('keydown').map((e) => e.key);
keys
  .flatMap((firstKey) => {
    if (firstKey === pattern[0]) {
      return keys
        .take(pattern.length - 1)
        .every((k, i) => k === pattern[i + 1]);
    }
  })
  .filter(matched => matched)
  .subscribe({next: _ => {
    console.log('Secret code matched!');
  }});

const pattern = [...];

// Imperative
document.addEventListener('keydown', e => {
  const key = e.key;
  if (key === pattern[0]) {
    let i = 1;
    const handler = (e) => {
      const nextKey = e.key;
      if (nextKey !== pattern[i++]) {
        document.removeEventListener('keydown', handler)
      } else if (pattern.length === i) {
        console.log('Secret code matched!');
        document.removeEventListener('keydown', handler)
      }
    }
    document.addEventListener('keydown', handler)
  }
})

The Observable API

Observables are first-class objects representing composable, repeated events. They're like Promises but for multiple events, and specifically with EventTarget integration, they are to events what Promises are to callbacks. They can be:

• Created by script or by platform APIs, and passed to anyone interested in consuming events via subscribe()
• Fed to operators like Observable.map(), to be composed & transformed without a web of nested callbacks

Better yet, the transition from event handlers ➡️ Observables is simpler than that of callbacks ➡️ Promises, since Observables integrate nicely on top of EventTarget, the de facto way of subscribing to events from the platform and custom script. As a result, developers can use Observables without migrating tons of code on the platform, since it's an easy drop-in wherever you're handling events today.

The proposed API shape is as follows:

partial interface EventTarget {
  Observable on(DOMString type, optional AddEventListenerOptions options);
};

// `SubscribeCallback` is where the Observable "creator's" code lives. It's
// called when `subscribe()` is called, to set up a new subscription.
callback SubscribeCallback = undefined (Subscriber subscriber);
callback ObserverCallback = undefined (any value);

dictionary Observer {
  ObserverCallback next;
  VoidFunction complete;
  ObserverCallback error;

  AbortSignal signal;
};

dictionary PromiseOptions {
  AbortSignal signal;
};

[Exposed=*]
interface Subscriber {
  undefined next(any result);
  undefined complete();
  undefined error(any error);

  readonly attribute AbortSignal signal;
};

callback Predicate = boolean (any value);
callback Reducer = any (any accumulator, any currentValue)
callback Mapper = any (any element, unsigned long long index)
// Differs from `Mapper` only in return type, since this callback is exclusively
// used to visit each element in a sequence, not transform it.
callback Visitor = undefined (any element, unsigned long long index)

[Exposed=*]
interface Observable {
  constructor(SubscribeCallback callback);
  undefined subscribe(Observer observer);

  undefined finally(VoidFunction callback);

  // Observable-returning operators. See "Operators" section below.
  Observable takeUntil(Observable notifier);
  Observable map(Mapper mapper);
  Observable filter(Predicate predicate);
  Observable take(unsigned long long);
  Observable drop(unsigned long long);
  Observable flatMap(Mapper mapper);
  Promise<sequence<any>> toArray(optional PromiseOptions options);
  Promise<undefined> forEach(Visitor callback, optional PromiseOptions options);

  // Promise-returning. See "Concerns" section below.
  Promise<any> every(Predicate predicate, optional PromiseOptions options);
  // Maybe? Promise<any> first(optional PromiseOptions options);
  Promise<any> find(Predicate predicate, optional PromiseOptions options);
  Promise<any> some(Predicate predicate, optional PromiseOptions options);
  Promise<any> reduce(Reducer reducer, optional any initialValue, optional PromiseOptions options);
};

The creator of an Observable passes in a callback that gets invoked synchronously whenever subscribe() is called. The subscribe() method can be called any number of times, and the callback it invokes sets up a new "subscription" by registering the caller of subscribe() as a Observer. With this in place, the Observable can signal any number of events to the Observer via the next() callback, optionally followed by a single call to either complete() or error(), signaling that the stream of data is finished.

const observable = new Observable(subscriber => {
  let i = 0;
  setInterval(() => {
    if (i >= 10)
      subscriber.complete();
    else
      subscriber.next(i++);
  }, 2000);
});

observable.subscribe({
  // Print each value the Observable produces.
  next: console.log
});

Issue: See #3 about having the Observable constructor being able to register teardown upon unsubscription.

While custom Observables can be useful on their own, the primary use case they unlock is with event handling. Observables returned by the new EventTarget#on() method are created natively with an internal callback that uses the same underlying mechanism as addEventListener(). Therefore calling subscribe() essentially registers a new event listener whose events are exposed through the Observer handler functions and are composable with the various combinators available to all Observables.

Lazy, synchronous delivery

Crucially, Observables are "lazy" in that they do not start emitting data until they are subscribed to, nor do they queue any data before subscription. They can also start emitting data synchronously during subscription, unlike Promises which always queue microtasks when invoking .then() handlers. Consider this example:

el.on('click').subscribe({next: () => console.log('One')});
el.on('click').find(() => {…}).then(() => console.log('Three'));
el.click();
console.log('Two');
// Logs "One" "Two" "Three"

Firehose of synchronous data

By using AbortController, you can unsubscribe from an Observable even as it synchronously emits data during subscription:

// An observable that synchronously emits unlimited data during subscription.
let observable = new Observable(subscriber => {
  let i = 0;
  while (true) {
    subscriber.next(i++);
  }
});

let controller = new AbortController();
observable.subscribe({next: data => {
  if (data > 100)
    controller.abort();
}, signal: controller.signal});

Operators

We propose the following operators in addition to the Observable interface:

• takeUntil(Observable)
  • Returns an observable that mirrors the one that this method is called on, until the input observable emits its first value
• finally()
  • Like Promise.finally(), it takes a callback which gets fired after the observable completes in any way (complete()/error())

Versions of the above are often present in userland implementations of observables as they are useful for observable-specific reasons, but in addition to these we offer a set of common operators that follow existing platform precedent and can greatly increase utility and adoption. These exist on other iterables, and are derived from TC39's iterator helpers proposal which adds the following methods to Iterator.prototype:

• map()
• filter()
• take()
• drop()
• flatMap()
• reduce()
• toArray()
• forEach()
• some()
• every()
• find()
• maybe: from()

We expect userland libraries to provide more niche operators that integrate with the Observable API central to this proposal, potentially shipping natively if they get enough momentum to graduate to the platform. But for this initial proposal, we'd like to restrict the set of operators to those that follow the precedent stated above, similar to how web platform APIs that are declared Setlike and Maplike have native properties inspired by TC39's Map and Set objects. Therefore we'd consider most discussion of expanding this set as out-of-scope for the initial proposal, suitable for discussion in an appendix. Any long tail of operators could conceivably follow along if there is support for the native Observable API presented in this explainer.

Note that the operators every(), find(), some(), and reduce() return Promises whose scheduling differs from that of Observables, which sometimes means event handlers that call e.preventDefault() will run too late. See the Concerns section which goes into more detail.

Background & landscape

To illustrate how Observables fit into the current landscape of other reactive primitives, see the below table which is an attempt at combining two other tables that classify reactive primitives by their interaction with producers & consumers:

History

Observables were first proposed to the platform in TC39 in May of 2015. The proposal failed to gain traction, in part due to some opposition that the API was suitable to be a language-level primitive. In an attempt to renew the proposal at a higher level of abstraction, a WHATWG DOM issue was filed in December of 2017. Despite ample developer demand, lots of discussion, and no strong objectors, the DOM Observables proposal sat mostly still for several years (with some flux in the API design) due to a lack of implementer prioritization.

Later in 2019, an attempt at reviving the proposal was made back at the original TC39 repository, which involved some API simplifications and added support for the synchronous "firehose" problem.

This repository is an attempt to again breathe life into the Observable proposal with the hope of shipping a version of it to the Web Platform.

Userland libraries

In prior discussion, Ben Lesh has listed several custom userland implementations of observable primitives, of which RxJS is the most popular with "47,000,000+ downloads per week."

• RxJS: Started as a reference implementation of the TC39 proposal, is nearly identical to this proposal's observable.
• Relay: A mostly identical contract with the addition of start and unsubscribe events for observation and acquiring the Subscription prior to the return.
• tRPC: A nearly identical implemention of observable to this proposal.
• XState: uses an observable interface in several places in their library, in particular for their Actor type, to allow subscriptions to changes in state, as shown in their useActor hook. Using an identical observable is also a documented part of access state machine changes when using XState with SolidJS.
• SolidJS: An identical interface to this proposal is exposed for users to use.
• Apollo GraphQL: Actually re-exporting from zen-observable as their own thing, giving some freedom to reimplement on their own or pivot to something like RxJS observable at some point.
• zen-observable: A reference implementation of the TC39 observable proposal. Nearly identical to this proposal.
• React Router: Uses a { subscribe(callback: (value: T) => void): () => void } pattern in their Router and DeferredData code. This was pointed out by maintainers as being inspired by Observable.
• Preact Uses a { subscribe(callback: (value: T) => void): () => void } interface for their signals.
• TanStack: Uses a subscribable interface that matches { subscribe(callback: (value: T) => void): () => void } in several places
• Redux: Implements an observable that is nearly identical to this proposal's observable as a means of subscribing to changes to a store.
• Svelte: Supports subscribing to observables that fit this exact contract, and also exports and uses a subscribable contract for stores like { subscribe(callback: (value: T) => void): () => void }.
• Dexie.js: Has an observable implementation that is used for creating live queries to IndexedDB.
• MobX: Uses similar interface to Observable internally for observation: { observe_(callback: (value: T)): () => void }.

UI Frameworks Supporting Observables

• Svelte: Directly supports implicit subscription and unsubscription to observables simply by binding to them in templates.
• Angular: Directly supports implicit subscription and unsubscription to observables using their | async "async pipe" functionality in templates.
• Vue: maintains a dedicated library specifically for using Vue with RxJS observables.
• Cycle.js: A UI framework built entirely around observables

Given the extensive prior art in this area, there exists a public "Observable Contract".

Additionally many JavaScript APIs been trying to adhere to the contract defined by the TC39 proposal from 2015. To that end, there is a library, symbol-observable, that ponyfills (polyfills) Symbol.observable to help with interoperability between observable types that adheres to exactly the interface defined here. symbol-observable has 479 dependent packages on npm, and is downloaded more than 13,000,000 times per week. This means that there are a minimum of 479 packages on npm that are using the observable contract in some way.

This is similar to how Promises/A+ specification that was developed before Promises were adopted into ES2015 as a first-class language primitive.

Concerns

One of the main concerns expressed in the original WHATWG DOM thread has to do with Promise-ifying APIs on Observable, such as the proposed first(). The potential footgun here with microtask scheduling and event integration. Specifically, the following innocent-looking code would not always work:

element.on('click').first().then(e => {
  e.preventDefault();
  // Do something custom...
});

If Observable#first() returns a Promise that resolves when the first event is fired on an EventTarget, then the user-supplied Promise .then() handler will run:

• ✅ Synchronously after event firing, for events triggered by the user
• ❌ Asynchronously after event firing, for all events triggered by script (i.e., element.click())
  • This means e.preventDefault() will have happened too late and effectively been ignored

Two things mitigate this concern. First, there is a very simple workaround to always avoid the case where your e.preventDefault() might run too late:

element.on('click').map(e => (e.preventDefault(), e)).first()

...or if Observable had a .do() method (see whatwg/dom#544 (comment)):

element.on('click').do(e => e.preventDefault()).first()

...or by modifying the semantics of first() to take a callback that produces a value that the returned Promise resolves to:

el.on("submit").first(e => e.preventDefault()).then(doMoreStuff)

Second, this "quirk" already exists in today's thriving Observable ecosystem, and there are no serious concerns or reports from that community that developers are consistently running into this. This gives some confidence that baking this behavior into the web platform will not be dangerous.

Standards venue

There's been much discussion about which standards venue should ultimately host an Observables proposal. The venue is not inconsequential, as it effectively decides whether Observables becomes a language-level primitive like Promises, that ship in all JavaScript browser engines, or a web platform primitive with likely (but technically optional) consideration in other environments like Node.js (see AbortController for example).

Observables purposefully integrate frictionlessly with the main event-emitting interface (EventTarget) and cancellation primitive (AbortController) that live in the Web platform. As proposed here, observables join this existing strongly-connected component from the DOM Standard: Observables depend on AbortController/AbortSignal, which depend on EventTarget, and EventTarget depends on both Observables and AbortController/AbortSignal. Because we feel that Observables fits in best where its supporting primitives live, the WHATWG standards venue is probably the best place to advance this proposal. Additionally, non-Web ECMAScript embedders like Node.js and Deno would still be able to adopt Observables, and are even likely to, given their commitment to Web platform aborting and events.

This does not preclude future standardization of event-emitting and cancellation primitives in TC39 in the future, something Observables could theoretically be layered on top of later. But for now, we are motivated to make progress in WHATWG.

In attempt to avoid relitigating this discussion, we'd urge the reader to see the following discussion comments:

• whatwg/dom#544 (comment)
• whatwg/dom#544 (comment)
• whatwg/dom#544 (comment)
• whatwg/dom#544 (comment)
• whatwg/dom#544 (comment)

User needs

Observables are designed to make event handling more ergonomic and composable. As such, their impact on end users is indirect, largely coming in the form of users having to download less JavaScript to implement patterns that developers currently use third-party libraries for. As stated above in the explainer, there is a thriving userland Observables ecosystem which results in loads of excessive bytes being downloaded every day.

In an attempt to codify the strong userland precedent of the Observable API, this proposal would save dozens of custom implementations from being downloaded every day.

Additionally, as an API like EventTarget, AbortController, and one related to Promises, it enables developers to build less-complicated event handling flows by constructing them declaratively, which may enable them to build more sound user experiences on the Web.

Authors:

• Dominic Farolino
• Ben Lesh