Week 15: RxJS Deep Dive

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

const plain = new Subject<number>();
plain.subscribe((v) => console.log('A:', v));
plain.next(1); // A: 1
plain.subscribe((v) => console.log('B:', v)); // B never gets 1 -- it subscribed too late
plain.next(2); // A: 2, B: 2

const withInitial = new BehaviorSubject<number>(0);
console.log(withInitial.value); // 0 -- always has a current value, readable synchronously
withInitial.subscribe((v) => console.log('C:', v)); // C: 0 -- gets the CURRENT value immediately

const replay = new ReplaySubject<number>(2); // remembers the last 2 values
replay.next(1);
replay.next(2);
replay.next(3);
replay.subscribe((v) => console.log('D:', v)); // D: 2, D: 3 -- the last two, not the first

searchTerm$.pipe(
  switchMap((term) => this.http.get<Result[]>(`/api/search?q=${term}`))
);
// New search term arrives while a request is in flight?
// The old request is CANCELLED. Only the latest term's result ever reaches you.

fileIds$.pipe(
  mergeMap((id) => this.http.post(`/api/files/${id}/process`, {}))
);
// Every inner Observable runs to completion, all in parallel, in whatever order they finish.
// Right for independent work; wrong when order or "only the latest" matters.

saveQueue$.pipe(
  concatMap((edit) => this.http.patch('/api/document', edit))
);
// Each save waits for the PREVIOUS save to finish before starting.
// Right whenever order matters -- out-of-order writes could corrupt the document.

submitClicks$.pipe(
  exhaustMap(() => this.http.post('/api/orders', this.orderPayload()))
);
// A second click while the first request is still in flight is IGNORED entirely.
// Right for submit buttons -- prevents accidental double-submission by design,
// not just by disabling the button (belt and suspenders).

combineLatest([this.searchTerm$, this.sortOrder$]).pipe(
  switchMap(([term, sort]) => this.productsService.search(term, sort))
);
// Re-emits whenever EITHER source emits, using the latest value from both.
// Requires every source to have emitted at least once before combineLatest emits anything.

forkJoin({
  user: this.http.get<User>('/api/me'),
  settings: this.http.get<Settings>('/api/settings'),
}).subscribe(({ user, settings }) => {
  // Runs once, only after BOTH requests complete -- like Promise.all
});

this.submitClicks$.pipe(
  withLatestFrom(this.formValue$),
  map(([, formValue]) => formValue)
).subscribe((formValue) => this.save(formValue));
// Only emits when submitClicks$ emits -- formValue$ changing alone does NOT trigger anything.
// It just supplies "whatever the form's latest value happens to be" at click time.

searchTerm$.pipe(
  switchMap((term) =>
    this.http.get(`/api/search?q=${term}`).pipe(
      catchError(() => of([])) // catches the INNER Observable's error
    )
  )
).subscribe((results) => this.results.set(results));

// One failed search -> that emission becomes []. The OUTER stream keeps living --
// the user can keep typing and future searches still work. This is usually what you want.

searchTerm$.pipe(
  switchMap((term) => this.http.get(`/api/search?q=${term}`)),
  catchError(() => of([])) // catches errors from the WHOLE outer pipeline
).subscribe((results) => this.results.set(results));

// One failed search -> catchError fires, but the OUTER Observable completes right after.
// The subscription is now dead -- typing more into the search box does nothing at all.

export class SearchBox implements OnDestroy {
  private sub = this.results$.subscribe((r) => this.render(r));

  ngOnDestroy() {
    this.sub.unsubscribe(); // miss this, and the subscription outlives the component
  }
}

import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

export class SearchBox {
  constructor() {
    this.results$.pipe(takeUntilDestroyed()).subscribe((r) => this.render(r));
  }
}

results$ = this.searchTerm$.pipe(switchMap((term) => this.search(term)));

@for (result of results$ | async; track result.id) {
  <li>{{ result.name }}</li>
}

import { Component, inject } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { toSignal } from '@angular/core/rxjs-interop';
import { Subject } from 'rxjs';
import { debounceTime, distinctUntilChanged, switchMap, catchError, of } from 'rxjs';

@Component({
  selector: 'app-typeahead-search',
  templateUrl: './typeahead-search.html',
})
export class TypeaheadSearch {
  private http = inject(HttpClient);
  private searchTerm$ = new Subject<string>();

  results = toSignal(
    this.searchTerm$.pipe(
      debounceTime(300),           // wait for a pause in typing
      distinctUntilChanged(),      // skip re-searching an unchanged term
      switchMap((term) =>
        term.length < 2
          ? of([])                 // don't search on 0-1 characters
          : this.search(term).pipe(catchError(() => of([])))
      )
    ),
    { initialValue: [] }
  );

  onInput(value: string) {
    this.searchTerm$.next(value);
  }

  private search(term: string) {
    return this.http.get<Result[]>(`/api/search?q=${term}`);
  }
}