Table of Contents
URL: https://www.progressiverobot.com/rxjs-subjects/
Introduction
A _subject_ in RxJS is a special hybrid that can act as both an observable and an observer at the same time. This way, data can be pushed into a subject, and the subject's subscribers will, in turn, receive that pushed data.
Subjects are useful for multicasting or for when a source of data is not easily transformed into an observable. It's easy to overuse subjects and oftentimes, as illustrated in this excellent post, subjects can be avoided when an observable source can be created otherwise.
In this post, you will learn about subjects, behavior subjects, and replay subjects.
Prerequisites
If you would like to follow along with this article, you will need:
- Some familiarity with RxJS observables and observers.
This tutorial was verified with rxjs v7.3.0.
Using Subjects
Creating a subject begins with new instance of RxJS's Subject:
const mySubject = new Rx.Subject();
Multiple subscriptions can be created and internally the subject will keep a list of subscriptions:
const subscription1 = mySubject.subscribe(x => console.log(`${x} ${x}`));
const subscription2 = mySubject.subscribe(x => console.log(x.toUpperCase()));
Data can be pushed into the subject using its next method:
mySubject.next('Hello!');
Running this script will produce the following output:
[secondary_label Output]
Hello! Hello!
HELLO!
For subscription1, this code will take the input and display it twice. For subscription2, this code will take the input and apply toUpperCase().
When data is pushed into a subject, it'll go through its internal list of subscriptions and next the data into each one.
Pushing Data to Subscriptions
Here's an example that demonstrates how data gets pushed to the subscriptions:
const mySubject = new Rx.Subject();
mySubject.next(1);
const subscription1 = mySubject.subscribe(x => {
console.log('From subscription 1:', x);
});
mySubject.next(2);
const subscription2 = mySubject.subscribe(x => {
console.log('From subscription 2:', x);
});
mySubject.next(3);
subscription1.unsubscribe();
mySubject.next(4);
With this example, here's the result that'll be printed in the console:
[secondary_label Output]
From subscription 1: 2
From subscription 1: 3
From subscription 2: 3
From subscription 2: 4
Note how subscriptions that arrive late are missing out on some of the data that's been pushed into the subject. We'll see how to address that later with _behavior subjects_ or _replay subjects_.
Multicasting Data to All Subscriptions
The real power of subjects comes into play with _multicasting_, where a subject is passed as the observer to an observable, which will mean that, when the observable emits, the data is multicasted to all of the subject's subscriptions:
Here's an example where a trickleWords observable emits a word every 750ms.
const mySubject = new Rx.Subject();
const words = ['Hot Dog', 'Pizza', 'Hamburger'];
const trickleWords = Rx.Observable.zip(
Rx.Observable.from(words),
Rx.Observable.interval(750),
word => word
);
const subscription1 = mySubject.subscribe(x => {
console.log(x.toUpperCase());
});
const subscription2 = mySubject.subscribe(x => {
console.log(
x
.toLowerCase()
.split('')
.reverse()
.join('')
);
});
trickleWords.subscribe(mySubject);
This will produce the following output after all the values have been emitted:
[secondary_label Output]
HOT DOG
god toh
PIZZA
azzip
HAMBURGER
regrubmah
For subscription1, the array of words have been modified with toUpperCase(). For subscription2, the array of words have been modified with toLowerCase() and `reverse().
Using asObservable
The asObservable operator can be used to transform a subject into an observable. This can be useful when you'd like to expose the data from the subject, but at the same time prevent having data inadvertently pushed into the subject:
const mySubject = new Rx.Subject();
const myObservable = mySubject.asObservable();
mySubject.next('Hello');
myObservable.next('World!');
This will produce the following output:
[secondary_label Output]
TypeError: myObservable.next is not a function
myObservable does not possess next, error, or complete.
Handling Errors
When a subject completes or errors out, all the internal subscriptions also complete or error out:
const mySubject = new Rx.Subject();
const subscription1 = mySubject.subscribe(null, error =>
console.log('From subscription 1:', error.message)
);
const subscription2 = mySubject.subscribe(null, error =>
console.log('From subscription 2:', error.message)
);
mySubject.error(new Error('Error!'));
This will produce the following output:
[secondary_label Output]
From subscription 1: Error!
From subscription 2: Error!
Error messages have been generated.
Using Replay Subjects
As discussed earlier, late subject subscriptions will miss out on the data that was emitted previously. _Replay subjects_ can help with that by keeping a buffer of previous values that will be emitted to new subscriptions.
Here's a usage example for replay subjects where a buffer of 2 previous values are kept and emitted on new subscriptions:
const mySubject = new Rx.ReplaySubject(2);
mySubject.next(1);
mySubject.next(2);
mySubject.next(3);
mySubject.next(4);
mySubject.subscribe(x => {
console.log('From subscription 1:', x);
});
mySubject.next(5);
mySubject.subscribe(x => {
console.log('From subscription 2:', x);
});
This will produce the following output:
[secondary_label Output]
From subscription 1: 3
From subscription 1: 4
From subscription 1: 5
From subscription 2: 4
From subscription 2: 5
A buffer of 2 values has been stored.
Using Behavior Subjects
_Behavior subjects_ are similar to replay subjects but will re-emit only the last emitted value or a default value if no value has been previously emitted:
const mySubject = new Rx.BehaviorSubject('Hello!');
mySubject.subscribe(x => {
console.log('From subscription 1:', x);
});
mySubject.next(5);
mySubject.subscribe(x => {
console.log('From subscription 2:', x);
});
This will produce the following output:
[secondary_label Output]
From subscription 1: Hello!
From subscription 1: 5
From subscription 2: 5
The default Hello! value was emitted.
Conclusion
In this post, you learned about subjects, behavior subjects, and replay subjects.
Continue your learning with An Introduction to RxJS Buffering Operators, RxJS: The From Operator, and How To Use the takeUntil RxJS Operator to Manage Subscriptions Declaratively.
