ReplaySubject, BehaviorSubject & AsyncSubject in Angular

ReplaySubject, BehaviorSubject & AsyncSubject are special types of subjects in Angular. In this tutorial let us learn what are they, how they work & how to use them in Angular

Table of Contents

BehaviorSubject
ReplaySubject
AsyncSubject
Reference

BehaviorSubject

BehaviorSubject requires an initial value and stores the current value and emits it to the new subscribers.

import { Component, VERSION } from "@angular/core";
import { BehaviorSubject, Subject } from "rxjs";

@Component({
  selector: "my-app",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent {
  subject$ = new BehaviorSubject("0");

  ngOnInit() {
    this.subject$.subscribe(val => {
      console.log("Sub1 " + val);
    });

    this.subject$.next("1");
    this.subject$.next("2");

    this.subject$.subscribe(val => {
      console.log("sub2 " + val);
    });

    this.subject$.next("3");
    this.subject$.complete();
  }
}



***Result***

Sub1 0
Sub1 1
Sub1 2
sub2 2
Sub1 3
sub2 3

StackBlitz

We create a new BehaviorSubject providing it an initial value or seed value. The BehaviorSubject stores the initial value.

subject$ = new BehaviorSubject("0");

As soon as the first subscriber subscribes to it, the BehaviorSubject emits the stored value. i.e. 0

    this.subject$.subscribe(val => {
      console.log("Sub1 " + val);
    });

We emit two more values. The BehaviorSubject stores the last value emitted i.e. 2

    this.subject$.next("1");
    this.subject$.next("2");

Now, Subscriber2 subscribes to it. It immediately receives the last value stored i.e. 2

    this.subject$.subscribe(val => {
      console.log("sub2 " + val);
    });

ReplaySubject

ReplaySubject replays old values to new subscribers when they first subscribe.

The ReplaySubject will store every value it emits in a buffer. It will emit them to the new subscribers in the order it received them. You can configure the buffer using the arguments bufferSize and windowTime

bufferSize: No of items that ReplaySubject will keep in its buffer. It defaults to infinity.

windowTime: The amount of time to keep the value in the buffer. Defaults to infinity.

Example


import { Component, VERSION } from "@angular/core";
import { ReplaySubject, Subject } from "rxjs";

@Component({
  selector: "my-app",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent {
  subject$ = new ReplaySubject();

  ngOnInit() {
    this.subject$.next("1");
    this.subject$.next("2");

    this.subject$.subscribe(
      val => console.log("Sub1 " + val),
      err => console.error("Sub1 " + err),
      () => console.log("Sub1 Complete")
    );

    this.subject$.next("3");
    this.subject$.next("4");

    this.subject$.subscribe(val => {
      console.log("sub2 " + val);
    });

    this.subject$.next("5");
    this.subject$.complete();

    this.subject$.error("err");
    this.subject$.next("6");

    this.subject$.subscribe(
      val => {
        console.log("sub3 " + val);
      },
      err => console.error("sub3 " + err),
      () => console.log("Complete")
    );
  }
}

***Result***
Sub1 1
Sub1 2
Sub1 3
Sub1 4
sub2 1
sub2 2
sub2 3
sub2 4
Sub1 5
sub2 5
Sub1 Complete
sub3 1
sub3 2
sub3 3
sub3 4
sub3 5
sub3 err

Stackblitz

First, we create a ReplaySubject

subject$ = new ReplaySubject();

ReplaySubject emits two values. It will also store these in a buffer.

    this.subject$.next("1");
    this.subject$.next("2");

We subscribe to it. The observer will receive 1 & 2 from the buffer

    this.subject$.subscribe(
      val => console.log("Sub1 " + val),
      err => console.error("Sub1 " + err),
      () => console.log("Sub1 Complete")
    );

We subscribe again after emitting two more values. The new subscriber will also receive all the previous values.

    this.subject$.next("3");
    this.subject$.next("4");

    this.subject$.subscribe(val => {
      console.log("sub2 " + val);
    });

We emit one more value & complete. All the subscribers will receive complete. They will not receive any further values or notifcations.

    this.subject$.next("5");
    this.subject$.complete();

We now fire an error notification and a value. None of the previous subscribers will receive this as they are already closed.

    this.subject$.error("err");
    this.subject$.next("6");

Now, we subscribe again. The subscriber will receive all the values up to Complete. But will not receive the Complete notification, instead, it will receive the Error notification.

    this.subject$.subscribe(
      val => {
        console.log("sub3 " + val);
      },
      err => console.error("sub3 " + err),
      () => console.log("Complete")
    );

AsyncSubject

AsyncSubject only emits the latest value only when it completes. If it errors out then it will emit an error, but will not emit any values.

import { Component, VERSION } from "@angular/core";
import { AsyncSubject, Subject } from "rxjs";

@Component({
  selector: "my-app",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent {
  subject$ = new AsyncSubject();

  ngOnInit() {
    this.subject$.next("1");
    this.subject$.next("2");

    this.subject$.subscribe(
      val => console.log("Sub1 " + val),
      err => console.error("Sub1 " + err),
      () => console.log("Sub1 Complete")
    );

    this.subject$.next("3");
    this.subject$.next("4");

    this.subject$.subscribe(val => {
      console.log("sub2 " + val);
    });

    this.subject$.next("5");
    this.subject$.complete();

    this.subject$.error("err");

    this.subject$.next("6");

    this.subject$.subscribe(
      val => console.log("Sub3 " + val),
      err => console.error("sub3 " + err),
      () => console.log("Sub3 Complete")
    );
  }
}


**Result **
Sub1 5
sub2 5
Sub1 Complete
Sub3 5
Sub3 Complete

Stackblitz

In the above example, all the subscribers will receive the value 5 including those who subscribe after the complete event.

But if the AsyncSubject errors out, then all subscribers will receive an error notification and no value.