Functional Reactive Programming:Working with RxJS

Oswald Campesato

Consultant/Training:www.iquarkt.com

ocampesato@yahoo.com

What is Functional Reactive Programming (FRP)?• 1) Functional programming:• is more declarative• often has more abstraction• can involve higher order functions

• 2) Reactive Programming was introduced in 1997:“programming with asynchronous data streams”• Multiple toolkits/libraries available• Supported languages JS, Java, Scala, Android, Swift, Go, .... NB: Elm is an FRP language: http://elm-lang.org/

What is Functional Reactive Programming (FRP)?A) Functional Reactive Programming:• a programming paradigm that was created by Conal Elliott• his definition has very specific semantics:• https://stackoverflow.com/questions/1028250/what-is-func

tional-reactive-programming

B) looser definition of FRP: a combination of 2 other concepts:• Reactive Programming: focuses on asynchronous data

streams, which you can listen to and react accordingly • Functional Programming: emphasizes calculations via

mathematical-style functions, immutability and expressiveness, and minimizes the use of variables and state

Popular Toolkits/Libraries for FRP• RxJS:https://github.com/Reactive-Extensions/RxJS

• Bacon.js:https://baconjs.github.io/

• Kefir.js:https://rpominov.github.io/kefir/

• most.js:https://github.com/cujojs/most

Promises versus RxJS• "Promises are good for solving asynchronous

operations such as querying a service with an XMLHttpRequest, where the expected behavior is one value and then completion.”

• "RxJS unifies both the world of Promises, callbacks as well as evented data such as DOM Input, Web Workers, Web Sockets.”

• Support for Web Sockets in RxJS version 5(?)

How You can Use Observables in RxJS• Orchestrate asynchronous data streams

• handle streams of data of indeterminate length

• Respond to mouse events and button clicks

• Generate SVG graphics/animation

• Combine with Promises (Rx.Observable.fromPromise())

• integrate with Angular 2, jQuery, …

A Vague Description of some Parts of FRP• Obtain streams of data from many sources:an array, list, function, website, ...

• define the required operations via operators

• operators include: map() and filter()

• method chaining of operators is supported

• invoke subscribe() to “make it happen”

Using ‘map’ and ‘filter’ (WITHOUT FRP)• var source = [0,1,2,3,4,5,6,7,8,9,10];

• var result1 = source.map(x => x*x)• .filter(x => x % 5 == 0); • console.log("result1: "+result1);• // output=?

• var result2 = source.filter(x => x % 5 == 0)• .map(x => x*x)• // output=?

• Q: what is the difference between these two?

Core JavaScript Files for RxJS (version 4)<script src="http://cdnjs.cloudflare.com/ajax/libs/rxjs/4.1.0/rx.js"></script><script src="http://cdnjs.cloudflare.com/ajax/libs/rxjs/4.1.0/rx.async.js"></script><script src="http://cdnjs.cloudflare.com/ajax/libs/rxjs/4.1.0/rx.binding.js"></script>

NB: RxJS v5 is currently in beta

JavaScript Files for some Operators in RxJS• <script

src="http://cdnjs.cloudflare.com/ajax/libs/rxjs/4.1.0/rx.time.js">

• </script>• <script

src="http://cdnjs.cloudflare.com/ajax/libs/rxjs/4.1.0/rx.coincidence.js">

• </script>• <script src="http://cdnjs.cloudflare.com/ajax/libs/rxjs-

dom/2.0.7/rx.dom.js">• </script>

What are Observables?• Think of them as “streams” or sets

• Can comprise any number of items (arbitrary time)

• They generate values when they are “subscribed”

• Can be cancelled and restarted

• their purpose is to avoid “callback hell”

• “unsubscribe” will “tear down” a producer

How do Observables Work?• Let x = Rx.Observable.(...)

• Let result = x.subscribe(valueFn, errorFn, CompleteFn)

• Do various things here...

