rxjs-patterns-for-angular

# RxJS Patterns for Angular Skill

This skill helps implement reactive patterns using RxJS in Angular applications.

## Core Principles

### Modern Angular + RxJS
- **Signals First**: Use Signals for state, RxJS for async operations
- **Auto Cleanup**: Use `takeUntilDestroyed()` for subscription management
- **Interop**: Use `toSignal()` and `toObservable()` for Signal/Observable conversion
- **AsyncPipe**: Prefer AsyncPipe in templates when not using Signals

### Key Concepts
- Observables for async data streams
- Operators for data transformation
- Subscription management and cleanup
- Error handling and retry logic

## Signal + RxJS Integration

### toSignal() - Observable to Signal

```typescript
import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { HttpClient } from '@angular/common/http';

@Component({
  selector: 'app-task-list',
  template: `
    @if (tasks(); as taskList) {
      @for (task of taskList; track task.id) {
        <div>{{ task.title }}</div>
      }
    }
  `
})
export class TaskListComponent {
  private http = inject(HttpClient);
  
  // Convert Observable to Signal
  tasks = toSignal(
    this.http.get<Task[]>('/api/tasks'),
    { initialValue: [] }
  );
}
```

### toObservable() - Signal to Observable

```typescript
import { Component, signal } from '@angular/core';
import { toObservable } from '@angular/core/rxjs-interop';
import { switchMap } from 'rxjs/operators';

@Component({
  selector: 'app-search',
  template: `
    <input 
      nz-input 
      [ngModel]="searchQuery()" 
      (ngModelChange)="searchQuery.set($event)" 
    />
    
    @if (results(); as resultList) {
      @for (result of resultList; track result.id) {
        <div>{{ result.name }}</div>
      }
    }
  `
})
export class SearchComponent {
  searchQuery = signal('');
  
  // Convert Signal to Observable and transform
  private searchQuery$ = toObservable(this.searchQuery);
  
  results = toSignal(
    this.searchQuery$.pipe(
      debounceTime(300),
      distinctUntilChanged(),
      switchMap(query => this.searchService.search(query))
    ),
    { initialValue: [] }
  );
}
```

## Subscription Management

### takeUntilDestroyed() - Auto Cleanup

```typescript
import { Component, inject, signal, DestroyRef } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { interval } from 'rxjs';

@Component({
  selector: 'app-timer',
  template: `<div>Time: {{ time() }}</div>`
})
export class TimerComponent {
  private destroyRef = inject(DestroyRef);
  time = signal(0);
  
  constructor() {
    // Subscription automatically cleaned up on component destroy
    interval(1000)
      .pipe(takeUntilDestroyed(this.destroyRef))
      .subscribe(value => this.time.set(value));
  }
}
```

### Manual Cleanup (Legacy Pattern - Avoid)

```typescript
// ❌ DON'T: Manual subscription management (old pattern)
export class LegacyComponent implements OnDestroy {
  private subscription = new Subscription();
  
  ngOnInit() {
    this.subscription.add(
      this.dataService.getData().subscribe(data => {
        // handle data
      })
    );
  }
  
  ngOnDestroy() {
    this.subscription.unsubscribe();
  }
}

// ✅ DO: Use takeUntilDestroyed()
export class ModernComponent {
  private destroyRef = inject(DestroyRef);
  data = signal<any>(null);
  
  constructor() {
    this.dataService.getData()
      .pipe(takeUntilDestroyed(this.destroyRef))
      .subscribe(data => this.data.set(data));
  }
}
```

## Common Operators

### switchMap - Switch to New Observable

```typescript
// Switch to new search on every query change
searchResults$ = this.searchQuery$.pipe(
  debounceTime(300),
  switchMap(query => this.http.get(`/api/search?q=${query}`))
);
```

### mergeMap - Merge Multiple Observables

```typescript
// Process all tasks in parallel
processTasks$ = this.tasks$.pipe(
  mergeMap(tasks => 
    from(tasks).pipe(
      mergeMap(task => this.processTask(task))
    )
  )
);
```

### concatMap - Process Sequentially

```typescript
// Process tasks one by one in order
processTasks$ = this.tasks$.pipe(
  concatMap(tasks =>
    from(tasks).pipe(
      concatMap(task => this.processTask(task))
    )
  )
);
```

### debounceTime - Debounce Input

