RxJS Subject Types: Subject, BehaviorSubject, ReplaySubject, AsyncSubject and When to Use None of Them

Version note: The examples in this post were run with RxJS 7.8.2, Angular 22.2, TypeScript 6.0 and Node.js 24 (the Angular project uses TypeScript 6.0, the range Angular 22.2 supports); timed examples run in virtual time on RxJS's TestScheduler, the Angular example in headless Chromium driven by Playwright 1.65.

Predict First

In Bookstore, the online bookstore used as the shared example across this site's series, Dune's stock goes down one by one: 12, 11, 10, 9. We push these numbers into four kinds of Subject and let a subscriber join late, after 10:

import { AsyncSubject, BehaviorSubject, ReplaySubject, Subject } from "rxjs";

// Dune's stock goes 12 -> 11 -> 10 -> 9; a late subscriber joins after 10
const subjects = {
  Subject: new Subject<number>(),
  BehaviorSubject: new BehaviorSubject<number>(12),
  "ReplaySubject(2)": new ReplaySubject<number>(2),
  AsyncSubject: new AsyncSubject<number>(),
};

for (const [name, subject] of Object.entries(subjects)) {
  const seen: string[] = [];
  subject.next(11);
  subject.next(10);
  subject.subscribe({ next: (stock) => seen.push(String(stock)), complete: () => seen.push("|") });
  subject.next(9);
  subject.complete();
  console.log(`${name.padEnd(17)} late subscriber saw: ${seen.join(" ")}`);
}

Write down what the late subscriber sees for each of the four lines before you read on. Which one sees 12?

So far every Observable we built had its own producer: a timer, an event, an HTTP request. A Subject is different: you call next on it yourself, and every subscriber gets the value. In the Angular series' "Angular'da RxJS'e Giriş" post we used a BehaviorSubject in a service; this post goes further. We compare the four Subject types, look at what they hand a late subscriber, see why a dead Subject fails silently and why a Subject in a public field causes trouble, and then decide in Angular 22 what belongs in a Subject and what belongs in a signal.

A Subject Is Both an Observable and an Observer

The answer:

Subject           late subscriber saw: 9 |
BehaviorSubject   late subscriber saw: 10 9 |
ReplaySubject(2)  late subscriber saw: 11 10 9 |
AsyncSubject      late subscriber saw: 9 |

A Subject has subscribe like an Observable and next, error, complete like an Observer. It keeps a list of subscribers and forwards each notification to all of them, so it is hot by nature: the values exist whether or not someone listens. The four types differ only in one thing, what a subscriber gets at the moment it subscribes:

In the output, Subject gave only 9, the value that came after the subscription. BehaviorSubject gave 10 immediately, because it always holds the current value; 12 was the initial value and had already been replaced. ReplaySubject(2) replayed the last two (11, 10) and then 9. AsyncSubject gave 9 only when complete was called; had the Subject never completed, the subscriber would have seen nothing at all. That is why AsyncSubject is rare in application code: it behaves like a Promise, and firstValueFrom, lastValueFrom or the multicasting operators of the next post cover the same need more clearly.

ReplaySubject: The Last N Values in a Time Window

ReplaySubject takes two limits: how many values to keep and for how long. The book detail page shows "recently viewed" books; we want at most three, and only from the last ten seconds:

import { ReplaySubject, timer } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";

runInVirtualTime("ReplaySubject(3, 10000): recently viewed books", () => {
  // keep at most 3 views, and only those from the last 10 seconds
  const viewed = new ReplaySubject<string>(3, 10_000);
  const views: [number, string][] = [[0, "Dune"], [2000, "Emma"], [4000, "The Hobbit"], [6000, "Middlemarch"], [13000, "Frankenstein"]];
  for (const [ms, title] of views) timer(ms).subscribe(() => viewed.next(title));

  for (const ms of [7000, 15000]) {
    timer(ms).subscribe(() => {
      const replayed: string[] = [];
      viewed.subscribe((title) => replayed.push(title)).unsubscribe();
      log(`late subscriber gets: ${replayed.join(", ")}`);
    });
  }
});

--- ReplaySubject(3, 10000): recently viewed books
 7000 ms  late subscriber gets: Emma, The Hobbit, Middlemarch
15000 ms  late subscriber gets: Middlemarch, Frankenstein

At 7 seconds four books had been viewed, the buffer kept the last three. At 15 seconds the window did the work: Emma (2 s) and The Hobbit (4 s) were older than ten seconds, so only Middlemarch and Frankenstein remained. The window is checked when values are added and when someone subscribes, so old values do not stay in memory forever. ReplaySubject() without arguments keeps everything; on a long-lived stream that is a memory leak, so always give at least a buffer size.

