We often deal with successive sequences of events. Think of mouse position movement; it produces a stream of cooperating events that we may need to handle. Imagine implementing a drag-drop for example.<\/p>\n
Reactive Stream<\/h3>\n
A reactive stream elevates the observer pattern to a uniform framework. It\u2019s been called \u201cThe observer pattern done right.\u201d It provides a pushed stream of items that we can transform, merge with other streams, or reduce to a single value. Functional programming techniques are in-play. Among them are reliance on first-class functions (passed as data), high-order functions (parameter functions; return value function), and lambdas (anonymous functions).<\/p>\n
Event stream processing resembles the iterator pattern familiar in data streams. Events are divas, however. They occur unpredictably. Thus, event streams use a push model. Iterators use a pull model. Three nutshell concepts are:<\/p>\n
- \n
- Reactive (adjective \u2013 medieval Latin): responds to a stimulus in a particular manner.<\/li>\n
- Stream (noun \u2013 middle English): a steady succession (as of items or events).<\/li>\n
- Push versus pull \u2013 event streams versus data streams.<\/li>\n<\/ul>\n
The following diagram shows a reactive stream pattern. An
Observable<\/code> wraps an item that is an event source. AnObservable.subscribe<\/code> triggers an intermittent stream of discrete items. It is the stream\u2019s listener. AnObservable<\/code> can wrap almost any kind of data, including promises, or nothingness.<\/p>\nA set of well-known operators can tap into its output stream. Those are higher-order functions that often take a parameter lambda, object, or a primitive. Each operator takes a value from the stream, returning a value onto the stream. Together, they resemble a monadic chain process. An operator\u2019s parameter function could internally start a new stream or combine two streams.<\/p>\n
![\"Reactive1\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive1-1024x502.png\")
<\/a><\/p>\nReactiveX<\/h3>\n
It\u2019s difficult to roll your own reactive stream framework. There are several packages available across platforms and languages. I focused on ReactiveX. See http:\/\/reactivex.io\/intro.html<\/a>. The introduction explains it nicely. It\u2019s an umbrella project that has a JavaScript binding named \u201cReactive Extensions,\u201d or RxJs.<\/p>\n
I used the RxJs framework in examples that follow. Refer to https:\/\/github.com\/Reactive-Extensions\/RxJS\/blob\/master\/readme.md<\/a>.
\nA side-benefit of ReactiveX technology is that it cleans up event listeners. Programmers that use traditional event listeners must be ever-mindful of disengaging unneeded listeners, or these will be memory leaks.<\/p>\nThe reactive mode of programming could be difficult to ease into. The next section of this article shows an interactive reference tool that help to visualize various operator functions. Afterward, I\u2019ll show six JsFiddle snippets that cover disparate use cases.<\/p>\n
Marble Diagrams<\/h3>\n
Applying operators for transforming, filtering, and merging streams is a distinct way of reasoning about an event solution. For reference, try this API reference: https:\/\/github.com\/Reactive-Extensions\/RxJS\/blob\/master\/doc\/api\/core\/observable.md<\/a>.<\/p>\n
Some of us can learn in a visually interactive manner. Look at the interactive visualization tool at http:\/\/rxmarbles.com\/<\/a>.<\/p>\n
Want to see what really happens when you operate on streams? For example, a marble diagram of the merge operator allows you to drag event marbles of two input streams to instantly see the output stream. Read on \u2026<\/p>\n
Combining<\/h3>\n
Here\u2019s the
merge<\/code> combining operator. The two topmost input streams apply merge operator to produce the bottom output stream.<\/p>\n![\"Reactive2\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive2-1024x481.png\")
<\/a><\/p>\nBelow, I\u2019ve repositioned the \u201c40\u201d and the \u201c1\u201d event marbles of the second input stream.<\/p>\n
See the stable ordering of the bottom output stream afterward? Apparently, Egon, of Ghostbusters, never tried crossing reactive streams.<\/p>\n
![\"Reactive3\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive3-1024x488.png\")
<\/a><\/p>\nTransforming<\/h3>\n
The well-known transformational
map<\/code>\u00a0operator occurs in other functional libraries that process streams, collections, or plain old arrays. Recalleach<\/code> orforEach<\/code> from lodash and underscore.<\/p>\nAdditionally, there is the JS Array prototype\u2019s
forEach<\/code>, or the Immutable.jsmap<\/code> function. All, including our RxJs map operator, pass a collection or stream to a caller-supplied function, creating a new output sequence.<\/p>\nHere\u2019s the map marble diagram. For each member of the input, its function parameter returns ten times its numeric input parameter. In the end, it transforms the input stream, producing a new stream consisting of each input multiplied by ten. It preserves ordering.<\/p>\n
![\"Reactive4\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive4-1024x362.png\")
<\/a><\/p>\nBelow, I moved the \u201c2\u201d event marble to the extreme left to show that ordering remains.<\/p>\n
![\"Reactive5\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive5-1024x364.png\")
<\/a><\/p>\nFiltering<\/h3>\n
We\u2019ll see a
takeUntil<\/code> filtering operator in a later drag-drop example. Look at that operator\u2019s marble diagram. The second input stream conditionally filters the first stream\u2019s members into the output stream. The second JS stream\u2019s earliest false \u201c0\u201d item halts output. Note the vertical bar on the bottom output that shows the end of the output stream.<\/p>\n![\"Reactive6\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive6-1024x480.png\")
<\/a><\/p>\nBelow, I\u2019ve moved the rightmost middle stream \u201c0\u201d item leftward to just after the \u201c2\u201d item of the upper stream. The
takeUntil<\/code> stops replicating the upper stream there, now producing a result stream containing only \u201c1\u201d and \u201c2\u201d. Again RxJs preserved the ordering.<\/p>\n![\"Reactive7\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive7-1024x478.png\")
<\/a><\/p>\nJSFiddle Examples<\/h3>\n
Each example depends only upon the RxJs library and JQuery. I dropped back to ES5 syntax in the online JSFiddle code because Safari doesn\u2019t support arrow functions. The following examples do use ES6 syntax. See https:\/\/babeljs.io\/docs\/learn-es2015\/<\/a> for succinct examples of ES6 and ES2015.<\/p>\n
Example 1: A Stream of Nothing<\/h2>\n
Recall that an
Observable<\/code> represents a push-based collection. In RxJS, an Observable can front any number of legal JavaScript entity or entities. Let\u2019s lean on the boundary. How about an empty Observable?<\/p>\nYes, but what use is a stream of nothing? Well, it\u2019s an edge case. Consider a dynamic application that might need to handle a source of any number of indeterminate dynamic items, including an undefined JS value. We can handle that.<\/p>\n
![\"Reactive8\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive8.png\")
<\/a><\/p>\n![\"Reactive9\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive9.png\")
<\/a><\/p>\nRefer to JSFiddle https:\/\/jsfiddle.net\/mauget\/3u4po86u\/<\/a>. It produces an
Observable<\/code> by mumbling anRx.Observable.empty()<\/code> incantation. Nothing happens until the Observable suffers asubscribe<\/code> function call. An RxJs subscribe can take three optional functional parameters:<\/p>\n- \n
onNext<\/code><\/li>\nonError<\/code><\/li>\nonCompletion<\/code><\/li>\n<\/ul>\nThe
onError<\/code> andonCompletion<\/code> calls are mutually exclusive. The framework invokes one or the other, not both. For effect I used awindow.setTimeout<\/code> to delay invoking the subscribe until three seconds pass. When the timer pops, I call subscribe. The stream immediately awakens to push the \u201cempty\u201d item. TheonCompletion<\/code> logs a message and updates the DOM with the message.<\/p>\nconst\r\n delay = 3, \/\/ seconds\r\n source = Rx.Observable.empty(), \/\/ Dormant until subscription occurs\r\n msg1 = `Will subscribe to empty source in ${delay} seconds ...\r\n`; \r\n\r\n$('body').append(msg1); \/\/ HTML output\r\nconsole.log(msg1); \/\/ deb log output -\r\n\r\nwindow.setTimeout( () => {\r\n const msg2 = `Subscribing ... here comes the data ...`;\r\n $('body').append(`\r\n${msg2}`);\r\n console.log(msg2);\r\n\r\n \/\/ Kick the stream\r\n source.subscribe(\r\n v => console.log(`onNext: ${v}`), \/\/ arg1 fn\r\n v => console.log(`onError: ${v}`), \/\/ arg2 fn\r\n v => { \/\/ arg3 fn\r\n const msg3 = `yes ... received the expected \"${v}\" from the stream.`;\r\n $('body').append(`\r\n\t\t${msg3}`);\r\n console.log(msg3);\r\n }\r\n );\r\n}, delay * 1000 );<\/pre>\nExample 2: Time Interval<\/h2>\n
Example 1 had a standard
window.setTimer<\/code> that provided a timed interval as a demo prop.<\/p>\nYou object: \u201cIsn\u2019t a timer expiration an event? I thought RxJs is about observing events!\u201d Yup. In that hello-world kind of snippet, I didn\u2019t want to confuse a trivial stream example by involving a second stream.<\/p>\n
Now you\u2019re ready, Grasshopper! Let\u2019s observe timer expirations via RxJs. An
Observable<\/code> stream of timer expirations will generate a dynamically increasing trip counter of values. Refer to the JSFiddle at https:\/\/jsfiddle.net\/mauget\/t03fcr8y\/<\/a>. It generated the before and after panels here.<\/p>\n![\"Reactive91\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive91.png\")
<\/a><\/p>\n![\"Reactive92\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive92.png\")
<\/a><\/p>\nLook at the code below. Notice the RxJs
interval<\/code>,skip<\/code>, andtake<\/code> operators that sequentially process the stream input items into an output stream? Those govern the pace, content, and extent of the product stream.<\/p>\nThe interval operator is the heart of the code. See https:\/\/github.com\/Reactive-Extensions\/RxJS\/blob\/master\/doc\/api\/core\/operators\/interval.md<\/a>. This is a reworked example from RxJs samples. At each delayed invocation the interval call injects a numeric value from a monotonically-increasing zero-origin set.<\/p>\n
The
skip(0)<\/code> operator omits the zeroth item. Thetake(10)<\/code> limits processing to ten output members.<\/p>\nThe
subscribe<\/code> call starts the stream. ItsonNext<\/code> functional parameter calculates a current odometer value from each emitted stream item. TheonError<\/code> andonCompletion<\/code> functional parameters log their respective calls.<\/p>\n\/\/ distance = rate * time\r\nconst R = 60; \/\/ mph \r\nlet t = 0; \/\/ We'll step up at intervals\r\n\r\nconst source = Rx.Observable\r\n .interval(500) \/\/ 1\/2 second interval\r\n .skip(1) \/\/ Skip zeroth\r\n .take(10); \/\/ Take next 10\r\n\r\n\/\/ Kick off ...\r\nconst subscription = source.subscribe(\r\n t => {\r\n console.log('hours: ' + t);\r\n $('#mph').html(R);\r\n $('#hours').html(t);\r\n $('#miles').html(R * t); \/\/ Do the math, display it\r\n },\r\n err => console.log('Error: ' + err),\r\n () => console.log('Completed'));<\/pre>\nExample 3: Interactive Character \/ Word Counter<\/h2>\n
Let\u2019s maintain a dynamic count of words and non-whitespace characters. They\u2019re entered or pasted live into an HTML textarea element. Backspace or delete keys decrement the counts. I count whatever is held in the textarea at each event instant.<\/p>\n
![\"Reactive93\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive93.png\")
<\/a><\/p>\nRefer to https:\/\/jsfiddle.net\/mauget\/o6rprdhx\/<\/a>. It\u2019s another reworked RxJs sample. It defines an
Observable<\/code> of a DOM textarea keyup event. TheObservable<\/code> fronts a push-based collection consisting of a single object that holds the character count and word count of the textarea. The snippet generates the contents from a jQuerymap<\/code> function chained to an event selector. The map callback uses regular expressions to count current non-whitespace characters and words held within the textarea.<\/p>\nNothing happens until the
subscribe<\/code> call on the Observable. Thesubscribe<\/code>\u2019sonNext<\/code> function parameter populates the result, much like a traditional event callback. I omittedonError<\/code> andonCompletion<\/code> callback functions. A pushed Observable-to-subscribe event pulse occurs at each keyup. TheonNext<\/code> carries out an instant update of the two counts in the UI and console log.<\/p>\n\/\/ #A. DOM elements-of-interest from DOM query selectors\r\nconst\r\n $txtInput = $('#txtInput'),\r\n $charCount = $('#charCount'),\r\n $wordCount = $('#wordCount');\r\n\r\n\/\/ #B. Observable of keyup triggered at textInput -- it's a monad\r\nRx.Observable.fromEvent($txtInput.get(0), 'keyup')\r\n .map(e => {\r\n return {\r\n cc: (e.target.value.match(\/[^\\s]\/g) || []).length,\r\n wc: (e.target.value.match(\/\\b\\S+\\b\/g) || []).length\r\n }\r\n })\r\n \/\/ #C. Kick off the stream\r\n .subscribe(v => {\r\n console.log(v);\r\n $charCount.html(` Chars: ${v.cc}`);\r\n $wordCount.html(` Words: ${v.wc}`);\r\n });<\/pre>\nExample 4: Drag and Drop<\/h2>\n
Here is a simple drag-drop example. It uses cooperating Observables on the events: mousedown, mousemove, and mouseup. These collectively implement dragging and dropping a DOM item anywhere within a panel.<\/p>\n
![\"Reactive94\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive94-1024x234.png\")
<\/a><\/p>\n![\"Reactive95\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive95-1024x249.png\")
<\/a><\/p>\nRefer to the JsFiddle at https:\/\/jsfiddle.net\/mauget\/9bhp0qu8\/<\/a>. Try dragging the fruit. This snippet, too, grew out of an RxJs example. It creates three kinds of mouse action observables. The code enables a user to reposition a draggable fruit image on a panel.<\/p>\n
The subscribe to the mousedown Observable fruit starts the action. The event item on the fruit chains to a
flatMap<\/code> operator chained to atakeUntil<\/code> mouseup operator. AflatMap<\/code> is a common function encountered in functional programming. It\u2019s amap<\/code> function that flattens its input sequences into a single output sequence.<\/p>\n(function(dragable) {\r\n\r\n \/\/ Observe three major kinds of mouse events:\r\n \/\/ mousedown, mouseup on the dragable; mousemove over the document.\r\n const\r\n mousedown = Rx.Observable.fromEvent(dragable, 'mousedown'),\r\n mousemove = Rx.Observable.fromEvent(document, 'mousemove'),\r\n mouseup = Rx.Observable.fromEvent(dragable, 'mouseup');\r\n\r\n \/\/ A. Detect mouse down; \r\n \/\/ B. While down, listen for mouse move and mouse up;\r\n \/\/ C. When mouse up: stop listening for moves and mouse up.\r\n const mousedrag = mousedown.flatMap( md => {\r\n\r\n \/\/ Capture mouse down offset position\r\n const\r\n startX = md.offsetX,\r\n startY = md.offsetY;\r\n\r\n \/\/ B1. Track pos differentials using mousemove, until mouseup\r\n return mousemove.map( mm => {\r\n mm.preventDefault();\r\n\r\n return {\r\n left: mm.clientX - startX,\r\n top: mm.clientY - startY\r\n };\r\n \/\/ C. stop the drag when mousup\r\n }).takeUntil(mouseup);\r\n })\r\n\r\n \/\/ Subscribe to mousemove's mousedrag stream:\r\n \/\/ B2. Update draggable's position from mousedrag stream of events\r\n mousedrag.subscribe( pos => {\r\n dragable.style.top = pos.top + 'px';\r\n dragable.style.left = pos.left + 'px';\r\n });\r\n\r\n})($('#dragable').get(0));<\/pre>\nExample 5: AJAX Promise \u2013 A Set of Quantum-Random Numbers<\/h2>\n
Let\u2019s display a set of profoundly random numbers requested from a REST service.<\/p>\n
![\"Reactive96\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive96-1024x626.png\")
<\/a><\/p>\nRefer to JSFiddle https:\/\/jsfiddle.net\/mauget\/b0jfmge6\/<\/a>. In this example, we access the random number REST service via AJAX. We can make an
Observable<\/code> for almost anything. We\u2019ll observe a promise. An AJAX call can return a promise to provide asynchronous behavior that we need to drive a UI from a latent network response. The stream will push the promise.<\/p>\nTo review, a promise is either pending or settled. If it settles to a
success<\/code> state, it forever returns the data it contains from that success. Thus you can callthen<\/code> on that promise whenever you desire, as often as you desire, but it will always return its settled success or reject value for its lifespan.<\/p>\nOur example observes a button click. The click
Observable<\/code>,onNext<\/code>, displays aworking<\/code> banner. A chainedflatMap<\/code> operator callsgetNumbers<\/code> to create a secondObservable<\/code> from a promise. That promise is an immediate result of an AJAX call to the REST service. This is easier to understand by looking at the snippet.<\/p>\nThe
flatMap<\/code> operator gates the settledthen<\/code> object payload of the promiseObservable<\/code> onto the button click stream as a flat sequence of numbers. A subscribe on the button click stream displays the random numbers.<\/p>\nI kick off activity by triggering a refresh button click at the bottom of the code. Notice what happens if you click the button repeatedly? You see the identical random number sequence repainted!<\/p>\n
I claimed the sequence is random, so why does it repeat? Answer: because the promise stream already resolved that settled one-shot AJAX promise. There is no second network invocation.<\/p>\n
const $button = $('#button'),\r\n $result = $('#result');\r\n\r\n\/\/ AJAX fn: returns observable promise\r\nconst getNumbers = count => {\r\n return Rx.Observable.fromPromise(\r\n \/\/ Get promise of: {\"type\":\"uint8\",\"length\":1,\"data\":[99],\"success\":true}\r\n $.ajax({\r\n url: `https:\/\/qrng.anu.edu.au\/API\/jsonI.php?length=${count}&type=uint8`\r\n })\r\n .promise());\r\n};\r\n\r\n\/\/ Observe button click, subscribe to result\r\nRx.Observable.fromEvent($button.get(0), 'click')\r\n .map(e => {\r\n $result.html('<span class=\"working\">(working)<\/span>');\r\n return e.target.value;\r\n }).flatMap(getNumbers(300))\r\n .subscribe(d => {\r\n let sep = '';\r\n console.log(d);\r\n if (d.success) {\r\n $result.empty();\r\n d.data.forEach(v => {\r\n $result.append(sep + v);\r\n sep = ', ';\r\n });\r\n }\r\n });\r\n\r\n$button.trigger('click');<\/pre>\nSidebar<\/h2>\n
As a physics major diverted to computer programming, I\u2019m compelled to speak about our data.<\/p>\n
The numbers originate from a vacuum chamber in Canberra, Australia. A vacuum that emits random numbers! The count of particles and photons is zero in a perfect dark classical vacuum. At the quantum level, however, a vacuum is weird. A lab at the Australian National University measures field variations from continually materializing and dematerializing virtual particle pairs. See https:\/\/en.wikipedia.org\/wiki\/Virtual_particle<\/a>.<\/p>\n
A REST service streams those values. Statistical tests show that the sequence is profoundly random.<\/p>\n
Example 6: AJAX Promise \u2013 A Quilt of Quantum-Random Colors<\/h2>\n
My wife is a quilter. I couldn\u2019t resist converting example 5 output into a random patch quilt.<\/p>\n
![\"Reactive97\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/07\/Reactive97-1024x517.png\")
<\/a><\/p>\nRefer to JSFiddle https:\/\/jsfiddle.net\/mauget\/9c2ecLqs\/<\/a>. The code is largely that of example 5, aside from presentation logic in the
subscribe<\/code> call. The subscribe callback forces each sequence item into a number of the interval 0 \u2026 255. It combines each group of three into one CSS RGB value. Finally, it emits each RGB value into a grid. Again, notice that the refresh button causes an identical quilt to render after the first rendering. That settled promise remains immutable and happy.<\/p>\nconst valCount = 1020;\r\nconst $button = $('#button'),\r\n $result = $('#result');\r\n \r\n\/\/ AJAX fn: returns observable promise\r\nconst getNumbers = count => {\r\n return Rx.Observable.fromPromise(\r\n \/\/ Get promise of: {\"type\":\"uint8\",\"length\":1,\"data\":[99],\"success\":true}\r\n $.ajax({\r\n url: `https:\/\/qrng.anu.edu.au\/API\/jsonI.php?length=${count}&type=uint8`})\r\n \t.promise());\r\n};\r\n\r\n\/\/ Listen for button clicks, do service request, subscribe to result\r\nRx.Observable.fromEvent($button.get(0), 'click')\r\n .map(e => {\r\n $result.html('<span class=\"working\">(working)<\/span>');\r\n return e.target.value;\r\n})\r\n .flatMap( getNumbers( valCount ) )\r\n .subscribe(data => {\r\n \tconsole.log(data);\r\n \tif (data.success) {\r\n \t $result.empty();\r\n \tlet cols = 0\r\n \tconst c = []; \/\/ rgb value\r\n\r\n \t data.data.forEach( v => {\r\n c.push( v % 256 );\r\n if (c.length > 2 ) {\r\n const rgb = `rgb(${c.pop()},${c.pop()},${c.pop()})`;\r\n $result.append(\r\n`<span style=\"background-color:${rgb}\">\u00a0\u00a0\u00a0\u00a0<\/span>`);\r\n\r\n if ( ++cols > 33 ) {\r\n $result.append(`\r\n`);\r\n cols = 0;\r\n }\r\n }\r\n });\r\n }\r\n});\r\n\r\n$button.trigger('click');<\/pre>\nWrap-Up<\/h2>\n
Reactive streams are a unified way of dealing with asynchronous events. They\u2019re a higher-level implementation of the observer pattern. They rely on functional programming. An
Observable<\/code> pushes events onto a stream. Operators modify it. A subscribe operator starts, handles, and ends stream processing.<\/p>\nThis paradigm may require a shift of your thinking for programming in an asynchronous environment. Marble diagrams help to understand many functional operators used in ReactiveX.<\/p>\n
We saw six operational RxJs snippets that support varying use cases. You could run and modify them on JSFiddle for familiarization. Reactive streaming libraries exist across platforms and operating systems. ReactiveX is fairly unified across the board, so concepts from this article could apply beyond JavaScript.<\/p>\n
\n