ReplaySubject

ReplaySubject is a type of Subject that maintains a buffer of previously emitted values from a source observable and delivers those stored values to every new subscriber immediately upon subscription. This ability to replay a stream of past values to late subscribers is what gives this subject its name. When an observer subscribes to a ReplaySubject, it first receives all buffered values, and thereafter continues to receive fresh emissions from the source observable as they occur. Even if a subscription is made long after the values were originally emitted, the ReplaySubject will still deliver the entire cached sequence.

In many ways, ReplaySubject resembles the BehaviorSubject, as both can provide previously recorded values to new subscribers. The key distinction is that a BehaviorSubject can only remember a single, most recent value, whereas a ReplaySubject can record and re-emit a full history of values.

Another significant difference emerges when the source observable terminates. Once a BehaviorSubject receives a complete or error notification and enters a stopped state, all future subscribers will only see that terminal notification, without ever receiving the cached value.

On the other hand, a ReplaySubject in a stopped state — whether due to completion or an error — will still replay its buffered values to new subscribers before sending them the complete or error notification.

When instantiating a ReplaySubject, you have two configuration parameters: bufferSize, which determines the maximum number of values to retain, and windowTime, which specifies the maximum age (in milliseconds) a value can have before it's discarded from the buffer. These settings can be used together.

For example, to keep a maximum of three values, provided they are no older than two seconds, you would use new ReplaySubject(3, 2000). The windowTime parameter can also be seen as the time span prior to a new subscription. In essence, the above configuration translates to “buffer the last three values that were emitted within two seconds before the subscription moment”.

The internal operation of ReplaySubject proceeds as follows:

  1. It establishes an internal registry for subscriptions.
  2. Upon a new subscription, it adds the observer to the registry and, if there are buffered values, it immediately replays them to that observer.
  3. Whenever the source emits or when next is invoked on the subject, it adds the new value to the buffer (evicting older ones if the buffer is full) and pushes the value to all registered observers.
  4. Should the source complete or when complete is called, the subject transitions to the stopped state, stores the completion notice in the buffer, notifies all current observers, and clears them from the registry.
  5. If an error occurs in the source or when error is called, the subject moves to stopped, caches the error notification, delivers it to all observers, and removes them from the registry.
  6. In the stopped state, new subscribers are not added to the registry; instead, the buffered values along with the terminal notification are immediately replayed to them.
  7. If a stopped subject is subscribed to a fresh source observable, any emissions from that source are simply ignored.

Usage

ReplaySubject is a practical choice for scenarios where you need to replay an event or a sequence of events. Because ReplaySubject does not require a default value (unlike BehaviorSubject), it is particularly useful for events that might never happen at all.

Consider a situation where you lazy-load a library that needs to process user interactions. Some events are likely to fire before the library finishes loading. In that case, you can capture these events in a ReplaySubject first, and then let the library subscribe to that subject once it's ready — the library will receive all prior events.

The following code snippet demonstrates this pattern:

const events = setUpListeners();
emulateLibraryLoad(events);

function emulateLibraryLoad(events) {
   setTimeout(() => {
       events.subscribe((event) => console.log(event));
   }, 3000);
}

function setUpListeners() {
   const events = new ReplaySubject();

   const clicks = fromEvent(document, 'click');
   const spacebars = fromEvent(document, 'keyup').pipe(filter((event: any) => event.code === 'Space'));

   merge(clicks, spacebars).pipe(
       tap((event) => events.next(event))
   ).subscribe();

   return events.asObservable();
}

Playground

Additional resources

See also