RxJS Presentation

autocomplete-example.ts

import { Observable, fromEvent } from 'rxjs';
import { ajax } from 'rxjs/ajax';
import { map, debounceTime, switchMap } from 'rxjs/operators';

function fetchSuggestions(text: string): Observable<string[]> {
  return ajax.getJSON(`https://example.com/suggestions?query=${encodeURL(text)}`);
}
  
const input = document.querySelector('input');
const completions$ = fromEvent(input, 'keyup').pipe(
  debounceTime(100),                          // prevent many consecutive requests
  map(x => x.target.value),                   // map to input text
  filter(x => x.length > 2),                  // filter text until you have enough character
  switchMap(text => fetchSuggestions(text)),  // switchMap to get the suggestions from the api 
);

completions$.subscribe(results => showResults(results)); // show results however you want

concepts.md

Intro

• rxjs - who when and what.
• what's async
• sync/async single/multi table
• pull vs push (observable is push, calling method is pull)
• promise compared to observable
• creating observables
  • what is an observer, subscriber (pass to observable ctor)
  • interval
  • wrapping btn.addEventListener
• operators (map, filter, tap)
• flatteners (Array<Array> -> Array - Observable<Observable> -> Observable)
• completion (via subscriber.complete() or subscription.unsubscribe())
• html/js examples - fromEvent
• autocomplete
• drag & drop
• Subject & Friends
• Observables are templates! No subscribe - no service!
  • cold obs -> create underlying subscribe for every subscription
  • hot obs -> create only once

create-an-interval-observable.ts

function interval(period: number): Observable<number> {
  // When you create an observable, you set up a `subscribe` method,
  // Which calls the methods of the subscribing `observer`.
  return new Observable((observer: Observer<number>): TeardownLogic => {
    let counter = 0;
    const token = setInterval(() => {
      observer.next(counter++);
    }, period);
  
    // You may return a `() => void` method, which will be called
    // When `Subscription#unsubscribe` is called for this observable
    return () => clearInterval(token);
  });
}
  
// Usage:
  
const obs$ = interval(1000).pipe(
  filter(v => v % 2 === 0),
);
  
const subscription = obs$.subscribe(v => console.log(`Value: ${v}`));

setTimeout(() => {
  subscription.unsubscribe();
}, 2000);

links.md

• The presentation code repo
• Presentation download links
• https://dev.to/aturingmachine/an-introduction-to-rxjs-observables-52ln
• https://github.com/blove/rxjs-intro
• https://rxmarbles.com/
• @angular/common/http - Creation of Http Request Observable
• RxJS v6.0
• Learn RxJS by building your own Observable
• Start a new Stackblitz RxJS Project

means-of-asynchronous-events.md

Name	Example	Value notification	Error notification	Completion notification	Aborting
Event Listeners	DOM events	✅	❌	❌	✅
Promises	fetch	✅	✅	❌	❌
Event Emitter	Node.js Readable Stream	✅	✅	✅	✅

Event listeners

const element = document.querySelector('button');
const handleEvent = event => console.log(event.clientX);

element.addEventListener('click', handleEvent);

setTimeout(() => {
  element.removeEventListener('click', handleEvent);
}, 2000);

Promises

fetch('https://jsonplaceholder.typicode.com/todos/1')
  .then(response => response.json())
  .then(json => console.log(json))
  .catch(err => console.error(err));

Node.js streams

const stream = fs.createReadStream('myfile.txt');
stream
  .on('data', data => console.log(data))
  .on('error', err => console.log(err))
  .on('end', () => console.log('Done'));

setTimeout(() => {
  stream.destroy();
}, 2000);

terminology.md

Terminology

Observable

Observables are a stream of events. They can be thought of as an array, except its items are received over time (if at all).

Observables can emit 3 types of events:

1. next: when a value is emitted
2. error: when an error occures
3. complete: when the stream has ended

Observables can be subscribed to. Subscribing to an observable is like calling forEach on an array, except the callback is called whenever an element is received.

When subscribing to an Observable, we pass it an Observer, and receive back a Subscription.

Observer

The event handler in the observable world. Can be represented by the following interface, where each field is the type of event being handled:

interface Observer<T> {
    next: (value: T) => void;
    error: (err: any) => void;
    complete: () => void;
}

Subscription

A Subscription is very much like the token you receive from setInterval, on which you can call clearInterval to close the interval.

A Subscription has an unsubscribe method, which can be called to unsubscribe from an observable. It is when an observable is unsubscribed from, when it can free whatever resources it has (f.e: call clearInterval).

unsubscribing a subscription does not trigger completion.

Operators

Like arrays have their operators (f.e: map, filter, and reduce), which take an array and return a new one, Observables have operators as well, which take an Observable, and return a new one.

Lets take the map operator. Calling arr.map(x => x * 2) on an array, will return a new array, with all of arrs items multiplied by 2.

Calling obs.pipe(map(x => x * 2)) returns a new Observable, where each emitted value, is the one emitted from obs multiplied by 2.

Operators are the bread & butter of Observables. Without them, an Observable is like a simple event handler. But using operators, and combining them in the right way, you can handle extremely complex situations, with very little effort, as RxJS provides a hefty suite of operators out of the box.

Subject

Subjects are Observables, literally. They inherit from Observables and act the exact same way, Except that Subjects can be thought of as a Read-Write version, whereas Observables are Read-Only.

Subjects are way to imperatively emit events. Subjects also act like Observers. You can pass a Subject to a subscribe call, and the Subject's next, error and complete methods will be called respectively. Of course - you don't have to subscribe to all 3 events of an observable, you can subscribe to whichever of them you need.

Scheduler

The Scheduler is RxJS mechanism for correctly timing events. There are several built-in schedulers, such as asyncScheduler, animationFrameScheduler, and more.

Schedulers also allow you to test your Observable code in a deterministic manner, since you can provide a TestScheduler with you can "control time" and make sure your observables work as expected.

the-table.md

	Single	Multiple
Sync	T	T[]
Async	Promise<T>	Observable<T>