Angular RxJS: Reactive Programming

import { Observable } from 'rxjs';

// Creating observables
const data$ = new Observable(subscriber => {
  subscriber.next('Hello');
  subscriber.next('World');
  subscriber.complete();
});

// Subscribing
data$.subscribe({
  next: value => console.log(value),
  error: err => console.error(err),
  complete: () => console.log('Complete')
});

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

// of
const numbers$ = of(1, 2, 3, 4, 5);

// from
const array$ = from([1, 2, 3, 4, 5]);
const promise$ = from(fetch('/api/users'));

// interval
const seconds$ = interval(1000);

// timer
const delayed$ = timer(1000);
const delayedInterval$ = timer(1000, 500);

// fromEvent
const click$ = fromEvent(document, 'click');

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

numbers$
  .pipe(
    filter(n => n % 2 === 0),
    map(n => n * 2),
    tap(n => console.log('Processing:', n))
  )
  .subscribe(result => console.log(result));

import { map } from 'rxjs/operators';

users$
  .pipe(
    map(users => users.map(user => user.name))
  )
  .subscribe(names => console.log(names));

import { filter } from 'rxjs/operators';

numbers$
  .pipe(
    filter(n => n > 5)
  )
  .subscribe(largeNumbers => console.log(largeNumbers));

import { switchMap } from 'rxjs/operators';

userId$
  .pipe(
    switchMap(id => this.http.get(`/api/users/${id}`))
  )
  .subscribe(user => console.log(user));

import { mergeMap } from 'rxjs/operators';

ids$
  .pipe(
    mergeMap(id => this.http.get(`/api/items/${id}`))
  )
  .subscribe(item => console.log(item));

import { debounceTime } from 'rxjs/operators';

searchInput$
  .pipe(
    debounceTime(300),
    distinctUntilChanged(),
    switchMap(query => this.search(query))
  )
  .subscribe(results => console.log(results));

import { catchError, of } from 'rxjs/operators';

this.http.get('/api/users')
  .pipe(
    catchError(error => {
      console.error('Error:', error);
      return of([]);
    })
  )
  .subscribe(users => console.log(users));

import { BehaviorSubject } from 'rxjs';

const user$ = new BehaviorSubject<User | null>(null);

user$.subscribe(user => console.log('User:', user));

user$.next({ name: 'John' });
user$.next({ name: 'Jane' });

// Get current value
console.log(user$.value);

import { ReplaySubject } from 'rxjs';

const events$ = new ReplaySubject(3); // Buffer last 3 values

events$.next('Event 1');
events$.next('Event 2');
events$.next('Event 3');

// Late subscribers receive buffered values
events$.subscribe(event => console.log(event));

import { AsyncSubject } from 'rxjs';

const result$ = new AsyncSubject();

result$.next('Value 1');
result$.next('Value 2');
result$.complete(); // Only emits last value on complete

// Cold - Each subscriber gets independent execution
const cold$ = new Observable(subscriber => {
  subscriber.next(Math.random());
});

// Hot - All subscribers share same execution
const hot$ = cold$.pipe(share());

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

this.http.get('/api/data')
  .pipe(
    retry(3),
    catchError(error => {
      if (error.status === 404) {
        return of([]);
      }
      return throwError(() => error);
    })
  )
  .subscribe();

@Injectable({ providedIn: 'root' })
export class UserService {
  constructor(private http: HttpClient) {}

  getUsers(): Observable<User[]> {
    return this.http.get<User[]>('/api/users')
      .pipe(
        catchError(this.handleError)
      );
  }

  getUser(id: number): Observable<User> {
    return this.http.get<User>(`/api/users/${id}`);
  }

  private handleError(error: HttpErrorResponse) {
    console.error('API Error:', error);
    return throwError(() => error);
  }
}

@Component({
  selector: 'app-users',
  template: `
    <div *ngFor="let user of users$ | async">
      {{ user.name }}
    </div>
  `
})
export class UsersComponent implements OnInit, OnDestroy {
  users$ = this.userService.getUsers();
  private destroy$ = new Subject<void>();

  constructor(
    private userService: UserService,
    private route: ActivatedRoute
  ) {}

  ngOnInit() {
    this.route.params
      .pipe(
        takeUntil(this.destroy$),
        switchMap(params => this.userService.getUser(params['id']))
      )
      .subscribe(user => console.log(user));
  }

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