rxjs

RxJS Patterns and Best Practices

Observable Creation

• Prefer Signals: Use Angular Signals for component state when possible

• Observables for Streams: Use Observables for continuous data streams (HTTP, events, timers)

• Creation Functions: Use appropriate creation functions

  // ✅ Good
  data$ = of(data);
  data$ = from(promise);
  data$ = fromEvent(element, 'click');
  
  // ❌ Avoid
  data$ = new Observable(subscriber => { ... });
  

Operators

Common Operators

• map: Transform values

  users$ = this.http
    .get<User[]>(url)
    .pipe(map((users) => users.map((u) => ({...u, fullName: `${u.firstName} ${u.lastName}`}))));
  

• filter: Filter values

  activeUsers$ = users$.pipe(filter((user) => user.active));
  

• switchMap: Cancel previous inner observables (HTTP requests)

  user$ = userId$.pipe(switchMap((id) => this.http.get<User>(`/users/${id}`)));
  

• mergeMap/flatMap: Concurrent inner observables

  results$ = ids$.pipe(mergeMap((id) => this.http.get<Result>(`/results/${id}`)));
  

• concatMap: Sequential inner observables

  results$ = ids$.pipe(concatMap((id) => this.http.get<Result>(`/results/${id}`)));
  

• exhaustMap: Ignore new emissions while inner observable is active

  save$ = saveAction$.pipe(exhaustMap(() => this.http.post('/save', data)));
  

• catchError: Handle errors

  data$ = this.http.get(url).pipe(
    catchError((error) => {
      console.error(error);
      return of(defaultValue);
    }),
  );
  

• tap: Side effects (logging, debugging)

  data$ = this.http.get(url).pipe(
    tap((data) => console.log('Received:', data)),
    tap({error: (err) => console.error('Error:', err)}),
  );
  

ngx-lift Operators

• createAsyncState: Transform Observable to AsyncState

  usersState$ = this.userService.getUsers().pipe(
    createAsyncState({
      next: (users) => console.log('Loaded users:', users),
      error: (err) => console.error('Failed to load users:', err),
    }),
  );