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

// Each operator wraps the previous Observable
of(1, 2, 3, 4, 5).pipe(
  filter(n => n % 2 === 0), // 2, 4
  map(n => n * 10),          // 20, 40
  take(1),                   // 20 (then complete)
).subscribe(console.log);

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

// Transform each emitted value
of({ id: 1, name: 'Alice', age: 30 }).pipe(
  map(user => user.name.toUpperCase())
).subscribe(console.log); // "ALICE"

// With HTTP
this.http.get<ApiResponse>('/api/user/1').pipe(
  map(response => response.data) // unwrap the data field
).subscribe(user => this.user = user);

// Modern equivalent using map for property extraction
users$.pipe(
  map(users => users.map(u => u.name))
).subscribe(names => console.log(names));

import { fromEvent } from 'rxjs';
import { scan } from 'rxjs/operators';

// Count total clicks
fromEvent(document, 'click').pipe(
  scan((count) => count + 1, 0)
).subscribe(count => console.log('Total clicks:', count));

import { of, interval, fromEvent } from 'rxjs';
import { filter, take, takeWhile, skip, distinctUntilChanged, debounceTime, throttleTime } from 'rxjs/operators';

// filter — only pass values matching predicate
of(1, 2, 3, 4, 5).pipe(
  filter(n => n > 3)
).subscribe(console.log); // 4, 5

// take — complete after N values
interval(100).pipe(take(3)).subscribe(console.log); // 0, 1, 2

// takeWhile — complete when condition is false
interval(100).pipe(
  takeWhile(n => n < 5)
).subscribe(console.log); // 0, 1, 2, 3, 4

// skip — ignore first N values
of(1, 2, 3, 4, 5).pipe(skip(2)).subscribe(console.log); // 3, 4, 5

// distinctUntilChanged — suppress consecutive duplicates
of(1, 1, 2, 2, 3).pipe(
  distinctUntilChanged()
).subscribe(console.log); // 1, 2, 3

// debounceTime — emit only after N ms of silence (great for search)
searchInput.valueChanges.pipe(
  debounceTime(300),
  distinctUntilChanged()
).subscribe(term => this.search(term));

// throttleTime — emit max once per N ms (rate limiting)
fromEvent(document, 'mousemove').pipe(
  throttleTime(100)
).subscribe(evt => console.log(evt));

import { switchMap, mergeMap, concatMap, exhaustMap } from 'rxjs/operators';

// switchMap — search: cancel stale request when user types again
searchControl.valueChanges.pipe(
  debounceTime(300),
  switchMap(term => this.http.get<Product[]>(`/api/search?q=${term}`))
).subscribe(results => this.results = results);

// mergeMap — load all user details in parallel
from(userIds).pipe(
  mergeMap(id => this.http.get<User>(`/api/users/${id}`))
).subscribe(user => this.users.push(user));

// concatMap — process items in strict order (e.g., upload queue)
from(filesToUpload).pipe(
  concatMap(file => this.uploadService.upload(file))
).subscribe(result => this.completedUploads.push(result));

// exhaustMap — prevent double-submit
fromEvent(submitButton, 'click').pipe(
  exhaustMap(() => this.http.post('/api/save', this.formData))
).subscribe(response => this.saved = true);

import { combineLatest, forkJoin, zip, merge, concat, withLatestFrom } from 'rxjs';

// forkJoin — wait for ALL to complete, then emit combined result (like Promise.all)
forkJoin({
  users: this.http.get<User[]>('/api/users'),
  roles: this.http.get<Role[]>('/api/roles'),
}).subscribe(({ users, roles }) => {
  this.users = users;
  this.roles = roles;
});

// combineLatest — emit array of latest values when ANY source emits
combineLatest([
  this.filterControl.valueChanges,
  this.sortControl.valueChanges,
  this.pageControl.valueChanges,
]).pipe(
  switchMap(([filter, sort, page]) =>
    this.http.get<Item[]>(`/api/items?filter=${filter}&sort=${sort}&page=${page}`)
  )
).subscribe(items => this.items = items);

