Max Gallo, at Codemotion Rome 2019<\/strong>, performed a live coding<\/span> session, where he showed us how we can reverse engineer the RxJS library<\/strong>, the Reactive Extensions for JavaScript<\/span> language<\/strong> which allows you to write reactive<\/span> programs<\/span> with a functional approach.<\/p>\n Max Gallo is an engineer working at DAZN, and he is experienced as a Full Stack Developer<\/span>, iOS Developer<\/span> and UI<\/span>\/UX Designer.<\/p>\n RxJS<\/strong> is a library with a clear API<\/span><\/strong> to work with both asynchronous<\/span><\/strong> and synchronous<\/strong> code, thanks to pipeline operators and using the concept of Observable.<\/p>\n The observer pattern is a software design<\/span> pattern in which an object, called the subject, maintains a list of its dependents, called observers, and automatically notifies them of any state changes (usually by calling one of their methods). <\/em>(Wikipedia<\/span>)<\/p>\n<\/blockquote>\n What is the goal of this topic? A lot of people are afraid of Observable and RxJS library, so Max\u2019s intention was to show us how we can build something like RxJS, of course, not all of the standard library, just some line of working demo code, and he did with a VI editor. Cool! We can now create custom operators.<\/p>\n An operator is simple a function<\/span>, they are the horse-power behind observables, providing an elegant, declarative and common API<\/span> to complex tasks in both async\/sync world.<\/em><\/p>\n<\/blockquote>\n If you\u2019ve used RxJS before and want to better understand some internal aspects of the library and the magic behind Observables, as well as the operators, then this post is for you.<\/p>\n Firstly, Max began with showing us some RxJS code, how to create an Observable from an Array with some operators within the pipeline operator, introduced with the latest RxJS 6 release (it was actually already added in\u00a0RxJS\u00a05.5), in substitution of dot-chaining operators. Then, a step forward, he rewrote a couple of the most common function<\/span> operators: from and map, fully compatible with RxJS. So we can now write our custom operators! Wow!<\/p>\n \u00a0<\/p>\n Let’s explore this. The purpose of the library is to offer a wide range of operators to transform, manipulate and create a stream of data from a source such as an event from UI<\/span>, from a XHR response or from some sort of data (an array, a collection or whatever you want to stream and observe).<\/p>\n A stream is a collection of future values and it\u2019s called Subject or Observable; it can emit three different kinds of event with a value (object, array, primitive value), an error or a completed event when the full cycle of values are finished.<\/p>\n Observable is an Object with\u00a0a subscribe\u00a0method<\/span> on it. This\u00a0subscribe\u00a0method<\/span> takes the\u00a0observer\u00a0as an argument and is used to invoke the execution of the Observable. In the RxJS library, the subscribe method<\/span> returns a reference to an unsubscribe, which allows us to spot listen to the values emitted by Observable.<\/p>\n Observables can either be asynchronous<\/span> or synchronous, depending on the underlying implementation. An example of an asynchronous<\/span> observable is one that wraps an XHR call or a timer function<\/span>, while a stream of values comes from an array can be synchronous.<\/p>\n However, we capture these emitted events only asynchronously, with an Observer, by defining three parameters:<\/p>\n Observer is a collection of callback functions, which reacts to the value arriving via Observable.<\/p>\n You can recap the relationship from Observable, Observer, Subscribe and Subscription with a type definition in TypeScript:<\/p>\n So far, we have read what is an Observable and how to subscribed to them. With RxJS we can program<\/span> in a functional way with asynchronous<\/span> values, we can use Array\u2019s map, filter or any other methods<\/span> and operators to modify the original Observable.<\/p>\n Those operators are chainable and composable and to work with them we have a method<\/span> on the Observable object named\u00a0pipe.<\/strong>\u00a0This pipe method<\/span> takes single or multiple operators as an input and returns a new Observable.<\/p>\n Here is an example of using a\u00a0filter\u00a0<\/strong>and\u00a0map<\/strong>\u00a0operator in RxJS:<\/p>\n The syntax looks like very familiar to the functional world where we have:<\/p>\n Let\u2019s explore RxJS by starting writing own operator. We will start with from<\/strong> operator to generate a stream of values.<\/p>\n To be compliant with RxJS specification, take some initial data and return an object with the pipe function<\/span>. The pipe method<\/span> accepts a set of functions<\/span> as parameters. We are going to use the rest operators from ES6<\/span> because we don\u2019t know how many functions<\/span> we will have. The output of the pipe is an Observable object with a subscribe method<\/span> used for consuming it:<\/p>\n The subscribe method<\/span> has an object called<\/span> dataObservable with a subscribe method<\/span> too, which takes the function<\/span> that will be the function<\/span> we pass to the subscribe method<\/span> to take the emitted value. This innermost subscribe method<\/span> emits the value with a forEach passing the next method<\/span>, in this case a simple console.log function<\/span>. At this time we don\u2019t apply any pipe function<\/span>, we do it early.<\/p>\n Before we move on and made our code more reusable<\/span>, we must define a function<\/span> called createObservable<\/em>:<\/p>\n As we can see from the picture, RxJS can chain more than one operator, where we got an observable source and produce another observable for the next operator. So we have the previous, the current and the next observable. When we have to produce the next value, we want to apply a pipe function<\/span>, so we need to define it inside our from function<\/span>:<\/p>\n Now we can define our custom map and filter:<\/p>\n The filter function<\/span> takes one parameter that returns a source observable, we construct the observable and subscribe to it where we need to check if my filter function<\/span> of value is true and, if that is true, we get to the next destination with this value.<\/p>\n So to recap, I create an observable from my initial data, I pipe all my functions<\/span> one with another, and at the end I subscribe to the final observable.<\/p>\n So far, all that we see is synchronous, we don\u2019t talk about order and speed. How I can control the order and the speed of the emission of my value.<\/p>\n Scheduler<\/span> in RxJS controls the order of event emissions and the speed of those event emissions.<\/p>\n In RxJS we have five kinds of schedulers available, three of them are very useful and common: asyncScheduler, queueScheduler and asapScheduler.<\/p>\n We can use Virtual Scheduler<\/span> to control the speed of our Observable, especially in test ambient:<\/p>\n So, to better understand how something works under the hood, try to disassemble, simplify and challenge it by reinventing the wheel to learn – and please share it with us. Good luck!<\/p>\n\n
\n
![\"\"](\"https:\/\/codemotion.com\/magazine\/wp-content\/uploads\/2019\/12\/MY-IMAGE_1_1c12aad4a022c2475d03af0169909343_800.jpg)
<\/picture>\nRxJS<\/h2>\n
\n
Observable.subscribe(observer:Observer):Subscription;<\/pre>\n
const { from } = require('rxjs');
const { map, filter } = require('rxjs\/operators');
const observable = from([1, 2, 3, 4, 5])
.pipe(
map ( x => x + 1),
filter( x => x % 2 === 0),
map( x => x - 1)
);
observable.subscribe(
val => console.log('odd:', val),
error => console.error(error),
() => console.log('complete')
);<\/pre>\n\n
Reinventing RxJS<\/h2>\n
function<\/span> from(initialData) {
\u00a0 \u00a0 return {
\u00a0 \u00a0 pipe(...pipeFunction) {
return {
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 subscribe(onNext, onError, onComplete) {
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }
\u00a0 \u00a0 \u00a0 \u00a0 }
}
}<\/pre>\nfunction<\/span> createObservable(operator) {
return {
\u00a0 \u00a0 subscribe(next) {
\u00a0 \u00a0 operator(next);
\u00a0 \u00a0 }
}
}<\/pre>\n![\"\"](\"https:\/\/codemotion.com\/magazine\/wp-content\/uploads\/2019\/12\/MY-IMAGE_2_633178b01c96c2d30db9407ec1d16f41_800.jpg)
<\/picture>\nfunction<\/span> from(initialData){
return {
\u00a0 \u00a0 \u00a0 \u00a0 pipe: (...pipeFunctions) => {
return {
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 subscribe: (onNext, onError, onComplete) => {
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 const dataObservable = createObservable(x => initialData.forEach(x))
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 let currentObservable = dataObservable;
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 pipeFunctions.forEach(pipeFunc => {
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 currentObservable = pipeFunc(currentObservable)
})
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 currentObservable.subscribe(onNext);
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 onComplete();
\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }
\u00a0 \u00a0 \u00a0 \u00a0 }
}
}
}<\/pre>\nfunction<\/span> map(mapFunction){
return sourceObservable => {
const currentObservable = createObservable(destinationNext => {
sourceObservable.subscribe(value => {
const newValue = mapFunction(value)
destinationNext(newValue);
})
})
return currentObservable;
}
}
function<\/span> filter(filterFunction){
return sourceObservable => {
const currentObservable = createObservable(destinationNext => {
sourceObservable.subscribe(value => {
if(filterFunction(value)){
destinationNext(value);
}
})
})
return currentObservable;
}
}<\/pre>\nconst { interval } = require('rxjs');
const { take, filter } = require('rxjs\/operators');
const source = interval(1000)
\u00a0 \u00a0 .pipe(take(3600));
source.subscribe(
\u00a0 \u00a0 console.log,
\u00a0 \u00a0 console.err,
() => console.log('--> Completed!'),
);
console.log('--> Subscribed!')<\/pre>\n