import { Observable, of, from, interval, fromEvent, timer } from 'rxjs';

// --- of() — emit a fixed set of values synchronously
const numbers$ = of(1, 2, 3);
numbers$.subscribe(console.log); // 1, 2, 3

// --- from() — convert a Promise or array to an Observable
const promise$ = from(fetch('/api/data').then(r => r.json()));
const array$ = from([10, 20, 30]);

// --- interval() — emit a number every N milliseconds
const tick$ = interval(1000); // 0, 1, 2, 3... every second

// --- fromEvent() — DOM events as an Observable
const clicks$ = fromEvent(document, 'click');

// --- timer() — emit once after a delay, or repeatedly
const delayed$ = timer(3000); // emits 0 after 3 seconds
const delayed2$ = timer(0, 1000); // starts immediately, repeats every second

// --- new Observable() — full manual control
const custom$ = new Observable<number>((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.complete();
});

import { interval } from 'rxjs';

const counter$ = interval(1000);

// subscribe() returns a Subscription object
const subscription = counter$.subscribe({
  next: (value) => console.log('Value:', value),
  error: (err) => console.error('Error:', err),
  complete: () => console.log('Completed'),
});

// Unsubscribe after 5 seconds to prevent memory leaks
setTimeout(() => {
  subscription.unsubscribe();
  console.log('Unsubscribed!');
}, 5000);

// Using async pipe — no manual subscription
@Component({
  template: `
    @if (user$ | async; as user) {
      <h1>{{ user.name }}</h1>
    }
  `,
})
export class UserComponent {
  user$ = this.userService.getUser(1);
}

import { Component, OnInit, inject } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { interval } from 'rxjs';

@Component({ selector: 'app-timer', standalone: true, template: `{{ count }}` })
export class TimerComponent implements OnInit {
  count = 0;
  private destroyRef = inject(DestroyRef); // injected automatically

  ngOnInit(): void {
    interval(1000)
      .pipe(takeUntilDestroyed(this.destroyRef))
      .subscribe(() => this.count++);
  }
}

import { Component, OnInit, OnDestroy } from '@angular/core';
import { Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

@Component({ selector: 'app-example', standalone: true, template: '' })
export class ExampleComponent implements OnDestroy {
  private destroy$ = new Subject<void>();

  init(): void {
    someObservable$
      .pipe(takeUntil(this.destroy$))
      .subscribe(handleValue);
  }

  ngOnDestroy(): void {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

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

// map — transform each value
of(1, 2, 3).pipe(
  map(x => x * 2)
).subscribe(console.log); // 2, 4, 6

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

// tap — side effects without transformation
of(1, 2, 3).pipe(
  tap(x => console.log('Before map:', x)),
  map(x => x * 10),
  tap(x => console.log('After map:', x))
).subscribe();

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

// debounceTime — wait for silence before emitting (search boxes)
searchControl.valueChanges.pipe(
  debounceTime(300),
  distinctUntilChanged()
).subscribe(searchTerm => this.search(searchTerm));

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

// switchMap — cancel previous inner Observable on new emission (search, navigation)
searchControl.valueChanges.pipe(
  debounceTime(300),
  switchMap(term => this.http.get<Result[]>(`/api/search?q=${term}`))
).subscribe(results => this.results = results);

// mergeMap — subscribe to all inner Observables concurrently
from([1, 2, 3]).pipe(
  mergeMap(id => this.http.get<User>(`/api/users/${id}`))
).subscribe(user => console.log(user)); // all 3 requests fire in parallel

// concatMap — wait for each inner Observable to complete before starting the next
from([1, 2, 3]).pipe(
  concatMap(id => this.http.get<User>(`/api/users/${id}`))
).subscribe(user => console.log(user)); // sequential: 1 → 2 → 3

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

// forkJoin — like Promise.all, emits when ALL complete
forkJoin({
  users: this.http.get<User[]>('/api/users'),
  posts: this.http.get<Post[]>('/api/posts'),
}).subscribe(({ users, posts }) => {
  console.log('Users:', users);
  console.log('Posts:', posts);
});

// combineLatest — emits when ANY source emits, with latest values from all
combineLatest([filter$, sort$, page$]).pipe(
  switchMap(([filter, sort, page]) =>
    this.http.get<Product[]>(`/api/products?filter=${filter}&sort=${sort}&page=${page}`)
  )
).subscribe(products => this.products = products);

// merge — interleave emissions from multiple Observables
merge(click$, keydown$).subscribe(event => console.log(event));

import { catchError, retry, of, throwError, EMPTY } from 'rxjs';

this.http.get<User[]>('/api/users').pipe(
  retry(2), // retry twice before error propagates
  catchError((err) => {
    if (err.status === 404) {
      return of([]); // return empty array as fallback
    }
    if (err.status === 401) {
      this.router.navigate(['/login']);
      return EMPTY; // cancel the stream
    }
    return throwError(() => err); // re-throw for higher-level handling
  })
).subscribe({
  next: (users) => this.users = users,
  error: (err) => this.errorMessage = err.message,
});

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

@Component({
  selector: 'app-search',
  standalone: true,
  template: `
    <input (input)="term.set($event.target.value)" placeholder="Search..." />
    @for (result of results(); track result.id) {
      <div>{{ result.name }}</div>
    }
  `,
})
export class SearchComponent {
  private http = inject(HttpClient);

  term = signal('');

  // Convert signal to Observable, pipe through HTTP, back to signal
  results = toSignal(
    toObservable(this.term).pipe(
      debounceTime(300),
      distinctUntilChanged(),
      switchMap(term => this.http.get<Result[]>(`/api/search?q=${term}`))
    ),
    { initialValue: [] }
  );
}