RxJS Subjects in Angular: Subject, BehaviorSubject, ReplaySubject and AsyncSubject

A normal Observable is like a private phone call: each subscriber starts its own conversation. A Subject is like a radio station: you can broadcast values to many listeners at once, and you can send values from the outside with next().

What you will learn

• What a Subject is, and hot versus cold Observables
• Subject, BehaviorSubject, ReplaySubject and AsyncSubject
• A real Angular example: a notification service
• When to use signals instead

Cold and hot

• A cold Observable (of, interval, http.get) runs again for every subscriber.
• A hot Observable shares one source between subscribers. Subjects are hot.

A Subject is both an Observable and an Observer: you can subscribe to it, and you can call next, error and complete on it.

The four kinds

// src/app/rxjs6/subjects.ts
import { AsyncSubject, BehaviorSubject, ReplaySubject, Subject } from 'rxjs';

// 1. Subject: no memory. Late subscribers miss earlier values.
export function plainSubject(): { early: number[]; late: number[] } {
  const subject = new Subject<number>();
  const early: number[] = [];
  const late: number[] = [];

  subject.subscribe((v) => early.push(v));
  subject.next(1);
  subject.next(2);
  subject.subscribe((v) => late.push(v)); // joins after 1 and 2
  subject.next(3);
  return { early, late }; // early [1,2,3], late [3]
}

// 2. BehaviorSubject: always has a current value, and replays it to new subscribers.
export function behaviorSubject(): { late: string[]; current: string } {
  const subject = new BehaviorSubject<string>('initial');
  subject.next('first');
  const late: string[] = [];
  subject.subscribe((v) => late.push(v)); // gets "first" immediately
  subject.next('second');
  return { late, current: subject.value }; // late ['first','second'], current 'second'
}

// 3. ReplaySubject: remembers the last N values and replays them to new subscribers.
export function replaySubject(): number[] {
  const subject = new ReplaySubject<number>(2); // buffer size 2
  subject.next(1);
  subject.next(2);
  subject.next(3);
  const late: number[] = [];
  subject.subscribe((v) => late.push(v));
  return late; // [2, 3]
}

// 4. AsyncSubject: emits ONLY the last value, and only when it completes.
export function asyncSubject(): { beforeComplete: number[]; afterComplete: number[] } {
  const subject = new AsyncSubject<number>();
  const values: number[] = [];
  subject.subscribe((v) => values.push(v));
  subject.next(1);
  subject.next(2);
  const beforeComplete = [...values]; // []
  subject.next(3);
  subject.complete();
  return { beforeComplete, afterComplete: values }; // [] then [3]
}

• `Subject`: Initial value: No; Replays to late subscribers: Nothing; Typical use: Events: clicks, "refresh" triggers
• `BehaviorSubject`: Initial value: Required; Replays to late subscribers: The latest value; Typical use: Current state: logged-in user, theme
• `ReplaySubject(n)`: Initial value: No; Replays to late subscribers: The last n values; Typical use: History: chat messages, recent logs
• `AsyncSubject`: Initial value: No; Replays to late subscribers: The final value, after completion; Typical use: "Result of a one-time job"

A notification service

// src/app/rxjs6/notification.service.ts
import { Injectable } from '@angular/core';
import { BehaviorSubject, Observable, Subject } from 'rxjs';

export interface Toast {
  id: number;
  text: string;
  kind: 'info' | 'error';
}

@Injectable({ providedIn: 'root' })
export class NotificationService {
  private nextId = 1;

  // Private subjects: only this service can push values.
  private readonly toastSubject = new Subject<Toast>(); // events
  private readonly countSubject = new BehaviorSubject<number>(0); // state

  // Public read-only Observables: everyone else can only listen.
  readonly toasts$: Observable<Toast> = this.toastSubject.asObservable();
  readonly count$: Observable<number> = this.countSubject.asObservable();

  show(text: string, kind: Toast['kind'] = 'info'): void {
    this.toastSubject.next({ id: this.nextId++, text, kind });
    this.countSubject.next(this.countSubject.value + 1);
  }
}

// src/app/rxjs6/toast-host.ts
import { AsyncPipe } from '@angular/common';
import { Component, inject, signal } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { NotificationService, Toast } from './notification.service';

