Subjects — BehaviorSubject, ReplaySubject and EventBus Patterns

A Subject is both an Observable (you can subscribe to it) and an Observer (you can call next(), error(), complete() on it). It is the RxJS equivalent of an event emitter. BehaviorSubject extends Subject by holding a current value — new subscribers immediately receive the latest value, making it ideal as a state container. Before Angular Signals, BehaviorSubject was the primary mechanism for reactive state in Angular services. Today both coexist — Signals for simpler local state, BehaviorSubject when the Observable pipeline is already RxJS-heavy.

Subjects in Practice #

import { Subject, BehaviorSubject, ReplaySubject } from 'rxjs';

// ── Subject — no initial value, subscribers miss past emissions ────────────
const clicks$ = new Subject<MouseEvent>();
clicks$.subscribe(e => console.log('Subscriber A clicked', e));
clicks$.next(new MouseEvent('click'));  // A receives this
// New subscriber after emission:
clicks$.subscribe(e => console.log('Subscriber B clicked', e));
clicks$.next(new MouseEvent('click'));  // Both A and B receive this

// ── BehaviorSubject — holds current value, new subs get it immediately ────
const count$ = new BehaviorSubject<number>(0);  // initial value = 0

count$.subscribe(v => console.log('A:', v));     // A: 0 (immediately)
count$.next(1);                                   // A: 1
count$.next(2);                                   // A: 2
count$.subscribe(v => console.log('B:', v));     // B: 2 (latest value)
count$.next(3);                                   // A: 3, B: 3

console.log(count$.value);  // read current value synchronously

// ── BehaviorSubject as service state ──────────────────────────────────────
@Injectable({ providedIn: 'root' })
export class CartService {
  private _items$ = new BehaviorSubject<CartItem[]>([]);

  // Expose as read-only Observable — prevents external .next() calls
  readonly items$ = this._items$.asObservable();
  readonly count$ = this._items$.pipe(map(items => items.length));
  readonly total$ = this._items$.pipe(
    map(items => items.reduce((sum, item) => sum + item.price * item.qty, 0))
  );

  addItem(item: CartItem): void {
    const current = this._items$.value;  // synchronous read of current value
    this._items$.next([...current, item]);
  }

  removeItem(id: string): void {
    this._items$.next(this._items$.value.filter(i => i.id !== id));
  }
}

// ── ReplaySubject — replays last N emissions to new subscribers ────────────
const log$ = new ReplaySubject<string>(3);  // buffer last 3
log$.next('Event 1');
log$.next('Event 2');
log$.next('Event 3');
log$.next('Event 4');  // pushes Event 1 out of the buffer

log$.subscribe(msg => console.log(msg));
// → Event 2, Event 3, Event 4 (last 3)

// ── Subject as event bus — cross-component communication ─────────────────
@Injectable({ providedIn: 'root' })
export class EventBusService {
  private _postPublished$ = new Subject<PostDto>();
  readonly postPublished$ = this._postPublished$.asObservable();

  publishPost(post: PostDto): void {
    this._postPublished$.next(post);
  }
}
// PostEditorComponent calls: eventBus.publishPost(savedPost)
// PostListComponent subscribes: eventBus.postPublished$.pipe(takeUntilDestroyed())

Common Mistakes #

Mistake 1 — Exposing Subject directly from service (bypasses encapsulation) #

❌ Wrong — public state$ = new BehaviorSubject(initialState); any component can mutate state directly.

✅ Correct — private _state$ = new BehaviorSubject(...); readonly state$ = this._state$.asObservable().

Mistake 2 — Using Subject instead of BehaviorSubject for state (new subscribers miss current value) #

❌ Wrong — Subject; component subscribes after state has been set; never receives current value.

✅ Correct — BehaviorSubject; new subscribers always receive the latest state immediately.