ReplaySubject(1) and BehaviorSubject look similar, but they are not the same. BehaviorSubject needs an initial value and gives it to the first subscriber even if nothing was ever pushed; ReplaySubject(1) gives nothing until the first next. If "no value yet" is a real state (the user has not logged in, the data has not arrived), ReplaySubject(1) describes it more honestly than a BehaviorSubject started with null.

After complete and error: A Dead Subject

A Subject follows the same contract as any Observable: after complete or error it is finished for good. What makes this dangerous is that it fails silently:

import { BehaviorSubject, Subject } from "rxjs";

const cartEvents = new Subject<string>();
cartEvents.subscribe({ next: (e) => console.log("A next:", e), complete: () => console.log("A complete") });
cartEvents.next("Dune added");
cartEvents.complete();
cartEvents.next("Emma added"); // ignored, no error
cartEvents.subscribe({ next: (e) => console.log("B next:", e), complete: () => console.log("B complete") });

const stock = new BehaviorSubject(12);
stock.complete();
stock.subscribe({ next: (n) => console.log("C next:", n), complete: () => console.log("C complete") });
console.log("getValue() after complete:", stock.getValue());

const prices = new Subject<number>();
prices.error(new Error("price feed lost"));
prices.subscribe({ next: (p) => console.log("D next:", p), error: (err: Error) => console.log("D error:", err.message) });

A next: Dune added
A complete
B complete
C complete
getValue() after complete: 12
D error: price feed lost

"Emma added" never reached anyone, and there was no error either; next on a stopped Subject simply does nothing. Subscriber B, who came after complete, got only the completion. The BehaviorSubject did not even give its current value to C, although getValue() still returns 12. And after error, every new subscriber gets the same error immediately. In practice this happens when someone calls complete() on a service-level Subject in a component's ngOnDestroy: the service lives on, but its Subject is dead for every component that comes later. Only the owner of a Subject should complete it, and a Subject that lives as long as the application rarely needs to be completed at all.

asObservable: Giving Out a Read-Only View

A Subject in a public field lets everyone who can see it emit values. Here is the difference between exposing the Subject and exposing a read-only view of it:

import { Subject, type Observable } from "rxjs";

class LeakyCartEvents {
  readonly added$ = new Subject<string>(); // anyone can call next
}

class CartEvents {
  readonly #added = new Subject<string>();
  readonly added$: Observable<string> = this.#added.asObservable(); // read-only view

  notifyAdded(title: string): void {
    this.#added.next(title);
  }
}

const leaky = new LeakyCartEvents();
leaky.added$.subscribe((title) => console.log(`toast: ${title} added to cart`));
leaky.added$.next("Free Bookmark"); // a component far away "helps" and fakes an event

const events = new CartEvents();
events.added$.subscribe((title) => console.log(`toast: ${title} added to cart`));
events.notifyAdded("Dune");
console.log("added$ has next:", "next" in events.added$);

toast: Free Bookmark added to cart
toast: Dune added to cart
added$ has next: false

