Subjects and BehaviorSubjects

Subjects are both Observables and Observers. They multicast values to multiple subscribers and can be manually pushed to.

Subject — Basic Multicast

import { Subject } from 'rxjs';

const subject = new Subject<string>();

// Subscriber A
subject.subscribe(value => console.log('A:', value));

// Subscriber B
subject.subscribe(value => console.log('B:', value));

// Push a value
subject.next('Hello');
// Output: A: Hello, B: Hello

// Late subscriber does NOT receive previous values
subject.subscribe(value => console.log('C:', value));

subject.next('World');
// Output: A: World, B: World (C missed 'Hello')

BehaviorSubject — Replays Latest Value

import { BehaviorSubject } from 'rxjs';

const behaviorSubject = new BehaviorSubject<string>('initial');

// Late subscriber immediately gets the latest value
behaviorSubject.subscribe(value => console.log('A:', value));
// Output: A: initial

behaviorSubject.next('updated');
// Output: A: updated

// New subscriber gets the latest value immediately
behaviorSubject.subscribe(value => console.log('B:', value));
// Output: B: updated

ReplaySubject — Replays N Values

import { ReplaySubject } from 'rxjs';

const replaySubject = new ReplaySubject<string>(2); // Buffer size 2

replaySubject.next('1');
replaySubject.next('2');
replaySubject.next('3');

replaySubject.subscribe(value => console.log('A:', value));
// Output: A: 2, A: 3 (last 2 values)

AsyncSubject — Final Value on Complete

import { AsyncSubject } from 'rxjs';

const asyncSubject = new AsyncSubject<string>();

asyncSubject.subscribe(value => console.log('A:', value));

asyncSubject.next('1');
asyncSubject.next('2');
asyncSubject.next('3');
asyncSubject.complete();
// Output: A: 3 (only the final value, on complete)

Subjects are the foundation of state management in Angular services.

BehaviorSubject for State

@Injectable({ providedIn: 'root' })
export class AuthService {
  private currentUser = new BehaviorSubject<User | null>(null);
  currentUser$ = this.currentUser.asObservable();

  login(credentials: Credentials): Observable<User> {
    return this.http.post<User>('/api/login', credentials).pipe(
      tap(user => this.currentUser.next(user))
    );
  }

  logout(): void {
    this.currentUser.next(null);
  }

  get isLoggedIn(): boolean {
    return this.currentUser.getValue() !== null;
  }
}

Subject for Event Bus

@Injectable({ providedIn: 'root' })
export class NotificationService {
  private notifications = new Subject<Notification>();
  notifications$ = this.notifications.asObservable();

  show(message: string, type: 'success' | 'error' | 'info' = 'info'): void {
    this.notifications.next({ message, type, id: Date.now() });
  }
}

// Component subscribes to notifications
@Component({
  selector: 'app-notification-listener',
  template: `...`,
})
export class NotificationListenerComponent implements OnInit, OnDestroy {
  private destroy$ = new Subject<void>();

  ngOnInit(): void {
    this.notificationService.notifications$.pipe(
      takeUntil(this.destroy$)
    ).subscribe(notification => {
      this.showToast(notification);
    });
  }

  ngOnDestroy(): void {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

Subject Type Selection