```typescript
// Wait 300ms after user stops typing
search$ = this.searchInput$.pipe(
  debounceTime(300),
  distinctUntilChanged(),
  switchMap(query => this.searchService.search(query))
);
```

### distinctUntilChanged - Skip Duplicates

```typescript
// Only emit when value actually changes
status$ = this.statusSubject$.pipe(
  distinctUntilChanged()
);
```

### filter - Filter Values

```typescript
// Only emit non-empty strings
nonEmptySearch$ = this.searchQuery$.pipe(
  filter(query => query.trim().length > 0),
  switchMap(query => this.search(query))
);
```

### map - Transform Values

```typescript
// Transform task to display format
taskDisplay$ = this.task$.pipe(
  map(task => ({
    title: task.title,
    status: task.status.toUpperCase(),
    dueDate: formatDate(task.dueDate)
  }))
);
```

### tap - Side Effects

```typescript
// Log without transforming
tasks$ = this.http.get<Task[]>('/api/tasks').pipe(
  tap(tasks => console.log('Loaded tasks:', tasks.length)),
  tap(tasks => this.analyticsService.track('tasks_loaded'))
);
```

## Combining Observables

### combineLatest - Wait for All

```typescript
import { combineLatest } from 'rxjs';

// Combine multiple observables
viewModel$ = combineLatest([
  this.tasks$,
  this.users$,
  this.settings$
]).pipe(
  map(([tasks, users, settings]) => ({
    tasks,
    users,
    settings
  }))
);

// Convert to Signal
viewModel = toSignal(this.viewModel$);
```

### forkJoin - Wait for All to Complete

```typescript
import { forkJoin } from 'rxjs';

// Load multiple resources in parallel
loadAll$ = forkJoin({
  tasks: this.taskService.getTasks(),
  users: this.userService.getUsers(),
  projects: this.projectService.getProjects()
}).pipe(
  map(({ tasks, users, projects }) => ({
    tasks,
    users,
    projects
  }))
);
```

### merge - Merge Multiple Streams

```typescript
import { merge } from 'rxjs';

// Combine multiple event streams
allEvents$ = merge(
  this.createEvent$,
  this.updateEvent$,
  this.deleteEvent$
).pipe(
  tap(event => this.handleEvent(event))
);
```

### zip - Pair Up Values

```typescript
import { zip } from 'rxjs';

// Pair up matching values from two streams
paired$ = zip(
  this.stream1$,
  this.stream2$
).pipe(
  map(([value1, value2]) => ({ value1, value2 }))
);
```

## Error Handling

### catchError - Handle Errors

```typescript
tasks$ = this.http.get<Task[]>('/api/tasks').pipe(
  catchError(error => {
    console.error('Failed to load tasks:', error);
    this.notificationService.error('Failed to load tasks');
    return of([]); // Return empty array as fallback
  })
);
```

### retry - Retry on Failure

```typescript
tasks$ = this.http.get<Task[]>('/api/tasks').pipe(
  retry(3), // Retry up to 3 times
  catchError(error => {
    console.error('Failed after 3 retries:', error);
    return of([]);
  })
);
```

### retryWhen - Conditional Retry with Backoff

```typescript
import { retryWhen, delay, scan, throwError } from 'rxjs';

tasks$ = this.http.get<Task[]>('/api/tasks').pipe(
  retryWhen(errors =>
    errors.pipe(
      scan((retryCount, error) => {
        if (retryCount >= 3) {
          throw error; // Max retries reached
        }
        console.log(`Retry ${retryCount + 1}/3`);
        return retryCount + 1;
      }, 0),
      delay(1000) // Wait 1 second between retries
    )
  ),
  catchError(error => {
    console.error('Failed after retries:', error);
    return of([]);
  })
);
```

## Real-Time Data

### interval - Periodic Updates

```typescript
import { interval, switchMap } from 'rxjs';

// Poll every 30 seconds
liveData$ = interval(30000).pipe(
  startWith(0), // Emit immediately
  switchMap(() => this.http.get('/api/live-data')),
  takeUntilDestroyed(this.destroyRef)
);

liveData = toSignal(this.liveData$);
```

### WebSocket Pattern

```typescript
import { webSocket } from 'rxjs/webSocket';

export class RealtimeService {
  private socket$ = webSocket('wss://api.example.com/ws');
  
  messages$ = this.socket$.pipe(
    catchError(error => {
      console.error('WebSocket error:', error);
      return EMPTY;
    }),
    retry({ delay: 5000 }) // Reconnect after 5 seconds
  );
  
  sendMessage(msg: any): void {
    this.socket$.next(msg);
  }
}
```

