Agenda

Interactive diagrams of Rx Observables

Code Example

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

const obs = of(1, 2, 3, 4, 5)
// pipe different operations
.pipe(
	filter(n => n % 2 !== 0),
	map(n => n * n)
	);

	// Subscribe to get values
	obs.subscribe(
		x => console.log(x),
		err => console.error("An error occurred", err),
		_ => console.log("Observable completed")
		);

switchMap

Typical usecase: HTTP request returns value which is needed for following HTTP request

Example: Router ID includes an ID which we need for an HTTP method call parameter

The switchMap operator also cancels previous in-flight requests

ngOnInit() {
	this.hero$ = this.route.paramMap.pipe(
	  switchMap((params: ParamMap) =>
		this.service.getHero(params.get('id')))
	);
}

• map() --> Similar to Array.map
• filter() --> Similar to Array.filter
• combineLatest() --> Combines multiple Observables to create an Observable whose values are calculated from the latest values of each of its input Observables.
• tap() --> Intercepts each emission on the source and runs a function, but returns an output which is identical to the source as long as errors don't occur.

Creational Observable

• of(1) --> Emits 1 and completes
• from([1, 2, 3]) --> Emits 1, 2, 3 and then completes
• empty() --> Completes directly (useful for tests/mocks)

Subjects are observer and observable at the same time

	import { Injectable } from '@angular/core';
	import { Observable, BehaviorSubject } from 'rxjs';
	
	@Injectable({
	  providedIn: 'root'
	})
	export class UserService {
	  get user$(): Observable {
		return this.usersSubject.asObservable();
	  }
	  
	  private users = [];
	  private usersSubject: BehaviorSubject = new BehaviorSubject(this.users);
	
	  constructor() {}
	
	  addUser(userName: string): void {
		this.usersSubject.next([...this.users, userName]);
	  }
	}

Promises (Creation and subscription)

Promises execute immediately, and just once

The computation of the result is initiated when the promise is created

There is no way to restart work

// initiate execution
new Promise((resolve, reject) => { executer_fn });
// handle return value
promise.then((value) => {
	// handle result here
});

Promises Chaining

Promises do not differentiate between the last .then clauses (equivalent to subscription) and intermediate .then clauses (equivalent to map).

promise.then((v) => 2*v);

Promise Cancellation

Promises are not cancellable

Real World Example

import { ajax } from 'rxjs/ajax';
import { map, catchError, retry } from 'rxjs/operators';

// Return "response" from the API. If an error happens,
// return an empty array.
const apiData = ajax('/api/data').pipe(
  retry(3), // Retry up to 3 times before failing
  map(res => {
	if (!res.response) {
	  throw new Error('Value expected!');
	}
	return res.response;
  }),
  catchError(err => of([]))
);

apiData.subscribe({
  next(x) { console.log('data: ', x); },
  error(err) { console.log('errors already caught... will not run'); }
});

Promises Error Handling

Promises push errors to the child promises.

promise.then(() => {
	throw Error('my error');
});

Router and HTTP need no manual unsubscribe

For other observables we can use:

• unsubscribe over the subscription object
• using takeUntil operator
• using async pipe

unsubscribe over the subscription object

export class UnsubscribeComponent implements OnInit, OnDestroy {
  subscription: Subscription | undefined;
  constructor(private upperCaseService: UpperCaseService) {}
 
  ngOnInit() {
    this.subscription = this.upperCaseService.getUpperCaseMessage()
      .subscribe((message: string) => this.message = message);
  }
 
  ngOnDestroy(): void {
    if (this.subscription) {
      this.subscription.unsubscribe();
    }
  }
}

takeUntil operator

export class TakeUntilComponent implements OnInit, OnDestroy {
  message: string;
  private unsubscribe$ = new Subject();
  constructor(private upperCaseService: UpperCaseService) {}
 
  ngOnInit() {
    this.upperCaseService.getUpperCaseMessage()
      .takeUntil(this.unsubscribe$)
      .subscribe((message: string) => this.message = message);
  }
 
  ngOnDestroy(): void {
    this.unsubscribe$.next();
    this.unsubscribe$.complete();
  }
}

Async Pipe

Subscribes to an observable or promise and returns the latest value it has emitted

Angular automatically unsubscribes if component is destroyed

@Component({
	selector: 'async-observable-pipe',
	template: `
Time: {{ time | async }}
`
})
export class AsyncObservablePipeComponent {
	time = new Observable(observer =>
	  setInterval(() => observer.next(new Date().toString()), 1000)
	);
}