Subjects & Async Patterns

Subjects bridge imperative and reactive code. They act as both Observable and Observer — you push values in imperatively and multiple subscribers receive them reactively.

Subject Types

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

// Subject — hot multicast; new subscribers miss past emissions
const events$ = new Subject<string>();
events$.subscribe(e => console.log('A:', e));
events$.next('click');  // A: click
events$.subscribe(e => console.log('B:', e)); // B subscribes late
events$.next('keydown'); // A: keydown, B: keydown (both receive)

// BehaviorSubject — remembers and immediately emits current value to new subscribers
// Best for: shared mutable state (current user, theme, loading flag)
const theme$ = new BehaviorSubject<'light' | 'dark'>('light');
theme$.subscribe(t => console.log('subscriber 1:', t)); // immediately: 'light'
theme$.next('dark');
theme$.subscribe(t => console.log('subscriber 2:', t)); // immediately: 'dark'
console.log(theme$.getValue()); // 'dark' — synchronous access

// ReplaySubject(n) — replays last n emissions to new subscribers
const log$ = new ReplaySubject<string>(3); // buffer last 3
log$.next('a');
log$.next('b');
log$.next('c');
log$.next('d');
log$.subscribe(x => console.log(x)); // d, c, b (last 3 in order: b, c, d)

// AsyncSubject — only emits last value on complete (like a resolved Promise)
const result$ = new AsyncSubject<number>();
result$.subscribe(x => console.log('result:', x));
result$.next(1);
result$.next(2);
result$.next(3);
result$.complete(); // result: 3 — only last value emitted

takeUntil & Unsubscribe Patterns

import { Subject, interval, fromEvent } from 'rxjs';
import { takeUntil, takeWhile, take } from 'rxjs/operators';

// 1. takeUntil — the idiomatic Angular/RxJS pattern for cleanup
//    emit until a 'destroyer' subject emits (on component destroy)
class MyComponent {
  private destroy$ = new Subject<void>();

  ngOnInit() {
    interval(1000).pipe(
      takeUntil(this.destroy$)
    ).subscribe(n => console.log('tick', n));

    fromEvent(window, 'resize').pipe(
      takeUntil(this.destroy$)
    ).subscribe(() => this.onResize());
  }

  ngOnDestroy() {
    this.destroy$.next(); // triggers completion of all takeUntil streams
    this.destroy$.complete();
  }

  onResize() { /* ... */ }
}

// 2. takeWhile — complete when predicate returns false
interval(500).pipe(
  takeWhile(n => n < 5)
).subscribe(console.log); // 0, 1, 2, 3, 4 then completes

// 3. Explicit unsubscribe — store and manually unsubscribe
import { Subscription } from 'rxjs';
class AnotherComponent {
  private subs = new Subscription();

  ngOnInit() {
    this.subs.add(
      interval(1000).subscribe(n => console.log(n))
    );
    this.subs.add(
      fromEvent(window, 'scroll').subscribe(() => this.onScroll())
    );
  }

  ngOnDestroy() {
    this.subs.unsubscribe(); // cleans up all at once
  }

  onScroll() { /* ... */ }
}

State Management with BehaviorSubject

import { BehaviorSubject } from 'rxjs';
import { map, distinctUntilChanged } from 'rxjs/operators';

// Lightweight store pattern (alternative to NgRx for small apps)
interface AppState {
  user: { name: string } | null;
  loading: boolean;
  count: number;
}

const initialState: AppState = { user: null, loading: false, count: 0 };

class Store {
  private state$ = new BehaviorSubject<AppState>(initialState);

  // Select a slice of state (only emits when that slice changes)
  select<K extends keyof AppState>(key: K) {
    return this.state$.pipe(
      map(state => state[key]),
      distinctUntilChanged()
    );
  }

  getState(): AppState {
    return this.state$.getValue();
  }

  // Immutable updates
  patch(partial: Partial<AppState>) {
    this.state$.next({ ...this.getState(), ...partial });
  }

  setUser(user: AppState['user']) { this.patch({ user }); }
  setLoading(loading: boolean) { this.patch({ loading }); }
  increment() { this.patch({ count: this.getState().count + 1 }); }
}

const store = new Store();
store.select('user').subscribe(u => console.log('user changed:', u));
store.select('loading').subscribe(l => console.log('loading:', l));

store.setUser({ name: 'Alice' }); // user changed: { name: 'Alice' }
store.increment();                 // count: 1 (no user/loading emission)

Async Patterns & Real-World Examples

import { fromEvent, Subject, interval } from 'rxjs';
import { debounceTime, distinctUntilChanged, switchMap, catchError, withLatestFrom, bufferTime } from 'rxjs/operators';
import { EMPTY } from 'rxjs';

// 1. Search-as-you-type with cancellation
const searchInput = document.getElementById('search') as HTMLInputElement;
const search$ = fromEvent(searchInput, 'input').pipe(
  map((e: Event) => (e.target as HTMLInputElement).value.trim()),
  debounceTime(350),        // wait for typing to pause
  distinctUntilChanged(),   // skip if same value
  filter(q => q.length > 1),
  switchMap(q =>            // cancel in-flight request on new keystroke
    from(fetch(`/api/search?q=${encodeURIComponent(q)}`).then(r => r.json())).pipe(
      catchError(() => EMPTY) // swallow errors, show nothing
    )
  )
);

// 2. Polling with start/stop control
const startPolling$ = new Subject<void>();
const stopPolling$ = new Subject<void>();

startPolling$.pipe(
  switchMap(() =>
    interval(5000).pipe(
      takeUntil(stopPolling$),
      switchMap(() => from(fetch('/api/status').then(r => r.json())))
    )
  )
).subscribe(status => console.log('status:', status));

startPolling$.next(); // begin polling
// stopPolling$.next(); // stop polling

// 3. Buffer user actions and batch-process
const userActions$ = new Subject<string>();
userActions$.pipe(
  bufferTime(2000),  // collect all actions in 2s windows
  filter(actions => actions.length > 0)
).subscribe(actions => {
  console.log('batch:', actions); // process in bulk
});

userActions$.next('view:page1');
userActions$.next('click:button1');
// After 2s: ['view:page1', 'click:button1']