Angular — Module 5: Reactive Programming Mastery — RxJS & Signals

import { Observable } from 'rxjs';

// An Observable is fundamentally a function that receives a Subscriber
// and defines how/when to push values into it.
const ticker$ = new Observable<number>((subscriber) => {
  let count = 0;
  const intervalId = setInterval(() => {
    subscriber.next(count++);
    if (count > 5) {
      subscriber.complete();
    }
  }, 1000);

  // The returned teardown function runs on unsubscribe OR complete/error —
  // this is what prevents memory leaks from abandoned intervals/subscriptions.
  return () => clearInterval(intervalId);
});

const subscription = ticker$.subscribe({
  next: (value) => console.log('tick:', value),
  error: (err: unknown) => console.error('ticker failed:', err),
  complete: () => console.log('ticker finished'),
});

// Cancels the interval early via the teardown function above.
setTimeout(() => subscription.unsubscribe(), 3500);

import { Injectable, inject } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { Subject, Observable, of } from 'rxjs';
import { debounceTime, distinctUntilChanged, filter, map, switchMap, catchError } from 'rxjs/operators';
import type { OperationTask } from '../models/operation-task.model';

@Injectable({ providedIn: 'root' })
export class TaskSearchService {
  private readonly http = inject(HttpClient);
  private readonly searchTerm$ = new Subject<string>();

  readonly results$: Observable<OperationTask[]> = this.searchTerm$.pipe(
    // filter: skip empty/whitespace-only queries entirely — no request fired
    filter((term) => term.trim().length > 0),

    // debounceTime: wait for a 300ms pause in typing before proceeding,
    // preventing a network request on every keystroke
    debounceTime(300),

    // distinctUntilChanged: skip re-querying if the debounced term is
    // identical to the last one that was actually searched
    distinctUntilChanged(),

    // switchMap: maps the search term to an HTTP Observable, and
    // CANCELS any in-flight previous request — critical for search-as-you-type,
    // where an earlier slow response must never overwrite a later one
    switchMap((term) =>
      this.http.get<OperationTask[]>('/api/operations/tasks/search', { params: { q: term } }).pipe(
        map((tasks) => tasks.filter((task) => task.status !== 'complete')),

        // catchError: contain the failure to THIS request only — returning
        // of([]) means the stream stays alive for the next search term
        // instead of terminating the whole results$ Observable on one bad request
        catchError((error: unknown) => {
          console.error('Task search failed:', error);
          return of([]);
        })
      )
    )
  );

  search(term: string): void {
    this.searchTerm$.next(term);
  }
}

<input type="search" (input)="onSearchInput($event)" placeholder="Search tasks..." />

<!-- async pipe subscribes on render and automatically unsubscribes on
     component destroy — no ngOnDestroy cleanup required for this stream. -->
<ul>
  <li *ngFor="let task of taskSearch.results$ | async">{{ task.title }}</li>
</ul>

import { Component, signal, computed, effect } from '@angular/core';
import type { OperationTask, TaskStatus } from '../models/operation-task.model';

@Component({
  selector: 'app-task-board',
  standalone: true,
  template: `
    <p>Showing {{ visibleTasks().length }} of {{ tasks().length }} tasks</p>
    <select (change)="onFilterChange($event)">
      <option value="all">All</option>
      <option value="in-progress">In Progress</option>
      <option value="blocked">Blocked</option>
    </select>
  `,
})
export class TaskBoardComponent {
  // signal(): a writable, observable container for a single value.
  readonly tasks = signal<OperationTask[]>([]);
  readonly statusFilter = signal<TaskStatus | 'all'>('all');

  // computed(): derives a new signal from others, re-evaluating ONLY
  // when tasks() or statusFilter() actually change — memoized, synchronous,
  // and requires no manual dependency array (unlike React's useMemo).
  readonly visibleTasks = computed(() => {
    const filterValue = this.statusFilter();
    const allTasks = this.tasks();
    return filterValue === 'all' ? allTasks : allTasks.filter((t) => t.status === filterValue);
  });

  constructor() {
    // effect(): runs a side effect whenever any signal read inside it
    // changes. Used for logging/analytics/localStorage sync — NOT for
    // deriving state (that's what computed() is for).
    effect(() => {
      console.log(`Filter changed to "${this.statusFilter()}", showing ${this.visibleTasks().length} tasks`);
    });
  }

  onFilterChange(event: Event): void {
    const select = event.target as HTMLSelectElement;
    this.statusFilter.set(select.value as TaskStatus | 'all');
  }
}

import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { OperationsService } from '../services/operations.service';

@Component({
  selector: 'app-task-board',
  standalone: true,
  template: `<p>{{ tasks()?.length ?? 0 }} tasks loaded</p>`,
})
export class TaskBoardComponent {
  private readonly operationsService = inject(OperationsService);

  // toSignal() subscribes once, converts each emission into the signal's
  // current value, and unsubscribes automatically on component destroy —
  // Observable ergonomics for the fetch, signal ergonomics for consumption.
  readonly tasks = toSignal(this.operationsService.getTasks());
}