Practical RxJS Patterns in Angular

Search with Debounce

@Component({
  template: `
    <input #searchInput placeholder="Search for a samurai...">
    @for (result of searchResults$ | async; track result.id) {
      <div>{{ result.name }}</div>
    }
  `
})
export class SearchComponent implements AfterViewInit {
  @ViewChild('searchInput') searchInput!: ElementRef;
  searchResults$!: Observable<Samurai[]>;

  private http = inject(HttpClient);

  ngAfterViewInit(): void {
    this.searchResults$ = fromEvent<InputEvent>(
      this.searchInput.nativeElement,
      'input'
    ).pipe(
      map(event => (event.target as HTMLInputElement).value),
      debounceTime(300),           // Wait 300ms
      distinctUntilChanged(),       // Ignore the same values
      filter(query => query.length >= 2),  // Min 2 characters
      switchMap(query =>            // Cancel previous query
        this.http.get<Samurai[]>(`/api/search?q=${query}`).pipe(
          catchError(() => of([]))  // Error handling
        )
      )
    );
  }
}

Polling with Retry

@Injectable({ providedIn: 'root' })
export class StatusService {
  private http = inject(HttpClient);

  getStatusWithPolling(): Observable<Status> {
    return interval(5000).pipe(  // Every 5 seconds
      startWith(0),              // Immediately on start
      switchMap(() =>
        this.http.get<Status>('/api/status').pipe(
          retry({
            count: 3,
            delay: 1000  // Retry 3 times every second
          }),
          catchError(error => {
            console.error('Status check failed:', error);
            return of({ status: 'unknown', error: true });
          })
        )
      ),
      shareReplay(1)  // Share result between subscribers
    );
  }
}

Loading with Cache

@Injectable({ providedIn: 'root' })
export class CachedDataService {
  private http = inject(HttpClient);
  private cache$ = new Map<string, Observable<any>>();

  getData<T>(url: string): Observable<T> {
    if (!this.cache$.has(url)) {
      this.cache$.set(url,
        this.http.get<T>(url).pipe(
          shareReplay({ bufferSize: 1, refCount: true }),
          tap({
            error: () => this.cache$.delete(url)
          })
        )
      );
    }
    return this.cache$.get(url)!;
  }

  invalidateCache(url: string): void {
    this.cache$.delete(url);
  }
}

Handling Multiple Dependent Requests

getSamuraiWithWeapons(id: number): Observable<SamuraiWithWeapons> {
  return this.http.get<Samurai>(`/api/samurai/${id}`).pipe(
    switchMap(samurai =>
      forkJoin({
        weapons: this.http.get<Weapon[]>(`/api/samurai/${id}/weapons`),
        skills: this.http.get<Skill[]>(`/api/samurai/${id}/skills`)
      }).pipe(
        map(({ weapons, skills }) => ({
          ...samurai,
          weapons,
          skills
        }))
      )
    )
  );
}

Signals vs RxJS - When to Use Which?

// Integration - the best of both worlds
@Component({...})
export class HybridComponent {
  private http = inject(HttpClient);

  // Signal for UI state
  isLoading = signal(false);
  selectedId = signal<number | null>(null);

  // RxJS for HTTP
  samurai$ = toObservable(this.selectedId).pipe(
    filter(id => id !== null),
    tap(() => this.isLoading.set(true)),
    switchMap(id => this.http.get<Samurai>(`/api/samurai/${id}`)),
    tap(() => this.isLoading.set(false))
  );

  // Observable → Signal
  samurai = toSignal(this.samurai$);
}