AsyncSubject Overview

AsyncSubject is a special type of Subject that retains the final value emitted by an observable prior to its termination and delivers it to every subscriber that comes after. It waits for the source to finish before determining which value counts as the last one, and only then does it forward that value to any current or pending observers.

Thanks to this design, subscribers can always retrieve the most recent value from the AsyncSubject without worrying about timing, even if they attach well after the value was originally produced.

There is a conceptual resemblance between AsyncSubject and a Promise. The key distinction lies in eagerness: a Promise executes its provided function right away. In contrast, AsyncSubject leaves the decision to subscribe in your hands, and because observables are inherently lazy, the producer logic inside the source observable runs only upon subscription.

When the source observable encounters an error, the AsyncSubject does not surface the last stored value. Rather, it forwards the error notification coming from the source to all subscribers, existing and new alike.

The inner mechanics of AsyncSubject are straightforward:

  1. Maintain an internal registry for subscriptions
  2. Upon each new subscription, append it to the registry
  3. When a value arrives (via next on the subject or from the source observable), replace any previously stored value with this new one
  4. When completion happens (via complete on the subject or from the source), mark the subject as stopped, then dispatch the latest stored value together with the completion notification to every current subscription; afterwards, purge the registry
  5. Should an error occur (via error on the subject or from the source), mark the subject as stopped and keep the error notification as the relevant payload; deliver the error notification to all current subscriptions, then purge the registry
  6. Once the subject is in a stopped state, no new subscriptions are added to the registry. Instead, if no error was encountered, the latest value plus completion is sent to the new observer immediately; otherwise only the error notification is passed on, without the last value
  7. If a stopped subject gets subscribed to a fresh source observable, all incoming values from that new source are disregarded

When to Reach for It

AsyncSubject fits perfectly when you need to fetch and cache resources that are loaded once. In a typical network call, what matters is the final response; to obtain it, you must wait until the request finishes loading, which corresponds to the moment an observable stream closes.

The following snippet illustrates this use case:

const cache = {};

function getResource(url) {
   if (!cache[url]) {
       cache[url] = new AsyncSubject();
       fetch(url)
           .then((response) => response.json())
           .then((data) => {
               cache[url].next(data);
               cache[url].complete();
           });
   }

   return cache[url].asObservable();
}

const url = 'https://api.mocki.io/v1/ce5f60e2';

getResource(url).subscribe((data) => console.log(data));

setTimeout(() => {
   // no request is made, data is served from the AsyncSubject's cache
   getResource(url).subscribe((data) => console.log(data));
}, 3000);

Interactive Demo

Further Reading