RxJS and Observables Basics in Angular

1import { of, from } from 'rxjs';
2
3// of() - emit values in sequence
4const numbers$ = of(1, 2, 3, 4, 5);
5numbers$.subscribe(value => console.log(value));
6// Output: 1, 2, 3, 4, 5
7
8// from() - convert array, promise, or iterable to observable
9const array$ = from([10, 20, 30]);
10array$.subscribe(value => console.log(value));
11// Output: 10, 20, 30
12
13const promise$ = from(fetch('/api/data'));
14promise$.subscribe(response => console.log(response));

1import { interval, timer } from 'rxjs';
2
3// interval() - emit values at specified interval
4const counter$ = interval(1000); // Emit every second
5counter$.subscribe(value => console.log(value));
6// Output: 0, 1, 2, 3, 4...
7
8// timer() - emit after delay, then optionally at intervals
9const delayed$ = timer(2000); // Emit after 2 seconds
10delayed$.subscribe(() => console.log('Delayed!'));
11
12const periodic$ = timer(1000, 500); // Delay 1s, then every 500ms
13periodic$.subscribe(value => console.log(value));

1import { Observable } from 'rxjs';
2
3const custom$ = new Observable(subscriber => {
4  // Emit values
5  subscriber.next('First');
6  subscriber.next('Second');
7
8  // Async emission
9  setTimeout(() => {
10    subscriber.next('Delayed');
11    subscriber.complete(); // Signal completion
12  }, 1000);
13
14  // Return cleanup function
15  return () => {
16    console.log('Observable unsubscribed');
17  };
18});
19
20const subscription = custom$.subscribe({
21  next: value => console.log(value),
22  error: err => console.error(err),
23  complete: () => console.log('Complete!')
24});

1import { of } from 'rxjs';
2import { map } from 'rxjs/operators';
3
4const numbers$ = of(1, 2, 3, 4, 5);
5const squared$ = numbers$.pipe(
6  map(x => x * x)
7);
8
9squared$.subscribe(value => console.log(value));
10// Output: 1, 4, 9, 16, 25

1import { of } from 'rxjs';
2import { filter } from 'rxjs/operators';
3
4const numbers$ = of(1, 2, 3, 4, 5, 6);
5const even$ = numbers$.pipe(
6  filter(x => x % 2 === 0)
7);
8
9even$.subscribe(value => console.log(value));
10// Output: 2, 4, 6

1import { of } from 'rxjs';
2import { tap, map } from 'rxjs/operators';
3
4const numbers$ = of(1, 2, 3);
5numbers$.pipe(
6  tap(x => console.log('Before:', x)),
7  map(x => x * 2),
8  tap(x => console.log('After:', x))
9).subscribe();
10// Output:
11// Before: 1
12// After: 2
13// Before: 2
14// After: 4...