## Loading States

### Share Loading State

```typescript
import { shareReplay } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class TaskService {
  private http = inject(HttpClient);
  
  // Cache and share the result
  tasks$ = this.http.get<Task[]>('/api/tasks').pipe(
    shareReplay({ bufferSize: 1, refCount: true })
  );
}
```

### Loading Indicator Pattern

```typescript
@Component({
  selector: 'app-task-list',
  template: `
    @if (loading()) {
      <nz-spin />
    } @else if (error()) {
      <nz-alert nzType="error" [nzMessage]="error()!" />
    } @else {
      @for (task of tasks(); track task.id) {
        <div>{{ task.title }}</div>
      }
    }
  `
})
export class TaskListComponent {
  private taskService = inject(TaskService);
  private destroyRef = inject(DestroyRef);
  
  loading = signal(false);
  error = signal<string | null>(null);
  tasks = signal<Task[]>([]);
  
  constructor() {
    this.loadTasks();
  }
  
  loadTasks(): void {
    this.loading.set(true);
    this.error.set(null);
    
    this.taskService.tasks$
      .pipe(takeUntilDestroyed(this.destroyRef))
      .subscribe({
        next: (tasks) => {
          this.tasks.set(tasks);
          this.loading.set(false);
        },
        error: (err) => {
          this.error.set(err.message || 'Failed to load tasks');
          this.loading.set(false);
        }
      });
  }
}
```

## Advanced Patterns

### Throttle vs Debounce

```typescript
import { throttleTime, debounceTime } from 'rxjs';

// Throttle: Emit first, then ignore for duration
throttled$ = this.clicks$.pipe(
  throttleTime(1000) // Max once per second
);

// Debounce: Wait for quiet period
debounced$ = this.input$.pipe(
  debounceTime(300) // Wait 300ms after last input
);
```

### Scan - Accumulate Values

```typescript
// Running total
total$ = this.amounts$.pipe(
  scan((acc, value) => acc + value, 0)
);

// History accumulation
history$ = this.events$.pipe(
  scan((history, event) => [...history, event], [] as Event[])
);
```

### startWith - Initial Value

```typescript
// Start with loading state
status$ = this.dataLoad$.pipe(
  map(() => 'loaded'),
  startWith('loading')
);
```

### pairwise - Previous + Current

```typescript
// Compare with previous value
changes$ = this.value$.pipe(
  pairwise(),
  map(([prev, curr]) => ({
    previous: prev,
    current: curr,
    diff: curr - prev
  }))
);
```

## Best Practices

### ✅ DO

```typescript
// Use toSignal() for reactive data in templates
data = toSignal(this.data$, { initialValue: [] });

// Use takeUntilDestroyed() for cleanup
this.data$.pipe(takeUntilDestroyed(this.destroyRef)).subscribe();

// Use switchMap for user-triggered requests
search$ = this.query$.pipe(switchMap(q => this.search(q)));

// Handle errors explicitly
data$ = this.http.get('/api/data').pipe(
  catchError(err => of(null))
);
```

### ❌ DON'T

```typescript
// Don't forget to unsubscribe
this.data$.subscribe(); // Memory leak!

// Don't use nested subscribes
this.data$.subscribe(data => {
  this.process(data).subscribe(); // Anti-pattern!
});

// Don't use async pipe with signals
@if (data$ | async) { } // Use signals instead
```

## Checklist

When using RxJS:

- [ ] Use toSignal() to convert Observables to Signals
- [ ] Use takeUntilDestroyed() for subscription cleanup
- [ ] Handle errors with catchError()
- [ ] Debounce user input (300ms)
- [ ] Use switchMap for cancellable requests
- [ ] Share expensive Observables with shareReplay()
- [ ] Provide initial values with startWith()
- [ ] Filter out empty/null values
- [ ] Test async operations
- [ ] Document complex operator chains

## References

- [RxJS Documentation](https://rxjs.dev/)
- [Angular Signals + RxJS Interop](https://angular.dev/guide/signals/rxjs-interop)
- [RxJS Operators](https://rxjs.dev/guide/operators)
- [Learn RxJS](https://www.learnrxjs.io/)