// withLatestFrom — combine with latest from secondary stream
saveButton.clicks$.pipe(
  withLatestFrom(this.currentUser$),
  switchMap(([_, user]) => this.saveService.save({ ...formData, userId: user.id }))
).subscribe();

// merge — interleave multiple Observables
merge(
  fromEvent(document, 'click'),
  fromEvent(document, 'keydown')
).subscribe(event => console.log('User action:', event.type));

import { catchError, retry, retryWhen, throwError, EMPTY, of } from 'rxjs';
import { delay, take } from 'rxjs/operators';

// catchError — handle errors, return fallback or re-throw
this.http.get<User[]>('/api/users').pipe(
  catchError((err) => {
    if (err.status === 404) return of([]);        // fallback value
    if (err.status === 401) return EMPTY;          // cancel silently
    return throwError(() => err);                  // re-throw
  })
).subscribe(users => this.users = users);

// retry — retry N times before emitting error
this.http.get('/api/flaky').pipe(
  retry({ count: 3, delay: 1000 }) // 3 retries, 1 second apart
).subscribe();

// retryWhen — custom retry logic (e.g., exponential backoff)
this.http.get('/api/data').pipe(
  retryWhen(errors =>
    errors.pipe(
      scan((retryCount, err) => {
        if (retryCount >= 3) throw err;
        return retryCount + 1;
      }, 0),
      delay(1000)
    )
  )
).subscribe();

import { tap, delay, timeout, finalize, shareReplay } from 'rxjs/operators';

// tap — side effects without transformation (logging, analytics)
this.http.get<User>('/api/user').pipe(
  tap(user => console.log('Fetched user:', user.name)),
  tap(user => this.analytics.track('user_loaded', user.id))
).subscribe(user => this.user = user);

// delay — add artificial delay (useful for testing)
of('hello').pipe(delay(2000)).subscribe(console.log);

// timeout — error if Observable doesn't emit within N ms
this.http.get('/api/data').pipe(
  timeout(5000) // throw TimeoutError if no response in 5s
).subscribe();

// finalize — cleanup on complete OR error (like try/finally)
this.http.get('/api/data').pipe(
  finalize(() => this.isLoading = false) // always runs
).subscribe();

// shareReplay — multicast and cache last N values for late subscribers
const user$ = this.http.get<User>('/api/me').pipe(
  shareReplay(1) // all subscribers share one HTTP call
);

// typeahead.component.ts
import { Component, OnInit, inject } from '@angular/core';
import { FormControl, ReactiveFormsModule } from '@angular/forms';
import { AsyncPipe } from '@angular/common';
import { HttpClient } from '@angular/common/http';
import { Observable, of } from 'rxjs';
import {
  debounceTime, distinctUntilChanged, switchMap,
  filter, catchError, tap
} from 'rxjs/operators';

@Component({
  selector: 'app-typeahead',
  standalone: true,
  imports: [ReactiveFormsModule, AsyncPipe],
  template: `
    <input [formControl]="searchControl" placeholder="Search products..." />

    @if (loading) { <span>Searching...</span> }

    @for (result of results$ | async; track result.id) {
      <div class="result">{{ result.name }}</div>
    }
  `,
})
export class TypeaheadComponent {
  private http = inject(HttpClient);
  searchControl = new FormControl('');
  loading = false;

  results$: Observable<Product[]> = this.searchControl.valueChanges.pipe(
    debounceTime(300),           // wait 300ms after user stops typing
    distinctUntilChanged(),      // don't fire for same value
    filter(term => (term?.length ?? 0) >= 2), // minimum 2 chars
    tap(() => this.loading = true),
    switchMap(term =>            // cancel previous request
      this.http.get<Product[]>(`/api/products/search?q=${term}`).pipe(
        catchError(() => of([]))  // empty results on error
      )
    ),
    tap(() => this.loading = false)
  );
}