rxjs-cheatsheet.ts

import { Subject, BehaviorSubject, ReplaySubject, Observable } from 'rxjs'; // Subjects and Observable
import { Subscription } from "rxjs"; // Subscription
import { combineLatest, of, pipe, } from "rxjs"; // Functions
import { delay, filter, map, share, switchMap, tap } from 'rxjs/operators'; // Operators
import { fromFetch } from "rxjs/fetch"; // Fetch subject (a special subject)

////////////////////////////////////////////////////////////////////////////////////////////////
// Creating an subject.
////////////////////////////////////////////////////////////////////////////////////////////////
const source$ = new Subject<number>();
const replayEverything$ = new ReplaySubject<number>(); // Replay all values on subscribe
const replayLastTen$ = new ReplaySubject<number>(10); // Replay the last 10 values on subscribe
const alwaysValue$ = new BehaviorSubject<number>(10); // Always starts with a value.

////////////////////////////////////////////////////////////////////////////////////////////////
// Piping the data.
////////////////////////////////////////////////////////////////////////////////////////////////
const double$ = source$.pipe(map(x => x * 2));
const quadruple = double$.pipe(map(x => x * 2));

// Multiple pipes and combining observables
const doubleTimesQuadrubple$ = combineLatest([double$, quadruple]).pipe(
  map(([doubleValue, quadruple]) => { return doubleValue * quadruple; }), // Passing multiple values
  tap(x => console.log("Before multiplied by π: " + x)),  // Without body aka single statement
  map(x => x * Math.PI), // Our change
  tap(x => { console.log("After multiplied by π: " + x) }), // With body
);

////////////////////////////////////////////////////////////////////////////////////////////////
// Sharing is caring
////////////////////////////////////////////////////////////////////////////////////////////////
const slowObs$ = source$.pipe(
  switchMap(x => invokeSlowApi(x)), // Invoking the API
);

const slowInc$ = slowObs$.pipe(map(x => x + 1));
const slowDec$ = slowObs$.pipe(map(x => x - 1));
const slowAvg$ = combineLatest([slowInc$, slowDec$]).pipe(
  map(([x, y]) => (x + y) / 2)
); // Invokes the API twice.

const shareObs$ = slowObs$.pipe(share()); // We add a sharing operator.
const fastInc$ = shareObs$.pipe(map(x => x + 1));
const fastDec$ = shareObs$.pipe(map(x => x - 1));
const fastAvg$ = combineLatest([fastInc$, fastDec$]).pipe(
  map(([x, y]) => (x + y) / 2)
); // Invokes the API once.

// NOTE: only use `share()` when you need it for performance reason. Don't use it after every 
//       API-call (you will have a bad time and I speak from personal experience).

////////////////////////////////////////////////////////////////////////////////////////////////
// SwitchMap is switching over to another subscription
////////////////////////////////////////////////////////////////////////////////////////////////
const countMode$ = new BehaviorSubject("numbers");
const numbers$ = of([0, 1, 2, 3, 4, 5]).pipe(delay(1000));
const letters$ = of(["a", "b", "c", "d", "e", "f"]).pipe(delay(1000));

countMode$.pipe(
  switchMap(mode => {
    if (mode == "numbers") return numbers$;
    else return letters$;
  })
);

// Assuming you have subscribed it will start
// > 0
// > 1
// > 3
// > 4
// > 5
// And then stops.

countMode$.next("numbers");
// Will restart the count
// > 0
// > 1...5

countMode$.next("letters");
// Will stop the current count and switch to letters
// > "a"
// > "b"
// > "c" ... "f"
// And stops.

// A practical example.
const refresh$ = new BehaviorSubject(undefined);
const user$ = refresh$.pipe(switchMap(() => fromFetch("/api/user/me")));
const roles$ = refresh$.pipe(switchMap(() => fromFetch("/api/user/me/roles")));

function refreshPage() {
  refresh$.next(undefined);
}


////////////////////////////////////////////////////////////////////////////////////////////////
// Subscriptions & lifetime.
////////////////////////////////////////////////////////////////////////////////////////////////
// Please note:
// - Creating an observable, will not do anything.
// - Only when you subscribe (in angular by using the async-pipe) the data will be used.
// - Unless you `share()` your pipe, all subscriptions start at the source and go through each
//   operator in your pipes.
var lastSrcVal = 0;
const storeObs$ = source$.pipe(tap(x => lastSrcVal = x));
source$.next(1); // No subscribers, so lastSrcVal remains `0`.

const srcSub1: Subscription = storeObs$.subscribe(); // Shortest way to subscribe.
source$.next(2); // Now lastSrcVal is changed from `0` to `2` (since source$ is not a replaySubject)

// Modern subscribe on a single line
const srcSub2: Subscription = source$.pipe(tap(x => lastSrcVal = x)).subscribe();
// Classic subscribe on a single line
const srcSub3: Subscription = source$.subscribe(x => lastSrcVal = x);

source$.next(3); // lastSrcVal will now be set to value `3` three times due to 3 subscriptions.

srcSub1.unsubscribe(); // Remove a subscription
source$.next(4); // lastSrcVal will now be set to value `4` two times due to 2 subscriptions.

// Make active subscription part of another subscription
const superSub = new Subscription();
superSub.add(srcSub2);
superSub.add(srcSub3);

// So that we can UNsubscribe in one step
// Same as calling the following manually: `srcSub2.unsubscribe(); srcSub3.unsubscribe();`
superSub.unsubscribe();

source$.next(5); // lastSrcVal remains 4 since all subscribers are done.

////////////////////////////////////////////////////////////////////////////////////////////////
// Creating your own operators (please AVOID doing this.)
const canBeDividedByTwo = (x: number) => (x % 2) == 0;
const canBeDividedBySix = (x: number) => (x % 6) == 0;
function canBeDividedByTwoAndSix(ignoreSix: boolean) {
  if (ignoreSix) {
    return filter(canBeDividedByTwo);
  }

  // Your own combined operator.
  return pipe(
    filter(canBeDividedByTwo), // We filter the data.
    filter(canBeDividedBySix),
  );
}

// Using your own operator
const normalThatCanBeDividedByTwoAndSix$ = source$.pipe(
  canBeDividedByTwoAndSix(false),
);


////////////////////////////////////////////////////////////////////////////////////////////////
// Angular: Async pipe
////////////////////////////////////////////////////////////////////////////////////////////////
// The following code in  HTML: `<div>{{source$ | async}}</div>`
// does the following:
// - Create a subscription after `ngOnInit()` is completed
// - **AND** unsubscribes when the component is no longer needed.


////////////////////////////////////////////////////////////////////////////////////////////////
// DUMMY CODE
////////////////////////////////////////////////////////////////////////////////////////////////
// Simulating a slow API call
function invokeSlowApi(x: number): Observable<number> {
  return of(x).pipe(
    tap(() => console.log("Invoking slow api")),
    delay(2000),
  )
}