Introduction to Rx without saying "reactive" or "functional"

https://webplat.doorkeeper.jp/events/42974

Transcript

1. Introduction to Reactive Extensions without saying “functional” or “reactive” Tetsuharu OHZEKI (April 28, 2016. html5j platform)
2. About me  Tetsuharu OHZEKI  twitter: saneyuki_s  Open Source Developer  Mozilla Committer (Firefox, Servo as reviewer)  ReactiveX/RxJS  And some others… (whatwg html, etc)  Working at VOYAGE GROUP, inc.  My main work is to develop an advertising management system with using TypeScript, babel compiler, React, and RxJS.
3. https://youtu.be/COviCoUtwx4
4. I don’t talk about…  Functional Programming  Functional Reactive Programming (FRP)  Fiber-Reinforced Plastics (FRP)  Enterprise Application Architecture Pattern  Read “Microsoft .NET - Architecting Applications for the Enterprise, 2nd Edition (Dino Esposito)”. It’s very nice.  Recommend to use RxJS
5. It is sometimes called “functional reactive programming” but this is a misnomer. ReactiveX may be functional, and it may be reactive, but “functional reactive programming” is a different animal. One main point of difference is that functional reactive programming operates on values that change continuously over time, while ReactiveX operates on discrete values that are emitted over time. http://reactivex.io/intro.html
6. I’ll talk about… Programming style for application engineering in practice
7. const subject = new SomeEventEmitter(); const flag1 = false; const flag2 = Date.now(); const flag3 = false; subject.on('bar', (e) => { flag1 = !flag1; }); subject.on('foo', (e) => { const now = Date.now(); if (now > (flag2 + 100)) { flag2 = now; flag3 = true; } }); subject.on('hogefuga', (e) => { if (flag1 && flag3) { alert('hello!'); } });
8. Events are everywhere…  Data Mutation  Domain Event  File I/O  Network I/O  Network Status  User Input  Geolocation  Battery  Memory Pressure  Device Orientation  Ambient light  Cluster Node’s status  Server monitoring  Logging
9. Events are queryable collection  If we regard an event stream as a sequenced data structure, we can regard a set of them as a collection.  Collection, we can query it.
11. Rx came from C#…  Originated in “Microsoft Live Labs Volta” (2008)  Designed by Erik Meijer and his team  Reactive Extensions in .NET (2009~)  “LINQ to Events”  RxJava is open sourced by Netflix (2012)  By Jefer Husain https://twitter.com/jhusain https://vimeo.com/110554082
12. Rx key components Observable Operator Subject Scheduler
13. Pull Push One T Promise<T> Zero- Infinity Iterable<T> Observable<T> Observable
14. Operator http://reactivex.io/documentation/operators/map.html
15. Subject http://reactivex.io/documentation/subject.html
16. Scheduler Event Loop Scheduler
17. Programming Key Points  Repeat these until you get an expected final value 1. Define a “source” of an event as Observable 2. Apply Operator, and create a new “check-point” Observable  Scheduler can control “timing” to deliver a value
18. Compose a new operator for you // Filter sequence, then transform from T to U. function filterMap<T, U>(source: Rx.Observable<T>, filterFn: (v: T) => boolean, mapFn: (v: T) => U): Rx.Observable<U>{ return source.filter(filterFn).map(mapFn); } const foo = filterMap(bar);
19. const subject = new SomeEventEmitter(); const flag1 = false; const flag2 = Date.now(); const flag3 = false; subject.on('bar', function(e) { flag1 = !flag1; }); subject.on('foo', function(e) { const now = Date.now(); if (now > (flag2 + 100)) { flag2 = now; flag3 = true; } }); subject.on('hogefuga', function(e) { if (flag1 && flag3) { alert('hello!'); } });
20. const bar = Observable.fromEvent(subject, 'bar') .scan((flag) => !flag, false) .filter((ok) => ok); const foo = Observable.fromEvent(subject, 'foo') .debounce(100); Observable.fromEvent(subject, 'hogefuga') .withLatestFrom(bar, foo, () => alert('hello!'));
21. alert() withLatestFrom filter scan bar debounce (100) foo barfoo
22. See the implementation: scan()
23. See the implementation: scan https://github.com/ReactiveX/rxjs/blob/a3ec896/src/operator/scan.ts
24. alert() WithLatestFromSubscriber FilterSubscriber ScanSubscriber bar DebounceSubscriber foo barfoo
25. Rx is just “advanced” observer pattern  Sort interfaces & calling convention  Define many middle-relay object as “Operator”
26. Pull Push Zero- Infinity Iterable<T> Observable<T>
27. Duality (Iterable vs Observable) Pull (e.g. Iterable) Calling “ancestor” data source from “descendant” consumer. Push (e.g. Observable) Calling “descendant” consumer from “ancestor” data source data source consumer Pull Push
28. e.g. Iterable + operator extensions (Let’s create!) const exiterable = ExIterable.create(iterable); const result = exiterable .filter((v) => v % 2 === 0) .map((v) => v * 2); const {done, value} = result.next(); for (let value of result) { … } // caution: blocking
29. e.g. Iterable -> Observable (Sync) // This push all values as sync const iterableToObservable = Rx.Observable.create((subscriber) => { for (const item of iterable) { // This need be blocking subscriber.next(item); } subscriber.complete(); }); // This cause blocking… iterableToObservable.subscribe((v) => console.log(v));
30. e.g. Observable -> Iterable (Sync) class ObservableToIter { constructor(source) { this._cache = cache; source.subscribe((v) => this._cache.push(v)); } next() { const done = this._cache.length > 0; let value = undefined; if (done) { value = this._cache.shift(); } return { done, value, }; } [Symbol.iterator]() { return this; } } const iter = new ObservableToIter(source); iter.next();  Limitations (Problems)  On consuming first value or initializing the converter, we need to get all value from sequence.  If observable returns value async, we don't transform observable to iterable, or wait to complete observable with blocking!  If observable is infinite sequence, we cannot return to consumer.
31. https://twitter.com/headinthebox/status/669963949027160064
32. e.g. Async Iteration (proposal for ECMA262) const result: Promise<IteratorResult<T>> = asyncIter.next(); result.then((done, value) => console.log(done, value)); for await (let value of asyncIter) { // not blocking // blah, blah, blah }
33. e.g. AsyncIterable -> Observable (PoC) // This can push all values async const asyncIterableToObservable = Rx.Observable.create((subscriber) => { for await (const item of asyncIterable) { // This need not to be blocking subscriber.next(item); } subscriber.complete(); }); asyncIterableToObservable.subscribe((v) => console.log(v));
34. e.g. Observable -> AsyncIterable (PoC) class ObservableToIter { constructor(source) { this._buffer = []; this._resolverCache = []; this._done = false; this._current = 0; source.subscribe((v) => { this._buffer.push({ ok: true, value: v, }); const lastIndex = this._buffer.length - 1; const resolver = this._resolverCache[lastIndex]; if (resolver !== undefined) { resolver({ done: this._done, value: v, }); this._resolverCache[lastIndex] = undefined; } }, () => {}, () => { this._done = true; for (const resolver of this._resolverCache) { if (resolver === undefined) { continue; } resolver({ done: true, }); } this._resolverCache = []; }); } next() { const done = this._done; if (done) { return Promise.resolve({ done }); } const current = this._current; this._current++; const result = this._buffer[current]; if (result !== undefined && result.ok) { return Promise.resolve({ done, value: result.value, }); } else { return new Promise((resolve) => { this._resolverCache[current] = resolve; }); } } [System.asyncIterator]() { return this; } } const source = Rx.Observable.create((o) => { const buffer: any = []; for (const i of [1, 2, 3, 4, 5]) { buffer.push(new Promise((r) => { window.setTimeout(() => { r(); o.next(i); }, i * 1000); })); } Promise.all(buffer).then(() => o.complete()); }); const iter = new ObservableToIter(source); await iter.next() // done: false, value: 1 await iter.next(); // done: false, value: 2 await iter.next(); // done: false, value: 3 await iter.next(); // done: false, value: 4 await iter.next(); // done: false, value: 5 await iter.next(); // done: false, value: undefined
35. Summary  Events are collections, we can query them  Push/Pull is duality of APIs Which is more suitable for your case?  Rx is advanced observer pattern implementation that provides highly composable push APIs