1import { fromEvent } from 'rxjs';
2import { switchMap, debounceTime } from 'rxjs/operators';
3
4// Search as user types
5const searchBox = document.getElementById('search');
6const search$ = fromEvent(searchBox, 'input').pipe(
7  debounceTime(300), // Wait 300ms after typing stops
8  switchMap(event => {
9    const query = event.target.value;
10    return fetch(\`/api/search?q=\${query}\`).then(r => r.json());
11  })
12);
13
14search$.subscribe(results => console.log(results));

1import { of, throwError } from 'rxjs';
2import { catchError, map } from 'rxjs/operators';
3
4const api$ = fetchData().pipe(
5  map(data => data.value),
6  catchError(error => {
7    console.error('Error occurred:', error);
8    return of({ value: 'default' }); // Return fallback
9  })
10);
11
12api$.subscribe(value => console.log(value));

1import { combineLatest, of } from 'rxjs';
2
3const name$ = of('John');
4const age$ = of(30);
5const city$ = of('New York');
6
7combineLatest([name$, age$, city$]).subscribe(
8  ([name, age, city]) => {
9    console.log(\`\${name}, \${age}, from \${city}\`);
10  }
11);
12// Output: John, 30, from New York

1import { Injectable, inject } from '@angular/core';
2import { HttpClient } from '@angular/common/http';
3import { Observable } from 'rxjs';
4import { map, catchError } from 'rxjs/operators';
5
6@Injectable({ providedIn: 'root' })
7export class UserService {
8  private http = inject(HttpClient);
9
10  getUsers(): Observable<User[]> {
11    return this.http.get<User[]>('/api/users').pipe(
12      map(users => users.sort((a, b) => a.name.localeCompare(b.name))),
13      catchError(error => {
14        console.error('Failed to load users:', error);
15        return of([]); // Return empty array on error
16      })
17    );
18  }
19
20  getUser(id: number): Observable<User> {
21    return this.http.get<User>(\`/api/users/\${id}\`);
22  }
23}

1import { Component, inject } from '@angular/core';
2import { CommonModule } from '@angular/common';
3import { UserService } from './user.service';
4
5@Component({
6  selector: 'app-user-list',
7  standalone: true,
8  imports: [CommonModule],
9  template: \`
10    <div>
11      @if (users$ | async; as users) {
12        @for (user of users; track user.id) {
13          <div class="user-card">
14            {{ user.name }} - {{ user.email }}
15          </div>
16        } @empty {
17          <p>No users found</p>
18        }
19      }
20
21      <!-- Loading state -->
22      @if (loading$ | async) {
23        <div class="spinner">Loading...</div>
24      }
25    </div>
26  \`
27})
28export class UserListComponent {
29  private userService = inject(UserService);
30
31  users$ = this.userService.getUsers();
32  loading$ = of(true).pipe(delay(2000), map(() => false));
33}

1import { Subject, BehaviorSubject, ReplaySubject } from 'rxjs';
2
3// Subject - no initial value, no replay
4const subject = new Subject<number>();
5subject.next(1);
6subject.subscribe(v => console.log('A:', v));
7subject.next(2); // A: 2
8subject.next(3); // A: 3
9
10// BehaviorSubject - requires initial value, replays last value
11const behavior = new BehaviorSubject<number>(0);
12behavior.next(1);
13behavior.next(2);
14behavior.subscribe(v => console.log('B:', v)); // B: 2 (gets last value)
15behavior.next(3); // B: 3
16
17// ReplaySubject - replays specified number of values
18const replay = new ReplaySubject<number>(2); // Replay last 2
19replay.next(1);
20replay.next(2);
21replay.next(3);
22replay.subscribe(v => console.log('C:', v));
23// C: 2, C: 3 (gets last 2 values)

1import { Injectable } from '@angular/core';
2import { BehaviorSubject, Observable } from 'rxjs';
3
4@Injectable({ providedIn: 'root' })
5export class StateService {
6  private stateSubject = new BehaviorSubject<AppState>({
7    user: null,
8    loading: false
9  });
10
11  // Expose as observable (hide subject methods)
12  state$: Observable<AppState> = this.stateSubject.asObservable();
13
14  updateUser(user: User): void {
15    const currentState = this.stateSubject.value;
16    this.stateSubject.next({
17      ...currentState,
18      user
19    });
20  }
21
22  setLoading(loading: boolean): void {
23    const currentState = this.stateSubject.value;
24    this.stateSubject.next({
25      ...currentState,
26      loading
27    });
28  }
29}

1import { Component, OnInit, OnDestroy } from '@angular/core';
2import { Subscription } from 'rxjs';
3
4@Component({...})
5export class MyComponent implements OnInit, OnDestroy {
6  private subscription = new Subscription();
7
8  ngOnInit() {
9    // Add all subscriptions
10    this.subscription.add(
11      this.service.getData().subscribe(data => {
12        // Handle data
13      })
14    );
15
16    this.subscription.add(
17      this.service.getOtherData().subscribe(data => {
18        // Handle other data
19      })
20    );
21  }
22
23  ngOnDestroy() {
24    // Unsubscribe all at once
25    this.subscription.unsubscribe();
26  }
27}

1import { Component, OnInit, inject } from '@angular/core';
2import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
3
4@Component({...})
5export class MyComponent implements OnInit {
6  private service = inject(DataService);
7  private destroyRef = inject(DestroyRef);
8
9  ngOnInit() {
10    // Automatically unsubscribes when component is destroyed
11    this.service.getData()
12      .pipe(takeUntilDestroyed(this.destroyRef))
13      .subscribe(data => {
14        // Handle data
15      });
16  }
17}