RxJS - BehaviorSubject, ReplaySubject and AsyncSubject

 

BehaviorSubject

Unlike plain Subject, BehaviorSubject can store the latest value and return it to Observers no matter the subscription is early or late.

Ex:

 

  // Cretae a plain Subject

  const subject = new Subject();

// Subscribe it before it emits any values

subject.subscribe(

next => {

console.log('(early) => ', next);

}

);

// Emit some values

subject.next(1);

subject.next(2);

// Subscribe it after 1 second delay

setTimeout(() => {

subject.subscribe(

next => {

console.log('(late) => ', next);

}

);

// Emit value after second subscription

subject.next(3);

}, 1000);

Result:

(early) => 1 (early) => 2 (early) => 3 (late) => 3

According to the result above, the second subscription did not get value 2 since the plain Subject did not store it.

Ex: (Using BehaviorSubject instead)

 

// Cretae a BehaviorSubject with default value 0

const subject = new BehaviorSubject(0);

// Subscribe it before it emits any values

subject.subscribe(

next => {

console.log('(early) => ', next);

}

);

// Emit some values

subject.next(1);

subject.next(2);

// Subscribe it after 1 second delay

setTimeout(() => {

subject.subscribe(

next => {

console.log('(late) => ', next);

}

);

// Emit value after second subscription

subject.next(3);

}, 1000);

Result:

(early) => 0 (Default Value) (early) => 1 (early) => 2 (late) => 2 (The latest value) (early) => 3 (late) => 3

After changing it to BehaviorSubject, we can get the latest/default value which is stored on it.

ReplaySubject

Using BehaviorSubject, you can get the latest value when subscribing.

But if the historical values do matter for you, then you need to use ReplaySubject instead.

Ex:

 

// Cretae a ReplaySubject

const subject = new ReplaySubject();

// Subscribe it before it emits any values

subject.subscribe(

next => {

console.log('(early) => ', next);

}

);

// Emit some values

subject.next(1);

subject.next(2);

// Subscribe it after 1 second delay

setTimeout(() => {

subject.subscribe(

next => {

console.log('(late) => ', next);

}

);

// Emit value after second subscription

subject.next(3);

}, 1000);

Result:

(early) => 1 (early) => 2 (late) => 1 (Historical data) (late) => 2        (Historical data) (early) => 3 (late) => 3

AsyncSubject

On the other hand, if you only care the latet value before completion, you need to use AsynSubject.

Ex:

 

  // Cretae a AsyncSubject

const subject = new AsyncSubject();

// Subscribe it before it emits any values

subject.subscribe(

next => {

console.log('(early) => ', next);

}

);

// Emit some values

subject.next(1);

subject.next(2);

// Subscribe it after 1 second delay

setTimeout(() => {

subject.subscribe(

next => {

console.log('(late) => ', next);

}

);

// Emit value after second subscription

subject.next(3);

// Complete the stream

subject.complete();

}, 1000);

Result:

(early) =>  3

(late) =>  3

NOTE: Without calling 'subject.complete()' from the example above, the observers will not get any values.