Observables and Operators

import { of } from 'rxjs';
import { map, filter } from 'rxjs/operators';

of(1, 2, 3, 4, 5).pipe(
  filter(n => n % 2 === 0),   // Keep only even numbers
  map(n => n * 10),            // Multiply by 10
).subscribe(console.log);
// 20, 40

source$.pipe(
  operator1(),
  operator2(),
  operator3(),
).subscribe(result => { /* ... */ });

import { map } from 'rxjs/operators';

// Transform API response
this.http.get<ApiResponse>('/api/users').pipe(
  map(response => response.data),           // Extract data
  map(users => users.filter(u => u.active)), // Filter active users
);

// Simple transforms
of(1, 2, 3).pipe(
  map(n => n * n), // Square each number
).subscribe(console.log); // 1, 4, 9

import { switchMap } from 'rxjs/operators';

// Type-ahead search: each keystroke cancels the previous request
this.searchControl.valueChanges.pipe(
  debounceTime(300),
  distinctUntilChanged(),
  switchMap(query => this.searchService.search(query)),
).subscribe(results => this.results = results);

import { mergeMap } from 'rxjs/operators';

// Process all items in parallel
from(userIds).pipe(
  mergeMap(id => this.userService.getUser(id)),
).subscribe(user => console.log(user));

import { concatMap } from 'rxjs/operators';

// Execute uploads one at a time
from(files).pipe(
  concatMap(file => this.uploadService.upload(file)),
).subscribe(result => console.log('Uploaded:', result));

import { filter } from 'rxjs/operators';

numbers$.pipe(
  filter(n => n > 10),
).subscribe(console.log); // Only values > 10

import { distinctUntilChanged } from 'rxjs/operators';

of(1, 1, 2, 2, 3, 1).pipe(
  distinctUntilChanged(),
).subscribe(console.log); // 1, 2, 3, 1

// With custom comparison
users$.pipe(
  distinctUntilChanged((prev, curr) => prev.id === curr.id),
);

import { take, takeUntil, takeWhile } from 'rxjs/operators';

// Take first 3 values then complete
interval(1000).pipe(take(3)).subscribe(console.log);
// 0, 1, 2

// Take until another Observable emits
interval(1000).pipe(
  takeUntil(this.destroy$),
).subscribe(console.log);

// Take while condition is true
of(1, 2, 3, 4, 5, 1, 2).pipe(
  takeWhile(n => n < 4),
).subscribe(console.log); // 1, 2, 3

import { skip } from 'rxjs/operators';

interval(1000).pipe(
  skip(3), // Skip first 3 emissions
).subscribe(console.log); // Starts from 3

import { debounceTime } from 'rxjs/operators';

// Wait 300ms after the user stops typing
this.searchInput.valueChanges.pipe(
  debounceTime(300),
).subscribe(query => this.search(query));

import { throttleTime } from 'rxjs/operators';

// Only emit once every 1000ms
fromEvent(window, 'scroll').pipe(
  throttleTime(1000),
).subscribe(() => this.checkScrollPosition());

import { combineLatest } from 'rxjs';

const filters$ = combineLatest([
  this.categoryFilter$,
  this.priceFilter$,
  this.sortOrder$,
]).pipe(
  map(([category, price, sort]) => ({ category, price, sort })),
);

filters$.subscribe(filters => this.loadProducts(filters));

import { forkJoin } from 'rxjs';

// Load multiple API calls in parallel, wait for all
forkJoin({
  users: this.http.get<User[]>('/api/users'),
  products: this.http.get<Product[]>('/api/products'),
  settings: this.http.get<Settings>('/api/settings'),
}).subscribe(({ users, products, settings }) => {
  this.users = users;
  this.products = products;
  this.settings = settings;
});

import { merge } from 'rxjs';

// Combine multiple event sources
const allClicks$ = merge(
  fromEvent(button1, 'click'),
  fromEvent(button2, 'click'),
  fromEvent(button3, 'click'),
);

allClicks$.subscribe(() => console.log('A button was clicked'));

import { catchError, of } from 'rxjs';

this.http.get<User[]>('/api/users').pipe(
  catchError(error => {
    console.error('Failed to load users:', error);
    return of([]); // Return empty array as fallback
  }),
).subscribe(users => this.users = users);

import { retry, catchError } from 'rxjs';

this.http.get('/api/data').pipe(
  retry(3),                    // Retry up to 3 times
  catchError(err => of(null)), // Fallback after retries exhausted
);

import { retryWhen, delay, take } from 'rxjs/operators';

this.http.get('/api/data').pipe(
  retryWhen(errors => errors.pipe(
    delay(2000), // Wait 2 seconds between retries
    take(3),     // Maximum 3 retries
  )),
);

import { tap } from 'rxjs/operators';

this.http.get<User[]>('/api/users').pipe(
  tap(users => console.log('Raw response:', users)),
  map(users => users.filter(u => u.active)),
  tap(active => console.log('Active users:', active.length)),
);

import { finalize } from 'rxjs/operators';

this.loading = true;
this.http.get('/api/data').pipe(
  finalize(() => this.loading = false), // Always runs
).subscribe(data => this.data = data);

import { shareReplay } from 'rxjs/operators';

// Cache the result and share with multiple subscribers
@Injectable({ providedIn: 'root' })
export class ConfigService {
  config$ = this.http.get<Config>('/api/config').pipe(
    shareReplay(1), // Cache last emission, share across subscribers
  );
}

export class SearchComponent {
  private searchService = inject(SearchService);
  searchControl = new FormControl('');

  results$ = this.searchControl.valueChanges.pipe(
    debounceTime(300),              // Wait for typing pause
    distinctUntilChanged(),          // Skip if same value
    filter(query => query.length >= 2), // Minimum 2 chars
    tap(() => this.loading = true),
    switchMap(query =>
      this.searchService.search(query).pipe(
        catchError(() => of([])),    // Return empty on error
      )
    ),
    tap(() => this.loading = false),
  );

  loading = false;
}