Subject vs BehaviorSubject vs ReplaySubject vs AsyncSubject – what and when

There are four main types of subjects: Subject, BehaviorSubject, ReplaySubject, and AsyncSubject. In this article, we will explore the differences between these four types and when to use each one.

January 8, 2023 rxjsangular

What is Subject?

Subject is a special type of Observable that allows emitting value between all subscribers. Based on the implementation you can see that Subject defines methods like:

1
export class Subject<T> extends Observable<T> implements SubscriptionLike {
2
  next(value: T) {
3
    /** ... */
4
    for (const observer of this.currentObservers) {
5
      observer.next(value);
6
    }
7
    /** ... */
8
  }
9


10
  error(err: any) {
11
    /** ... */
12
  }
13


14
  complete() {
15
    /** ... */
16
  }
17
}

When to use Subject?

Subject is useful if you want to emit events (ex: click event, page changed event, modal closed event) and you are not interested in previous events.

1
import { Component, OnDestroy } from "@angular/core";
2
import { Subject } from "rxjs";
3


4
@Component({
5
  selector: "app-root",
6
  templateUrl: "./app.component.html",
7
})
8
export class ModalComponent implements OnDestroy {
9
  private close$$ = new Subject<void>();
10
  modalClosed$ = this.close$$.asObservable();
11


12
  ngOnDestroy() {
13
    this.close$$.next();
14
  }
15
}

What is BehaviorSubject?

BehaviorSubject is a special type of Subject that allows emitting value between all subscribers (like Subject) and storing the last emitted value. BehaviorSubject require to define the initial value. BehaviorSubject is useful if you know the initial state otherwise you should try to use ReplaySubject.

When a new subscriber arrives BehaviorSubject emit stored value to him (look at _subscribe method).

1
export class BehaviorSubject<T> extends Subject<T> {
2
  constructor(private _value: T) {
3
    /** ... */
4
  }
5


6
  get value(): T {
7
    return this.getValue();
8
  }
9


10
  /** @internal */
11
  protected _subscribe(subscriber: Subscriber<T>): Subscription {
12
    const subscription = super._subscribe(subscriber);
13
    !subscription.closed && subscriber.next(this._value);
14
    return subscription;
15
  }
16


17
  getValue(): T {
18
    /** ... */
19
    return this._value;
20
  }
21


22
  next(value: T): void {
23
    this._value = value;
24
    super.next(value);
25
  }
26
}

A new subscriber will not receive a value if BehaviorSubject is completed.

1
import { BehaviorSubject } from "rxjs";
2


3
const behavior$$ = new BehaviorSubject<number>(1);
4
behavior$$.complete();
5


6
// there will be no logs
7
behavior$$.subscribe((event) => console.log("Message: ", event));

Getting stored value

You can get the current state without calling subscribe method. To do it use getValue().

1
import { BehaviorSubject } from "rxjs";
2


3
const state$$ = new BehaviorSubject<AppState>({ state: "PRISTINE" });
4


5
console.log(state$$.getValue()); // -> { state: 'PRISTINE' }

When to use BehaviorSubject?

BehaviorSubject is useful if you want to store the current state of the application and you know the initial state or you can emulate it by marking the state as pristine.

1
@Component({
2
  selector: "app-root",
3
  templateUrl: "./app.component.html",
4
})
5
export class AppComponent {
6
  state$$ = new BehaviorSubject<AppState>({ state: "PRISTINE" });
7


8
  constructor(private projectService: ProjectService) {}
9


10
  load(): void {
11
    this.state$$.next({ state: "LOADING" });
12


13
    this.projectService.getProjects().subscribe((projects) =>
14
      this.state$$.next({
15
        state: "LOADED",
16
        data: projects,
17
      }),
18
    );
19
  }
20
}

Passing null as an initial state is not a good idea if null does not represent a real state.

What is ReplaySubject?

ReplaySubject is a special type of Subject that allows emitting values between all subscribers (like Subject) and storing emitted values based on time and/or buffer size. ReplaySubject does not require to define an initial value, so is useful if the initial value is unknown (in contrast to BehaviorSubject).

When a new subscriber arrives ReplaySubject emits all values from the buffer (look at _subscribe method). If the buffer is empty no events will be emitted.

