RXJS Objervable, Subject, BehaviorSubject, ReplaySubject

Photo by Michael Dziedzic on Unsplash

RXJS is a library that serves as a tool for carrying out asynchronous programming in JavaScript, wherein Observable and Subject are two distinct categories within the RXJS ecosystem that are leveraged for asynchronous programming purposes.

Observable, as a construct, represents a given data flow and can potentially generate one or more values, which are subsequently dispatched to subscribers for consumption. Observable instances are classified as “cold,” connoting their inactivity in terms of value generation until such time as subscribers opt to subscribe to the Observable.

Subject, on the other hand, is a composite construct that can act in dual capacities, either as an Observable or as an Observer, and is capable of producing one or more values that are dispatched to its subscribers. The classification of Subject instances as “hot” relates to their generation of values even before subscribers have had the opportunity to subscribe to the Subject object.

Consider, for instance, the scenario where we have an apparatus designed to measure pressure, and this instrument is responsible for generating a data stream which we aim to process using RXJS. By means of an Observable construct, we can effect a transformation of the raw data emanating from the pressure measuring device into an Observable object, the values of which are dispatched to subscribers upon subscription.

const pressureReading$ = new Observable(subscriber => {
  const intervalId = setInterval(() => {
    const pressure = // The pressure gauge data is assigned to this variable.
    subscriber.next(pressure); // Send data to subscribers
  }, 1000);

  // Clear timer when unsubscribe
  return () => clearInterval(intervalId);
});

// Subscribe to Observable
const subscription = pressureReading$.subscribe({
  next: pressure => console.log(`Pressure: ${pressure} bar`)
});

// Unsubscribe from Observable
subscription.unsubscribe();

Through the utilization of the Subject construct, we are afforded the opportunity to convert the data originating from the pressure gauge device into a Subject object, which in turn dispatches the data to its subscribers upon their subscription. However, it is important to note that due to its “hot” configuration, a Subject construct inherently generates data even prior to subscribers subscribing to it.

const pressureReading$ = new Subject();

setInterval(() => {
  const pressure = // The pressure gauge data is assigned to this variable.
  pressureReading$.next(pressure); // Send data to subscribers
}, 1000);

// Subscribe to Subject
const subscription = pressureReading$.subscribe({
  next: pressure => console.log(`Pressure: ${pressure} bar`)
});

// Unsubscribe from Subject
subscription.unsubscribe();

Although the utilization of Observable and Subject constructs in these examples is similar, the Subject’s capacity for generating data even before subscribers subscribe to it, due to its “hot” configuration, sets it apart. Consequently, in cases where Observables are “cold” and it is necessary to generate data before subscribers subscribe, the use of Subject is more appropriate.

The notion of “data production by an object before subscribers subscribe” refers to the generation of values by Observable or Subject objects without waiting for subscribers to subscribe.

For instance, a Subject object produces a series of data, and then subscribers subscribe to it. In this case, subscribers receive all the data that the Subject has produced. If the Subject does not generate data before subscribers subscribe, subscribers can only receive the data produced after subscribing.

This difference arises from the distinction between “hot” and “cold” Observable structures. A “hot” Observable or Subject sends its generated data to all subscribers, while a “cold” Observable sends its generated data only after subscribers have subscribed.

Subject and BehaviorSubject Differences

Subject and BehaviorSubject are two important types commonly used in the RxJS library.

Subject acts as a bridge between Observable and Observer. Therefore, Subject behaves like an Observable and contains methods such as subscribe, next, error, and complete.

BehaviorSubject, on the other hand, is a variation of Subject. That is, BehaviorSubject is also an Observable and a Subject at the same time. However, unlike Subject, BehaviorSubject provides an initial value related to the previous state when the subscribers subscribe for the first time.

In summary, BehaviorSubject has an initial value before the first subscription and this value is automatically sent with each new subscription. In contrast, Subject does not have such a feature.

Let’s imagine an application where the user can change the theme color. In such a scenario, BehaviorSubject can be utilized to track the theme color chosen by the user.

// To track the theme color using BehaviorSubject
const theme$ = new BehaviorSubject('light');

// Change theme color
function changeTheme(theme) {
  theme$.next(theme);
}

// Subscribe to BehaviorSubject
const subscription1 = theme$.subscribe({
  next: theme => console.log(`Theme changed: ${theme}`)
});

// Another subscription to BehaviorSubject
const subscription2 = theme$.subscribe({
  next: theme => console.log(`Theme changed: ${theme}`)
});

// Chane the theme color
changeTheme('dark');

In this example, BehaviorSubject sends the initial value “light” when it is first subscribed. Later, when the user changes the theme color, BehaviorSubject sends the new value “dark” to all subscribers.

BehaviorSubject is a hot Observable. This means that when BehaviorSubject is created, it immediately produces the initial value and returns it before any subscriber exists.

