Upgrade to Pro — share decks privately, control downloads, hide ads and more …

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

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

Tetsuharu Ohzeki

April 29, 2016
Tweet

More Decks by Tetsuharu Ohzeki

Other Decks in Programming

Transcript

  1. 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.
  2. 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
  3. 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
  4. 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!'); } });
  5. 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
  6. 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.
  7. 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
  8. 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
  9. 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);
  10. 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!'); } });
  11. 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!'));
  12. Rx is just “advanced” observer pattern  Sort interfaces &

    calling convention  Define many middle-relay object as “Operator”
  13. 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
  14. 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
  15. 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));
  16. 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.
  17. 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 }
  18. 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));
  19. 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
  20. 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