Angular RxJS Basics

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

// Create an Observable that emits 3 values
const myObservable$ = new Observable<number>(observer => {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete(); // No more values
});

// Subscribe to start receiving values
myObservable$.subscribe({
  next: value => console.log(&#039;Received:&#039;, value),
  error: err => console.error(&#039;Error:&#039;, err),
  complete: () => console.log(&#039;Stream completed!&#039;)
});
// Output: Received: 1, Received: 2, Received: 3, Stream completed!

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

// of() — emit values immediately
of(1, 2, 3).subscribe(v => console.log(v));

// from() — emit each element of an array or Promise
from([10, 20, 30]).subscribe(v => console.log(v));
from(fetch('/api/users')).subscribe(); // Converts Promise to Observable

// interval() — emit incrementing number every N milliseconds
interval(1000).subscribe(i => console.log('Tick:', i)); // 0, 1, 2...

// fromEvent() — DOM events as Observables
fromEvent(document, 'click').subscribe(event => console.log('Clicked!', event));

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

of(1, 2, 3, 4, 5, 6)
  .pipe(
    filter(n => n % 2 === 0),   // Keep only even: 2, 4, 6
    map(n => n * 10),           // Multiply by 10: 20, 40, 60
    take(2)                     // Only take 2 values: 20, 40
  )
  .subscribe(result => console.log(result));
// Output: 20, 40

this.http.get<any[]>(&#039;/api/products&#039;).pipe(
  map(products => products.filter(p => p.inStock))
).subscribe(inStockProducts => this.products = inStockProducts);

this.searchInput.valueChanges.pipe(
  debounceTime(300),                          // Wait 300ms after user stops typing
  distinctUntilChanged(),                     // Only if value actually changed
  switchMap(term => this.apiService.search(term)) // Cancel old, start new request
).subscribe(results => this.results = results);

import { forkJoin } from 'rxjs';

// Make two API calls simultaneously, wait for both
forkJoin({
  user: this.http.get('/api/user/1'),
  products: this.http.get('/api/products')
}).subscribe(({ user, products }) => {
  console.log('Both loaded:', user, products);
});

import { Component, OnInit } from '@angular/core';
import { FormControl } from '@angular/forms';
import { debounceTime, distinctUntilChanged, switchMap } from 'rxjs/operators';

@Component({
  selector: 'app-search',
  template: `
    <input [formControl]="searchCtrl" placeholder="Search products..." />
    <ul>
      <li *ngFor="let result of results">{{ result.name }}</li>
    </ul>
  `
})
export class SearchComponent implements OnInit {
  searchCtrl = new FormControl(&#039;&#039;);
  results: any[] = [];

  constructor(private productService: ProductService) {}

  ngOnInit(): void {
    this.searchCtrl.valueChanges.pipe(
      debounceTime(400),
      distinctUntilChanged(),
      switchMap(term => this.productService.search(term || &#039;&#039;))
    ).subscribe(results => this.results = results);
  }
}