In the first class a component far away called next on added$ and the toast announced a free bookmark that nobody added. Nothing in the code shows who did it; you search the project for .next(. In the second class the Subject is a private field (#added) and the outside world only gets asObservable(), an Observable without next. The only way to emit is the notifyAdded method, which you can find, test and log. The same rule applies to signals: Angular's asReadonly() gives out the value without set, as we will see in a moment.

Back to the Warehouse: Event or State?

In the creation functions post, a ★★☆ exercise asked how subscriber B could also see the warehouse events it missed, and what would go wrong. Here is the answer with Subjects:

import { ReplaySubject, Subject, timer } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";

// the warehouse example from the creation functions post, now with Subjects
runInVirtualTime("Subject: B misses earlier events", () => {
  const restocked = new Subject<string>();
  timer(100).subscribe(() => restocked.next("Dune"));
  timer(300).subscribe(() => restocked.next("Emma"));
  restocked.subscribe((title) => log(`A: ${title} restocked`));
  timer(250).subscribe(() => restocked.subscribe((title) => log(`B: ${title} restocked`)));
});

runInVirtualTime("ReplaySubject(1): B gets the last event, even when it is stale", () => {
  const restocked = new ReplaySubject<string>(1);
  const soldOut = new Subject<string>();
  timer(100).subscribe(() => restocked.next("Dune"));
  timer(200).subscribe(() => soldOut.next("Dune")); // Dune sells out again, restocked$ does not know
  soldOut.subscribe((title) => log(`warehouse: ${title} sold out`));
  timer(250).subscribe(() => restocked.subscribe((title) => log(`B: ${title} restocked (replayed)`)));
});

--- Subject: B misses earlier events
  100 ms  A: Dune restocked
  300 ms  A: Emma restocked
  300 ms  B: Emma restocked
--- ReplaySubject(1): B gets the last event, even when it is stale
  200 ms  warehouse: Dune sold out
  250 ms  B: Dune restocked (replayed)

With a plain Subject, B joined at 250 ms and missed "Dune restocked". ReplaySubject(1) fixes that, but the second scenario shows the problem the exercise hinted at: Dune sold out again at 200 ms, and B, joining at 250 ms, was still told that Dune had just been restocked. The replayed message was true when it happened and is false now.

The lesson is the most useful one in this post. An event says "something happened"; a state says "this is how things are". Replaying an event to someone who joins later turns history into a false present. If B needs to know whether Dune is in stock, B needs state: a stock value per book that both "restocked" and "sold out" update. That is exactly where a BehaviorSubject was used in older Angular code, and where Angular 22 uses a signal.

Angular 22: State in a Signal, Events in a Subject

Bookstore's Angular client, bookstore-web, now has a cart. The snapshot contains the project in the bookstore-web/ folder (npm install and npm start there, with node api/server.ts running in the repository root). The cart has two kinds of information: what is in the cart now, which the header badge and the cart page show at any time, and the fact that a book was just added, which the toast shows once. The first is state, the second an event:

import { computed, Service, signal } from '@angular/core';
import { Subject, type Observable } from 'rxjs';
import type { Book } from '../catalog/book';

export interface CartItem {
  book: Book;
  quantity: number;
}

@Service()
export class CartStore {
  // state: what the cart contains right now -> signal
  readonly #items = signal<CartItem[]>([]);
  readonly items = this.#items.asReadonly();
  readonly count = computed(() => this.#items().reduce((sum, i) => sum + i.quantity, 0));
  readonly total = computed(() => this.#items().reduce((sum, i) => sum + i.quantity * i.book.price, 0));

  // event: something just happened -> Subject, exposed read-only
  readonly #added = new Subject<Book>();
  readonly added$: Observable<Book> = this.#added.asObservable();

  add(book: Book): void {
    this.#items.update((items) => {
      const existing = items.find((i) => i.book.isbn === book.isbn);
      return existing
        ? items.map((i) => (i === existing ? { ...i, quantity: i.quantity + 1 } : i))
        : [...items, { book, quantity: 1 }];
    });
    this.#added.next(book);
  }
}

@Service() is what ng generate service produces in Angular 22; it is the short form of @Injectable({ providedIn: 'root' }). The items live in a private writable signal and go out through asReadonly(); count and total are computed, so the header simply reads cart.count() and is updated when the signal changes. The "added" event is a private Subject with an asObservable() view. The toast component turns it into a message that disappears after three seconds:

import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { concat, map, of, switchMap, timer } from 'rxjs';
import { CartStore } from './cart-store';

@Component({
  selector: 'app-cart-toast',
  template: `
    @if (message(); as text) {
      <p class="toast" role="status">{{ text }}</p>
    }
  `,
})
export class CartToast {
  readonly message = toSignal(
    inject(CartStore).added$.pipe(
      // show the message for 3 seconds; a new event replaces the current one
      switchMap((book) =>
        concat(of(`${book.title} added to cart`), timer(3000).pipe(map(() => null))),
      ),
    ),
    { initialValue: null },
  );
}

This is a job a signal cannot do well and RxJS does in one line: every event starts a three-second timer, and a new event cancels the old timer through switchMap. toSignal turns the result into a signal for the template and unsubscribes when the component is destroyed; its details come in the Angular post of this series. The catalog page calls cart.add(book) on a button click. We clicked Dune, then Emma one second later, in headless Chromium:

 0.4 s catalog loaded         | Cart (0): 0 TRY | (no toast)
 0.5 s click Dune             | Cart (1): 450 TRY | Dune added to cart
 1.7 s click Emma 1 s later   | Cart (2): 770 TRY | Emma added to cart
 4.2 s 2.5 s later            | Cart (2): 770 TRY | Emma added to cart
 5.2 s another 1 s later      | Cart (2): 770 TRY | (no toast)
Frankenstein button disabled: true

The header followed the signal on every click. The toast switched to Emma at 1.7 s, and at 4.2 s it was still visible: Dune's timer would have ended at about 3.5 s, but switchMap had cancelled it and started a new one for Emma, which ended at about 4.7 s. Frankenstein's button is disabled because its stock is 0.

Why not keep the cart in a BehaviorSubject, as in the "Angular'da RxJS'e Giriş" post? It still works, but in Angular 22 change detection is zoneless and components are OnPush by default (see "Angular 22: Varsayılan OnPush, Stable Signal Forms ve Angular 12 ile Proje Farkı" in the Angular series). A BehaviorSubject needs an async pipe or toSignal in every template that reads it, its current value is reached through getValue() or a subscription, and derived values need map plus combineLatest. A signal is read synchronously, derived with computed, and tells Angular exactly which view to update. The other way round, a signal is a poor event channel: setting the same value twice is not a change, and nothing waits for "the next one". So the division is simple: state goes into a signal, events go into a Subject (or an Observable from another source), and RxJS handles time.

When to Use None of Them: The Global Event Bus

A popular use of Subject is an application-wide event bus: one Subject, event types as strings, and every part of the application emits and listens:

import { filter, map, Subject } from "rxjs";

// an app-wide "event bus": one Subject, string event types, payload of any shape
type BusEvent = { type: string; payload: unknown };
const bus = new Subject<BusEvent>();
const on = (type: string) =>
  bus.pipe(
    filter((e) => e.type === type),
    map((e) => e.payload),
  );

on("cart:added").subscribe((p) => console.log("header badge got", p));
on("cart:add").subscribe((p) => console.log("toast got", p)); // typo: compiles, never fires

bus.next({ type: "cart:added", payload: { isbn: "9780441172719", title: "Dune" } });
bus.next({ type: "cart:added", payload: "Emma" }); // a different payload shape, also compiles

header badge got { isbn: '9780441172719', title: 'Dune' }
header badge got Emma

The toast never fired, because its listener waits for cart:add instead of cart:added; the compiler cannot help, because both are just strings. The header badge received two different payload shapes and must defend itself against both. And the question you ask during every bug hunt, "who sends this event?", has only one answer: search the whole code base for the string. The bus decouples the code so well that nobody can follow it any more.

Before you create a Subject, ask whether you need one at all. If the value comes from somewhere else (a DOM event, an HTTP call, a router event), use that Observable directly with fromEvent or the API's own Observable. If it is state, use a signal. If a child component reports to its parent, use an output(). A Subject is the right tool for an event that your own code produces and that several independent parts consume, and even then it belongs inside a typed service like CartStore, not in a global bus.

In Production

Subjects that live in root services live as long as the application. Their subscribers usually do not: every component that subscribes manually and forgets to unsubscribe stays in memory with its whole view. Prefer toSignal, the async pipe or takeUntilDestroyed, which clean up on destroy. In the Memory panel of Chrome DevTools, a heap snapshot after navigating back and forth a few times shows such leaks as component instances and detached DOM elements whose count keeps growing.

Calling next inside a subscriber of the same Subject is another source of surprises: the inner value is delivered to the remaining subscribers before the outer one, so they can see the values in a different order. If one event must trigger another, model it as a pipe (map, switchMap) rather than a next inside a subscribe.

Finally, log at the owner. Because every emission goes through a method like notifyAdded, a single line there (or a tap on the private Subject) records every event with its payload, which is exactly what a global bus makes impossible.

Common Pitfalls

• Exposing a Subject as a public field. Anyone can call next; keep it private and expose asObservable().
• Calling complete() on a shared Subject in a component's ngOnDestroy. Every later subscriber gets only completion, and later next calls are ignored without an error.
• Replaying events with ReplaySubject to late subscribers. A replayed event can describe a situation that no longer exists; keep state as state.
• Using ReplaySubject() without a buffer size on a long-lived stream. It keeps every value forever.
• Starting a BehaviorSubject with null to mean "not loaded yet" and then filtering null everywhere. ReplaySubject(1) or a signal with an explicit loading state says it better.
• Building a global event bus with string event types. Typos compile, payloads have no type, and nobody can find the sender.

A Small Experiment

★☆☆ In 01-predict.ts, move the late subscribe after complete(). Predict each line before you run it; which Subject still gives the late subscriber a value?

★★☆ Fix the warehouse example: keep a BehaviorSubject<Record<string, number>> with the stock per title, update it from both "restocked" and "sold out", and let B derive "is Dune in stock?" with map and distinctUntilChanged. Then write the same thing in bookstore-web with a signal and computed, and compare the amount of code.

★★★ Add an "Undo" button to the cart toast: clicking it removes the book that was just added and hides the toast at once. Keep CartStore's Subject private, and decide whether "undo" is a new event, a state change, or both.

A Subject shares values that our own code produces. In the next post we share an existing Observable instead: why two async pipes on the same HttpClient call send two requests, and how share, shareReplay and connectable turn a cold Observable into a shared one without leaking memory or caching a failed response forever.