Therefore, BehaviorSubject does not require waiting to produce source data and can send these data to later subscribers as well, thanks to its ability to save the data it produces before the first subscriber.

ReplaySubject

ReplaySubject is a type present in the RxJS library and is a hot Observable. ReplaySubject stores and later resends the data that was published before the subscribe operation.

ReplaySubject is used to share your source data in a different way and is useful in scenarios where you want previous data to be resent.

ReplaySubject works just like BehaviorSubject, but has a buffering capacity to allow subscribers to receive data from before any given point in time.

ReplaySubject can be created with a specified buffering capacity. For instance, if the buffering capacity is 2, the last two values are resent.

The following example creates a sample dataset accessible by two different subscribers using ReplaySubject:

// Create dataset using ReplaySubject
const data$ = new ReplaySubject(2); // BUffering capacity 2

data$.next('data 1');
data$.next('data 2');
data$.next('data 3');

// createing 2 subscriber
const subscription1 = data$.subscribe({
  next: data => console.log(`Subscribe 1: ${data}`)
});

const subscription2 = data$.subscribe({
  next: data => console.log(`Subscribe 2: ${data}`)
});

In this example, the buffering capacity is set to 2. Later, three data points are published. The last two data points are buffered and sent to both subscribers. The subscribers receive the previous data as well as the data after their subscription.

If there was only one subscription, ReplaySubject would emit according to the buffering capacity. For instance, when the buffering capacity is set to n, the last n emitted values are stored, and a new subscriber receives the last n emitted values before any new values are emitted.

For example, in the following code, a ReplaySubject with a buffering capacity of 3 is created and only one subscriber is created:

const subject = new ReplaySubject(3);

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe(value => console.log(`Received value: ${value}`));

In this code, values 1, 2, 3, and 4 are sent sequentially to the ReplaySubject. Since the buffer capacity is set to 3, the subscriber will receive the previous last three values. Therefore, “Received value: 2”, “Received value: 3”, and “Received value: 4” are printed to the console in order.

We can create a ReplaySubject with a buffer capacity of 3 and create 3 subscribers. Thus, thanks to the buffer capacity, the previous last 3 values become available for each subscriber.

const subject = new ReplaySubject(3);

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe(value => console.log(`Received by Subscriber 1: ${value}`));

subject.next(5);

subject.subscribe(value => console.log(`Received by Subscriber 2: ${value}`));

subject.next(6);

subject.subscribe(value => console.log(`Received by Subscriber 3: ${value}`));

subject.next(7);

In this code, first, 4 different values are sent to the ReplaySubject sequentially, and each value is received by all three subscribers. Then, values 5 and 6 are sent, and both subscribers receive these values. Finally, a value of 7 is sent, and this value is also delivered to all subscribers.

The values printed to the console are as follows:

Received by Subscriber 1: 2
Received by Subscriber 1: 3
Received by Subscriber 1: 4
Received by Subscriber 2: 3
Received by Subscriber 2: 4
Received by Subscriber 2: 5
Received by Subscriber 1: 6
Received by Subscriber 2: 6
Received by Subscriber 3: 4
Received by Subscriber 3: 5
Received by Subscriber 3: 6
Received by Subscriber 1: 7
Received by Subscriber 2: 7
Received by Subscriber 3: 7

Why doesn’t Subscriber 3 see the value “4”?

When Subscriber 3 subscribes, the value “4” has not yet been sent to the ReplaySubject. The last three values (2, 3, and 4) that are within the buffer capacity have already been buffered by the ReplaySubject. Since no subscriber has subscribed before the value “4” is sent to the ReplaySubject, this value has not been buffered by the ReplaySubject and therefore cannot be seen by Subscriber 3.

When Subscriber 3 subscribes to the ReplaySubject, it can receive the last three values within the buffer capacity (5, 6, and 7) and all subsequent values. However, since the value “4” was not within the buffer capacity, Subscriber 3 cannot see it.

CONCLUSION

Subject can be considered as a subset of Observables and represents a data stream. Multiple subscribers can listen to the same data stream using the subscribe method of a Subject. However, since a Subject only publishes the current value to its subscribed observers, they cannot access previously published values.

BehaviorSubject is derived from Subject and holds onto the most recently published value. When a new subscriber subscribes, this last value is immediately sent. Therefore, BehaviorSubject is useful in situations where the most recent value of a data stream is needed. Unlike Subject, it takes an initial value.

ReplaySubject works similarly to BehaviorSubject, but also stores multiple previous values within the buffer capacity. When a new subscriber subscribes, the last values that have been buffered by ReplaySubject and all values sent afterwards can be seen.

In conclusion, the RxJS library offers different types of data streams such as Subject, BehaviorSubject, and ReplaySubject, and the properties of these types are selected according to different scenarios.