• result.unsubscribe();

Observable Creation in RxJSObservable.of(...)

Observable.from(promise/iterable/observable);

Observable.fromEvent(...)

Observables from HTTP in Angular 2

Additional RxJS modules/libraries

Operators in Observables+ Operators are methods in Observables+ Operators allow you to compose new observables+ Create custom operators based on RxJS operators:Rx.Observable.prototype.myGreatOperator = ....

General syntax of Observable operators:let obs = Rx.Observable .firstOperator() .secondOperator() .evenMoreOperatorsIfYouWant() .subscribe(....); // now stuff happens=

Result:obs is an Observable “connected” to a source

Using Observable, ‘range’, and ‘filter’ in RxJS • var source = Rx.Observable• .range(0, 20) • .filter(x => x < 4) • //output=?

• var source = Rx.Observable• .range(0, 20) • .filter(x => x < 4) • .subscribe(x => console.log("x = "+x))• //output=?// ObservableRangeFilter1.html

Using ‘from’ and ‘map’ in RxJS (2a)• Rx.Observable• .from(['a1','a2','a3'])• .map((item) => { • item = item.toUpperCase()+item;• return item; • }) • .subscribe(str => console.log("item: "+str));• // output:A1a1A2a2A3a3// ObservableMapUpper1.html

Using ‘from’ and ‘map’ in RxJS (2b)var x = Rx.Observable • .from(['a1','a2','a3'])• .map((item) => { • item = item.toUpperCase()+item;• return item; • }) x.subscribe(str => console.log("item: "+str));// output:A1a1A2a2A3a3

Observables: Exercises #1• Modify ObservableMapUpper1.html in the exercises

1) Display the array elements in reverse order

2) Prepend the terminal digit and generate this output:1a1, 2a2, 3a3

3) Concatenate the elements of the input array

Using ‘interval’, ‘take’, and ‘map’ in RxJS (3a)// ObservableTake.html

var source = Rx.Observable .interval(1000) .take(4) .map(i => ['1','2','3','4','5'][i]);

var result = source.subscribe(x => console.log("x = "+x));

// output =?

Using ‘interval’, ‘take’, and ‘map’ in RxJS (3b)var source2 = Rx.Observable .interval(1000) .take(4) .map(i => ['1','2','3','4','5'][i]);

var subscription = source2.subscribe( x => console.log('source2 onNext: %s', x), e => console.log('source2 onError: %s', e), () => console.log('source2 onCompleted'));

• // output =?

Using ‘interval’, ‘take’, and ‘map’ in RxJS (3c)• Output from BOTH observables is interleaved:• x = 1 • source2 onNext: 1• x = 2 • source2 onNext: 2• x = 3 • source2 onNext: 3• x = 4 • source2 onNext: 4• source2 onCompleted

Observables: Exercises #2• Modify ObservableTake.html in the exercises

1) Change the second 1000 to 500 and predict the outcome

2) Emit only even numbers from the first Observable

3) Compute squares of odd numbers in the second Observable

Modify HTML Content via Observables (1)

// ObservableDivElement1.html

<div id="div1">This is a DIV element</div><script>• let div1 = document.querySelector('#div1')

• var stream = Rx.Observable• .interval(500)• .take(10)• .map(x => x*x)• .subscribe(x => div1.innerHTML += x); </script>

Modify HTML Content via Observables (2)

// ObservableDivElement2.html

<div id="div1">This is a DIV element</div><div id="div2">This is a DIV element</div><script> let div1 = document.querySelector('#div1') let div2 = document.querySelector('#div2')

var stream = Rx.Observable .interval(500) .take(10) .map(x => x*x) .subscribe(x => { div1.innerHTML += x; div2.innerHTML += x; })</script>

Observables and SVG Graphics/Animation

1) Observables and SVG graphics:// SVGObservables1.html

2) Observables and SVG animation:// SVGObservables1Anim1.html

