RxJS - Working with Subjects

A subject is an observable that can multicast i.e. talk to many observers. Consider a button with an event pstener, the function attached to the event using add pstener is called every time the user cpcks on the button similar functionapty goes for subject too.

We are going to discuss the following topics in this chapter −

Create a subject

What is the Difference between Observable and Subject?

Behaviour Subject

Replay Subject

AsyncSubject

Create a subject

To work with subject, we need to import Subject as shown below −

import { Subject } from  rxjs ;

You can create a subject object as follows −

const subject_test = new Subject();

The object is an observer that has three methods −

next(v)

error(e)

complete()

Subscribe to a Subject

You can create multiple subscription on the subject as shown below −

subject_test.subscribe({
   next: (v) => console.log(`From Subject : ${v}`)
});
subject_test.subscribe({
   next: (v) => console.log(`From Subject: ${v}`)
});

The subscription is registered to the subject object just pke addpstener we discussed earper.

Passing Data to Subject

You can pass data to the subject created using the next() method.

subject_test.next("A");

The data will be passed to all the subscription added on the subject.

Example

Here, is a working example of the subject −

import { Subject } from  rxjs ;

const subject_test = new Subject();

subject_test.subscribe({
   next: (v) => console.log(`From Subject : ${v}`)
});
subject_test.subscribe({
   next: (v) => console.log(`From Subject: ${v}`)
});
subject_test.next("A");
subject_test.next("B");

The subject_test object is created by calpng a new Subject(). The subject_test object has reference to next(), error() and complete() methods. The output of the above example is shown below −

Output

We can use complete() method to stop the subject execution as shown below.

Example

import { Subject } from  rxjs ;

const subject_test = new Subject();

subject_test.subscribe({
   next: (v) => console.log(`From Subject : ${v}`)
});
subject_test.subscribe({
   next: (v) => console.log(`From Subject: ${v}`)
});
subject_test.next("A");
subject_test.complete();
subject_test.next("B");

Once we call complete the next method called later is not invoked.

Output

Let us now see how to call error () method.

Example

Below is a working example −

import { Subject } from  rxjs ;

const subject_test = new Subject();

subject_test.subscribe({
   error: (e) => console.log(`From Subject : ${e}`)
});
subject_test.subscribe({
   error: (e) => console.log(`From Subject : ${e}`)
});
subject_test.error(new Error("There is an error"));

Output

What is the Difference between Observable and Subject?

An observable will talk one to one, to the subscriber. Anytime you subscribe to the observable the execution will start from scratch. Take an Http call made using ajax, and 2 subscribers calpng the observable. You will see 2 HttpHttp requests in the browser network tab.

Example

Here is a working example of same −

import { ajax } from  rxjs/ajax ;
import { map } from  rxjs/operators ;

let final_val = ajax( https://jsonplaceholder.typicode.com/users ).pipe(map(e => e.response));
let subscriber1 = final_val.subscribe(a => console.log(a));
let subscriber2 = final_val.subscribe(a => console.log(a));

Output

Now, here the problem is, we want the same data to be shared, but not, at the cost of 2 Http calls. We want to make one Http call and share the data between subscribers.

This will be possible using Subjects. It is an observable that can multicast i.e. talk to many observers. It can share the value between subscribers.

Example

Here is a working example using Subjects −

import { Subject } from  rxjs ;
import { ajax } from  rxjs/ajax ;
import { map } from  rxjs/operators ;

const subject_test = new Subject();

subject_test.subscribe({
   next: (v) => console.log(v)
});
subject_test.subscribe({
   next: (v) => console.log(v)
});

let final_val = ajax( https://jsonplaceholder.typicode.com/users ).pipe(map(e => e.response));
let subscriber = final_val.subscribe(subject_test);

Output

Now you can see only one Http call and the same data is shared between the subscribers called.

Behaviour Subject

Behaviour subject will give you the latest value when called.

You can create behaviour subject as shown below −

import { BehaviorSubject } from  rxjs ;
const subject = new BehaviorSubject("Testing Behaviour Subject"); 
// initiapzed the behaviour subject with value:Testing Behaviour Subject

Example

Here is a working example to use Behaviour Subject −

import { BehaviorSubject } from  rxjs ;
const behavior_subject = new BehaviorSubject("Testing Behaviour Subject"); 
// 0 is the initial value

behavior_subject.subscribe({
   next: (v) => console.log(`observerA: ${v}`)
});

behavior_subject.next("Hello");
behavior_subject.subscribe({
   next: (v) => console.log(`observerB: ${v}`)
});
behavior_subject.next("Last call to Behaviour Subject");

Output

Replay Subject

A replaysubject is similar to behaviour subject, wherein, it can buffer the values and replay the same to the new subscribers.

Example

Here is a working example of replay subject −

import { ReplaySubject } from  rxjs ;
const replay_subject = new ReplaySubject(2); 
// buffer 2 values but new subscribers

replay_subject.subscribe({
   next: (v) => console.log(`Testing Replay Subject A: ${v}`)
});

replay_subject.next(1);
replay_subject.next(2);
replay_subject.next(3);
replay_subject.subscribe({
   next: (v) => console.log(`Testing Replay Subject B: ${v}`)
});

replay_subject.next(5);

The buffer value used is 2 on the replay subject. So the last two values will be buffered and used for the new subscribers called.

Output

AsyncSubject

In the case of AsyncSubject the last value called is passed to the subscriber and it will be done only after complete() method is called.

Example

Here is a working example of the same −

import { AsyncSubject } from  rxjs ;

const async_subject = new AsyncSubject();

async_subject.subscribe({
   next: (v) => console.log(`Testing Async Subject A: ${v}`)
});

async_subject.next(1);
async_subject.next(2);
async_subject.complete();
async_subject.subscribe({
   next: (v) => console.log(`Testing Async Subject B: ${v}`)
});

Here, before complete is called the last value passed to the subject is 2 and the same it given to the subscribers.

Output