angular-rxjs-patterns

# Angular RxJS Patterns Master RxJS in Angular for handling async operations, data streams, and reactive programming patterns. ## Observable Creation ### Basic Observable Creation ```typescript import { Observable, of, from, interval, fromEvent } from 'rxjs'; // of - emit values in sequence const numbers$ = of(1, 2, 3, 4, 5); // from - convert array, promise, or iterable const fromArray$ = from([1, 2, 3]); const fromPromise$ = from(fetch('/api/data')); // interval - emit numbers at intervals const timer$ = interval(1000); // Every second // fromEvent - DOM events const clicks$ = fromEvent(document, 'click'); // Custom observable const custom$ = new Observable(subscriber => { subscriber.next(1); subscriber.next(2); subscriber.complete(); }); ``` ### HttpClient Observables ```typescript import { HttpClient } from '@angular/common/http'; import { Injectable, inject } from '@angular/core'; import { Observable } from 'rxjs'; @Injectable({ providedIn: 'root' }) export class DataService { private readonly http = inject(HttpClient); getData(): Observable<Data[]> { return this.http.get<Data[]>('/api/data'); } getItem(id: string): Observable<Data> { return this.http.get<Data>(`/api/data/${id}`); } createItem(data: Data): Observable<Data> { return this.http.post<Data>('/api/data', data); } updateItem(id: string, data: Data): Observable<Data> { return this.http.put<Data>(`/api/data/${id}`, data); } deleteItem(id: string): Observable<void> { return this.http.delete<void>(`/api/data/${id}`); } } ``` ## Common Operators ### Transformation Operators ```typescript import { map, pluck, switchMap, mergeMap, concatMap } from 'rxjs/operators'; import { of } from 'rxjs'; // map - transform values const numbers$ = of(1, 2, 3).pipe( map(n => n * 2) // 2, 4, 6 ); // pluck - extract property (deprecated, use map) const users$ = of( { name: 'John', age: 30 }, { name: 'Jane', age: 25 } ).pipe( map(user => user.name) // 'John', 'Jane' ); // switchMap - cancel previous, emit new searchControl.valueChanges.pipe( switchMap(term => this.searchService.search(term)) ).subscribe(results => { this.results = results; }); // mergeMap - run in parallel const ids$ = of(1, 2, 3); ids$.pipe( mergeMap(id => this.getUser(id)) // All requests in parallel ).subscribe(); // concatMap - run in sequence ids$.pipe( concatMap(id => this.getUser(id)) // One at a time ).subscribe(); ``` ### Filtering Operators ```typescript import { filter, take, takeUntil, takeWhile, distinctUntilChanged } from 'rxjs/operators'; // filter - only emit matching values of(1, 2, 3, 4, 5).pipe( filter(n => n % 2 === 0) // 2, 4 ); // take - first N values interval(1000).pipe( take(5) // First 5 emissions ); // takeUntil - until another observable emits const destroy$ = new Subject(); source$.pipe( takeUntil(destroy$) ).subscribe(); // distinctUntilChanged - skip duplicate consecutive values of(1, 1, 2, 2, 3, 3).pipe( distinctUntilChanged() // 1, 2, 3 ); ``` ### Combination Operators ```typescript import { combineLatest, merge, concat, forkJoin, zip } from 'rxjs'; import { startWith } from 'rxjs/operators'; // combineLatest - emit when any source emits combineLatest([ this.user$, this.settings$ ]).pipe( map(([user, settings]) => ({ user, settings })) ).subscribe(); // merge - emit from any source merge( this.clicks$, this.hovers$ ).subscribe(); // concat - emit in sequence concat( this.loadUser$, this.loadSettings$ ).subscribe(); // forkJoin - wait for all to complete forkJoin({ user: this.getUser(), posts: this.getPosts(), comments: this.getComments() }).subscribe(({ user, posts, comments }) => { // All complete }); // zip - pair values from sources zip( of(1, 2, 3), of('a', 'b', 'c') ).pipe( map(([num, letter]) => `${num}${letter}`) ); // '1a', '2b', '3c' ``` ### Utility Operators ```typescript import { tap, delay, debounceTime, throttleTime, distinctUntilChanged } from 'rxjs/operators'; // tap - side effects (logging, etc.) source$.pipe( tap(value => console.log('Value:', value)), map(value => value * 2) ); // delay - delay emissions of(1, 2, 3).pipe( delay(1000) // Delay 1 second ); // debounceTime - wait for pause in emissions searchControl.valueChanges.pipe( debounceTime(300) // Wait 300ms after user stops typing ); // throttleTime - emit first value, ignore for duration clicks$.pipe( throttleTime(1000) // Only once per second ); // distinctUntilChanged - skip duplicates input$.pipe( distinctUntilChanged() // Only when value changes ); ``` ## Error Handling ### catchError - Handle Errors ```typescript import { catchError } from 'rxjs/operators'; import { of, EMPTY, throwError } from 'rxjs'; // Return fallback value this.http.get('/api/data').pipe( catchError(error => { console.error('Error:', error); return of([]); // Return empty array }) ); // Return empty observable source$.pipe( catchError(() => EMPTY) // Complete without emitting ); // Re-throw error source$.pipe( catchError(error => { console.error('Error:', error); return throwError(() => new Error('Custom error')); }) ); // Handle different error types source$.pipe( catchError(error => { if (error.status === 404) { return of(null); } return throwError(() => error); }) ); ``` ### retry and retryWhen ```typescript import { retry, retryWhen, delay, take } from 'rxjs/operators'; // Simple retry this.http.get('/api/data').pipe( retry(3) // Retry up to 3 times ); // Retry with delay this.http.get('/api/data').pipe( retryWhen(errors => errors.pipe( delay(1000), // Wait 1 second take(3) // Max 3 retries ) ) ); // Exponential backoff this.http.get('/api/data').pipe( retryWhen(errors => errors.pipe( mergeMap((error, index) => { if (index >= 3) { return throwError(() => error); } const delayMs = Math.pow(2, index) * 1000; return of(error).pipe(delay(delayMs)); }) ) ) ); ``` ## Subscription Management ### takeUntilDestroyed (Preferred) Use `takeUntilDestroyed()` from `@angular/core/rxjs-interop` — no `ngOnDestroy` needed, and no manual `Subject<void>` to manage: ```typescript import { Component, inject } from '@angular/core'; import { takeUntilDestroyed } from '@angular/core/rxjs-interop'; @Component({ selector: 'app-my-component', standalone: true }) export class MyComponent { private readonly dataService = inject(DataService); constructor() { this.dataService.data$.pipe( takeUntilDestroyed() // automatically unsubscribes when component destroys ).subscribe(data => { this.data = data; }); this.dataService.other$.pipe( takeUntilDestroyed() ).subscribe(other => { this.other = other; }); } } ``` ### DestroyRef for Manual Subscriptions When subscribing imperatively outside the constructor (e.g. on user action), use `DestroyRef` directly: ```typescript import { Component, inject } from '@angular/core'; import { DestroyRef } from '@angular/core'; @Component({ selector: 'app-my-component', standalone: true }) export class MyComponent { readonly #destroyRef = inject(DestroyRef); startPolling() { const sub = interval(5000).subscribe(() => this.poll()); this.#destroyRef.onDestroy(() => sub.unsubscribe()); } } ``` ### Async Pipe (No Manual Unsubscribe) ```typescript import { Component, inject } from '@angular/core'; import { AsyncPipe } from '@angular/common'; import { Observable } from 'rxjs'; @Component({ selector: 'app-user-list', standalone: true, imports: [AsyncPipe], template: ` @if (users$ | async; as users) { @for (user of users; track user.id) { <div>{{ user.name }}</div> } } @if (loading$ | async) { <div>Loading...</div> } @if (error$ | async; as error) { <div>Error: {{ error }}</div> } ` }) export class UserListComponent { private readonly userService = inject(UserService); users$: Observable<User[]> = this.userService.getUsers(); loading$: Observable<boolean> = this.userService.loading$; error$: Observable<string | null> = this.userService.error$; } ``` ## Subjects ### Subject - Multicast ```typescript import { Subject } from 'rxjs'; const subject = new Subject<number>(); // Multiple subscribers subject.subscribe(val => console.log('A:', val)); subject.subscribe(val => console.log('B:', val)); subject.next(1); // A: 1, B: 1 subject.next(2); // A: 2, B: 2 ``` ### BehaviorSubject - Current Value ```typescript import { BehaviorSubject } from 'rxjs'; const subject = new BehaviorSubject<number>(0); // Initial value subject.subscribe(val => console.log('A:', val)); // A: 0 subject.next(1); // A: 1 subject.next(2); // A: 2 subject.subscribe(val => console.log('B:', val)); // B: 2 (latest value) // Common pattern for state management @Injectable({ providedIn: 'root' }) export class StateService { private stateSubject = new BehaviorSubject<State>(initialState); state$ = this.stateSubject.asObservable(); updateState(newState: State) { this.stateSubject.next(newState); } get currentState(): State { return this.stateSubject.value; }