Key RxJS Operators Every Angular Developer Should Know

// search.service.ts
import { inject, Injectable } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { Subject, of } from 'rxjs';
import { debounceTime, distinctUntilChanged, filter, map, switchMap, tap, catchError } from 'rxjs';
 
export interface Product {
  id: number;
  title: string;
  thumbnail: string;
  price: number;
}
 
interface ProductAPIResponse {
  products: Product[];
  total: number;
}
 
@Injectable({
  providedIn: 'root',
})
export class SearchService {
  private http = inject(HttpClient);
  private readonly API_URL = 'https://dummyjson.com/products/search';
 
  // 1. Subject to feed user keystrokes into the pipeline
  private searchTermSubject = new Subject<string>();
 
  // 2. Composed reactive stream
  readonly products$ = this.searchTermSubject.asObservable().pipe(
    // ⏰ 1. Debounce: wait for 400ms pause in typing
    debounceTime(400),
 
    // 🔍 2. Distinct: skip if the term is identical to the previous emission
    distinctUntilChanged(),
 
    // ✂️ 3. Filter: only search if length is at least 3 characters
    filter((term) => term.trim().length >= 3),
 
    // 📝 4. Tap: side-effect for debugging/logging
    tap((term) => console.log(`Searching API for: ${term}`)),
 
    // 🔄 5. SwitchMap: cancel previous in-flight HTTP request and switch to new one
    switchMap((term) => this.fetchProducts(term)),
 
    // 🎯 6. Map: extract the array of products from the response
    map((response) => response.products)
  );
 
  searchByText(term: string): void {
    this.searchTermSubject.next(term);
  }
 
  private fetchProducts(term: string) {
    return this.http.get<ProductAPIResponse>(`${this.API_URL}?q=${term}`).pipe(
      // 🛡️ 7. CatchError: prevent stream termination on HTTP error
      catchError((error) => {
        console.error('Error fetching products:', error);
        return of({ products: [], total: 0 }); // Fallback empty response
      })
    );
  }
}

// search.component.ts (Modern Standalone & Signals)
import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { SearchService } from './search.service';
 
@Component({
  selector: 'app-search',
  standalone: true,
  template: `
    <div class="search-container">
      <h1>Product Live Search</h1>
      
      <input
        type="search"
        placeholder="Search products (e.g. phone, laptop)..."
        #searchInput
        (input)="onSearch(searchInput.value)"
      />
 
      <div class="results-grid">
        @for (product of products(); track product.id) {
          <div class="product-card">
            <img [src]="product.thumbnail" [alt]="product.title" />
            <h3>{{ product.title }}</h3>
            <p>\${{ product.price }}</p>
          </div>
        } @empty {
          <p class="empty-state">No products found. Start typing to search.</p>
        }
      </div>
    </div>
  `,
})
export class SearchComponent {
  private searchService = inject(SearchService);
 
  // Convert RxJS search pipeline into a synchronous Signal with initial empty array
  products = toSignal(this.searchService.products$, { initialValue: [] });
 
  onSearch(term: string): void {
    this.searchService.searchByText(term);
  }
}