@Component({
  selector: 'app-toast-host',
  imports: [AsyncPipe],
  template: `
    <p>Notifications sent: {{ notifications.count$ | async }}</p>
    <ul>
      @for (t of toasts(); track t.id) {
        <li [class.error]="t.kind === 'error'">{{ t.text }}</li>
      }
    </ul>
  `,
  styles: `.error { color: crimson; }`,
})
export class ToastHost {
  protected readonly notifications = inject(NotificationService);
  protected readonly toasts = signal<Toast[]>([]);

  constructor() {
    this.notifications.toasts$
      .pipe(takeUntilDestroyed())
      .subscribe((toast) => this.toasts.update((list) => [...list, toast]));
  }
}

// src/app/rxjs6/sender.ts
import { Component, inject } from '@angular/core';
import { NotificationService } from './notification.service';

@Component({
  selector: 'app-sender',
  template: `
    <button type="button" (click)="notifications.show('Saved successfully')">Send info</button>
    <button type="button" (click)="notifications.show('Something failed', 'error')">Send error</button>
  `,
})
export class Sender {
  protected readonly notifications = inject(NotificationService);
}

// src/app/app.ts
import { Component } from '@angular/core';
import { Sender } from './rxjs6/sender';
import { ToastHost } from './rxjs6/toast-host';

@Component({
  imports: [Sender, ToastHost],
  selector: 'app-root',
  templateUrl: './app.html',
  styleUrl: './app.css',
})
export class App {}

<!-- src/app/app.html -->
<main>
  <h1>Subjects</h1>
  <app-sender />
  <app-toast-host />
</main>

This is the communication through a service pattern. Two components that are not parent and child talk through a shared service that owns a Subject. The two key habits:

1. Keep the Subject private and expose asObservable(), so outsiders cannot call next.
2. Use a Subject for events and a BehaviorSubject for state.

Subject or signal?

Since signals arrived, many cases that used BehaviorSubject are simpler as a signal:

// BehaviorSubject style
private readonly _user$ = new BehaviorSubject<User | null>(null);
readonly user$ = this._user$.asObservable();

// signal style (simpler, no subscribe or unsubscribe, easy in templates)
private readonly _user = signal<User | null>(null);
readonly user = this._user.asReadonly();

Use signals for current state that templates read. Keep Subjects for events over time (clicks, messages, triggers) and when you need RxJS operators such as debounceTime or switchMap. Convert between them with toSignal and toObservable.

Tests

// src/app/rxjs6/subjects.spec.ts
import { TestBed } from '@angular/core/testing';
import { firstValueFrom } from 'rxjs';
import { NotificationService } from './notification.service';
import { asyncSubject, behaviorSubject, plainSubject, replaySubject } from './subjects';

describe('subjects', () => {
  it('Subject has no memory', () => {
    expect(plainSubject()).toEqual({ early: [1, 2, 3], late: [3] });
  });

  it('BehaviorSubject replays the current value', () => {
    expect(behaviorSubject()).toEqual({ late: ['first', 'second'], current: 'second' });
  });

  it('ReplaySubject replays the last N values', () => {
    expect(replaySubject()).toEqual([2, 3]);
  });

  it('AsyncSubject emits only the last value, on complete', () => {
    expect(asyncSubject()).toEqual({ beforeComplete: [], afterComplete: [3] });
  });

  it('NotificationService broadcasts to every listener and counts', async () => {
    const service = TestBed.inject(NotificationService);
    const first = firstValueFrom(service.toasts$);
    const second = firstValueFrom(service.toasts$);

    service.show('Hello');

    expect((await first).text).toBe('Hello');
    expect((await second).text).toBe('Hello');
    expect(await firstValueFrom(service.count$)).toBe(1);
  });
});

Try it yourself

1. Make NotificationService keep the last 3 toasts with a ReplaySubject(3), and show them in a component that is created later.
2. Replace the count BehaviorSubject by a signal and compare the template code.
3. Add a clear() method that empties the list through a second Subject.

Quick check

1. What is the difference between Subject and BehaviorSubject?
2. Why expose asObservable()?
3. What does AsyncSubject emit, and when?

Next, continue with Test Angular Components and Services with Vitest.