1
export class ReplaySubject<T> extends Subject<T> {
2
  constructor(
3
    private _bufferSize = Infinity,
4
    private _windowTime = Infinity,
5
    private _timestampProvider: TimestampProvider = dateTimestampProvider,
6
  ) {
7
    /**... */
8
  }
9


10
  next(value: T): void {
11
    /** ... */
12
    _buffer.push(value);
13
    /** ... */
14
    this._trimBuffer();
15
    super.next(value);
16
  }
17


18
  /** @internal */
19
  protected _subscribe(subscriber: Subscriber<T>): Subscription {
20
    /** ... */
21
    this._trimBuffer();
22
    /** ... */
23
    const copy = _buffer.slice();
24
    for (let i = 0; i < copy.length && !subscriber.closed; i += _infiniteTimeWindow ? 1 : 2) {
25
      subscriber.next(copy[i] as T);
26
    }
27
    /** ... */
28
  }
29


30
  private _trimBuffer() {
31
    /** ... */
32
  }
33
}

Thanks to _bufferSize and _windowTime params you can define how many values and how long you want to store them.

ReplaySubject does not have a method to get the current buffer size so checking if the buffer is empty/full is not possible.

Buffer is trimmed only if a new subscriber has been added or a new value has been emitted. That means ReplaySubject it will not emit trimmed values even after window time has arrived.

By default buffer size is set as Infinity. If you do not need historical values do not forget to set the buffer size as one.

1
const token$ = new ReplaySubject<string>(1);

If you subscribe after calling complete method you will also receive the latest value (unlike BehaviorSubject).

1
const replay$$ = new ReplaySubject<number>();
2


3
replay$$.next(1);
4
replay$$.next(2);
5
replay$$.next(3);
6
replay$$.complete();
7


8
replay$$.subscribe((e) => console.log(e)); // -> 1, 2, 3

When to use ReplaySubject?

ReplaySubject is useful if you want to store the state of part of an application but you do not know the initial state.

1
@Component({
2
  selector: "app-root",
3
  templateUrl: "./app.component.html",
4
})
5
export class AppComponent {
6
  project$$ = new ReplaySubject<Project>(1);
7


8
  constructor(
9
    private projectService: ProjectService,
10
    private router: Router,
11
    private activatedRoute: ActivatedRoute,
12
  ) {}
13


14
  ngOnInit(): void {
15
    this.activatedRoute.paramMap
16
      .pipe(
17
        map((params) => params.get("projectId")),
18
        switchMap((id) => this.projectService.getProject(id)),
19
      )
20
      .subscribe((project) => this.project$$.next(project));
21
  }
22


23
  openProjectDetails() {
24
    this.project$$
25
      .pipe(
26
        take(1),
27
        switchMap((project) => this.router.navigate(["/", "projects", project.id])),
28
      )
29
      .subscribe();
30
  }
31
}

What is AsyncSubject?

AsyncSubject is a special type of Subject that allows emitting value between all subscribers (like Subject) but only the latest value on complete.

1
export class AsyncSubject<T> extends Subject<T> {
2
  private _value: T | null = null;
3
  private _hasValue = false;
4


5
  next(value: T): void {
6
    this._value = value;
7
    this._hasValue = true;
8
  }
9


10
  complete(): void {
11
    /** ... */
12
    this._hasValue && super.next(this._value!);
13
    /** ... */
14
  }
15
}

Note that the AsyncSubject will only emit a value when the complete function is called, and it will only emit the last value that was passed to the next function.

If the complete function is not called, the AsyncSubject will not emit any values.

If you subscribe after the calling complete method you will also receive the latest value.

1
const asyncSubject$$ = new AsyncSubject<number>();
2


3
// you will receive only the last value after complete
4
asyncSubject$$.subscribe((e) => console.log(e)); // -> 3
5


6
asyncSubject$$.next(1);
7
asyncSubject$$.next(2);
8
asyncSubject$$.next(3);
9
asyncSubject$$.complete();

When to use AsyncSubject?

AsyncSubject is useful if you want to cache a value that never changed.

1
@Component({
2
  selector: "app-root",
3
  templateUrl: "./app.component.html",
4
})
5
export class AppComponent implements OnInit {
6
  token$$ = new AsyncSubject<string>();
7


8
  constructor(private projectService: ProjectService) {}
9


10
  ngOnInit() {
11
    this.projectService.getTemporaryToken().subscribe((token) => {
12
      this.token$$.next(token);
13
      this.token$$.complete();
14
    });
15
  }
16
}

Do you like the content?

Your support helps me continue my work. Please consider making a donation.

Buy Me A Coffee

Donations are accepted through PayPal or Stripe. You do not need a account to donate. All major credit cards are accepted.

Leave a comment