3) Observables and SVG “follow the mouse”:// SVGObservables1MouseMove1.html

4) Rxjs and SVG graphics/animation:https://github.com/ocampesato/rxjs-svg-graphics

Some Available Operatorsmap() <= we’ve seen this in Angular 2filter() <= we’ve seen this in Angular 2reduce()first()last()take()skip()toArray()isEmpty()startWith()

Observables: Exercises #3

• Modify: ObservableMapUpper1.html1) Create an Observable with the first() operator

2) Create an Observable with the last() operator

3) What does this Observable emit:var source = Rx.Observable .return(8) .startWith(1, 2, 3) .subscribe(x => console.log("x = "+x));

Merging/Joining Operatorsmerge()mergeMap()concat()concatMap()switch()switchMap()zip()forkJoin() <= requests from multiple sites are mergedwithLatestFrom()combineLatest()

Map-Related Operatorsmap()flatMap()flatMapLatest()mergeMap()concatMap()switchMap()flatten() <= this is different from flatMap*()

Map-Related Operators• map(): items of the observable are mapped or

transformed into something else

• flatMap() (several variants): takes a function returning an observable from each item of the source observable, which produces makes a stream of streams (where stream = observable = sequence of items), and is "flattened" into a single stream / observable by flapMap

• flatMapLatest(): a flatMap where only the items of the current observable are emitted: if a new observable appears, then the values of the previous observable are ignored.

Map-Related Operators• concatMap(): uses concat() so that intermediate

results are not 'interleaved', which can happen with flatMap() because the latter uses merge()

• merge(): combine multiple Observables into one (interleaving can occur)

• concat(): combine multiple Observables sequentially into one (no interleaving occurs)

Observables: Exercises #4 (self-study)

1) Create an Observable with the merge() operator

2) Create an Observable with the zip() operator

3) Create an Observable with the concat() operator

“Hot” versus “Cold” Observables“cold” observables: a new producer for each consumersimilar to watching a recorded movie

“hot” observables:one producer for many consumerssimilar to watching a live streaminvoke “publish” to make an observable “hot”

=> all observables are “cold” by default

From Promises to Observables// define a Promise:var p = new Promise();

// define an Observable from the Promise p:var x = Observable.fromPromise(p);

// do something with x ...

Error Handling with Observables: catchmyObservable.catch( error => { if(error instanceof MyError) { return Observable.of(“MyError”); } else { throw error;});

myObservable.finally(() => { console.log(“done”);});

Retrying ObservablesmyObservable.retry(3);

myObservable.retryWhen(errors => errors.delay(3000));

Note 1: unsubscribe enables the Observer to instruct the Observable to ‘tear down’

Note 2: completion handler enables the Observable to indicate that it has completed successfully

Creating an Observable in v5let obs = new Observable(observer => { myAsyncMethod((err,value) => { if(err) { observer.error(err); } else { observer.next(value); // older v4: onNext observer.complete(); // older v4: onComplete } });});

NB: ‘on’ prefix has been dropped in v5

Further Samples and Reading1) Create a toggle button with RxJS:https://www.themarketingtechnologist.co/create-a-simple-toggle-button-with-rxjs-using-scan-and-startwith/2) RxJS and mouse events:http://jsfiddle.net/dinkleburg/ay8afp5f

3) http://reactivex.io/learnrx/

4) http://rxmarble.com

5) http://cycle.js.org/basic-examples.html

Recent/Upcoming Books and Training1) HTML5 Canvas and CSS3 Graphics (2013)2) jQuery, CSS3, and HTML5 for Mobile (2013)3) HTML5 Pocket Primer (2013)4) jQuery Pocket Primer (2013)5) HTML5 Mobile Pocket Primer (2014)6) D3 Pocket Primer (2015)7) Python Pocket Primer (2015)8) SVG Pocket Primer (2016)9) CSS3 Pocket Primer (2016)10) Angular 2 Pocket Primer (2016)