What is RxJS?

Reactive programming is an asynchronous programming paradigm concerned with data streams and the propagation of change - Wikipedia

RxJS is a library for reactive programming using observables that makes it easier to compose asynchronous or callback-based code - RxJS docs

The essential concepts in RxJS are

• An Observable is a stream of data

• Observers can register up to 3 callbacks:

• next is called 1:M time to push new values to the observer

• error is called at most 1 time when an error occurred

• complete is called at most 1 time on completion

• Subscription "kicks off" the observable stream

Without subscribing the stream won't start emitting values. This is what we call a cold observable.

It's similar to subscribing to a newspaper or magazine... you won't start getting them until you subscribe. Then, it creates a 1 to 1 relationship between the producer (observable) and the consumer (observer).

What are RxJS operators?

Operators are pure functions that enable a functional programming style of dealing with collections with operations. There are two kinds of operators:

• Creation operators
• Pipeable operators: transformation, filtering, rate limiting, flattening

Subjects are a special type of Observable that allows values to be multicast to many Observers. While plain Observables are unicast (each subscribed Observer owns an independent execution of the Observable), Subjects are multicast. This is what we call a hot observable.

In this article, I will focus on the zip, combineLatest and forkJoin operators. These are RxJS combination operators, which means that they enable us to join information from multiple observables. Order, time, and structure of emitted values are the primary differences among them.

Let's look at each one individually.

zip()

• zip doesn’t start to emit until each inner observable emits at least one value
• zip emits as long as emitted values can be collected from all inner observables
• zip emits values as an array

Let’s imagine that you are with Mario and Luigi, two of your best friends, at the best Italian restaurant in Rome. Each one of you orders a drink, a pizza, and a dessert. You specify to the waiter to bring the drinks first, then the pizzas, and finally the desserts.

This situation can be represented with 3 different observables, representing the 3 different orders. In this specific situation, the waiter can use the zip operator to bring (emit) the different order items by category.

❗️❗️❗️Warning❗️❗️❗️

If you go back to the same Italian restaurant with your girlfriend, but she doesn’t want to eat, this is what will happen:

If the waiter$ uses the zip operator, you will only get your drink!

Why?

Because, when the waiter$ emits the drinks, the girlfriend$ observable is complete and no more value can be collected from it. Hopefully, the waiter$ can use another operator for us so we don't break up with our girlfriend ?

combineLatest()

• combineLatest doesn’t start to emit until each inner observable emits at least one value
• When any inner observable emits a value, emit the last emitted value from each

At the exact same restaurant, the smart waiter$ now decide to use combineLatest operator.

❗️❗️❗️Warning❗️❗️❗️

With combineLatest, the order of the provided inner observables does matter.

If you$ is provided first to waiter$, it will emit only one value ["Tiramisu", "Sprite"].

This is happening because combineLatest doesn’t start to emit until each inner observable emits at least one value. girlfriend$ starts emitting when the first inner observable emits its last value. Then, combineLatest emits the last values collected from both inner observables.

forkJoin()

• forkJoin emits the last emitted value from each inner observables after they all complete
• forkJoin will never emit if one of the observables doesn’t complete

When you go to the restaurant and order for a pizza, you don’t want to know all the steps about how the pizza is prepared. If the cheese is added before the tomatoes or the opposite. You just want to get your pizza! This is where forkJoin comes into play.

❗️❗️❗️Warning❗️❗️❗️

• If one of the inner observables throws an error, all values are lost
• forkJoin doesn’t complete

• If you are only concerned when all inner observables complete successfully, you can catch the error from the outside
• Then, forkJoin completes

• If you don’t care that inner observables complete successfully or not, you must catch errors from every single inner observable
• Then, forkJoin completes

Personally, when I go to a restaurant with friends, I don’t care if one of them receives a burnt pizza. I just want mine ? so I will ask the waiter$ to catch the errors from inner observables individually.

Wrap up

We covered a lot in this article! Good examples are important to better understand RxJS operators and how to choose them wisely.

For combination operators like zip, combineLatest, and forkJoin the order of inner observables that you provide is also critical, as it can drives you to unexpected behaviours.

There is much more to cover within RxJS and I will do it in further articles.

I hope you enjoyed this article! ?

? You can follow me on Twitter to get notified about new Angular/RxJS blog posts and cool tips!

Redux-Observable is an RxJS-based middleware for Redux that allows developers to work with async actions. It's an alternative to redux-thunk and redux-saga.

This article covers the basics of RxJS, how to setup Redux-Observables, and some of its practical use-cases. But before that, we need to understand the Observer Pattern.

Observer Pattern

In Observer pattern, an object called "Observable" or "Subject", maintains a collection of subscribers called "Observers." When the subjects' state changes, it notifies all its Observers.

In JavaScript, the simplest example would be event emitters and event handlers.

When you do .addEventListener, you are pushing an observer into the subject's collection of observers. Whenever the event happens, the subject notifies all the observers.

Observer Pattern

RxJS

As per the official website,

RxJS is JavaScript implementation of ReactiveX, a library for composing asynchronous and event-based programs by using observable sequences.

In simple terms, RxJS is an implementation of the Observer pattern. It also extends the Observer pattern by providing operators that allow us to compose Observables and Subjects in a declarative manner.

Observers, Observables, Operators, and Subjects are the building blocks of RxJS. So let's look at each one in more detail now.

Observers

Observers are objects that can subscribe to Observables and Subjects. After subscribing, they can receive notifications of three types - next, error, and complete.

Any object with the following structure can be used as an Observer.

interface Observer<T> {
    closed?: boolean;
    next: (value: T) => void;
    error: (err: any) => void;
    complete: () => void;
}

When the Observable pushes next, error, and complete notifications, the Observer's .next, .error, and .complete methods are invoked.

Observables

Observables are objects that can emit data over a period of time. It can be represented using the "marble diagram".

Successfully completed Observable

Where the horizontal line represents the time, the circular nodes represent the data emitted by the Observable, and the vertical line indicates that the Observable has completed successfully.

Observable with an error

Observables may encounter an error. The cross represents the error emitted by the Observable.

The "completed" and "error" states are final. That means, Observables cannot emit any data after completing successfully or encountering an error.

Creating an Observable

Observables are created using the new Observable constructor that takes one argument - the subscribe function. Observables can also be created using some operators, but we will talk about that later when we talk about Operators.

import { Observable } from 'rxjs';

const observable = new Observable(subscriber => {
   // Subscribe function 
});

Subscribing to an Observable

Observables can be subscribed using their .subscribe method and passing an Observer.

observable.subscribe({
    next: (x) => console.log(x),
    error: (x) => console.log(x),
    complete: () => console.log('completed');
});

Execution of an Observable

The subscribe function that we passed to the new Observable constructor is executed every time the Observable is subscribed.

The subscribe function takes one argument - the Subscriber. The Subscriber resembles the structure of an Observer, and it has the same 3 methods: .next, .error, and .complete.

Observables can push data to the Observer using the .next method. If the Observable has completed successfully, it can notify the Observer using the .complete method. If the Observable has encountered an error, it can push the error to the Observer using the .error method.

// Create an Observable
const observable = new Observable(subscriber => {
   subscriber.next('first data');
   subscriber.next('second data');
   setTimeout(() => {
       subscriber.next('after 1 second - last data');
       subscriber.complete();
       subscriber.next('data after completion'); // <-- ignored
   }, 1000);
   subscriber.next('third data');
});

// Subscribe to the Observable
observable.subscribe({
    next: (x) => console.log(x),
    error: (x) => console.log(x),
    complete: () => console.log('completed')
});

// Outputs:
//
// first data
// second data
// third data
// after 1 second - last data
// completed

Observables are Unicast

Observables are unicast, which means Observables can have at most one subscriber. When an Observer subscribes to an Observable, it gets a copy of the Observable that has its own execution path, making the Observables unicast.

It is like watching a YouTube video. All viewers watch the same video content but, they can be at watching different segments of the video.

Example: let us create an Observable that emits 1 to 10 over 10 seconds. Then, subscribe to the Observable once immediately, and again after 5 seconds.

// Create an Observable that emits data every second for 10 seconds
const observable = new Observable(subscriber => {
    let count = 1;
    const interval = setInterval(() => {
        subscriber.next(count++);

        if (count > 10) {
            clearInterval(interval);   
        }
    }, 1000);
});

// Subscribe to the Observable
observable.subscribe({
    next: value => {
        console.log(`Observer 1: ${value}`);
    }
});

// After 5 seconds subscribe again
setTimeout(() => {
    observable.subscribe({
        next: value => {
            console.log(`Observer 2: ${value}`);
        }
    });
}, 5000);

/* Output

Observer 1: 1
Observer 1: 2
Observer 1: 3
Observer 1: 4
Observer 1: 5
Observer 2: 1
Observer 1: 6
Observer 2: 2
Observer 1: 7
Observer 2: 3
Observer 1: 8
Observer 2: 4
Observer 1: 9
Observer 2: 5
Observer 1: 10
Observer 2: 6
Observer 2: 7
Observer 2: 8
Observer 2: 9
Observer 2: 10

*/

In the output, you can notice that the second Observer started printing from 1 even though it subscribed after 5 seconds. This happens because the second Observer received a copy of the Observable whose subscribe function was invoked again. This illustrates the unicast behaviour of Observables.

Subjects

A Subject is a special type of Observable.

Creating a Subject

A Subject is created using the new Subject constructor.

import { Subject } from 'rxjs';

// Create a subject
const subject = new Subject();

Subscribing to a Subject

Subscribing to a Subject is similar to subscribing to an Observable: you use the .subscribe method and pass an Observer.

subject.subscribe({
    next: (x) => console.log(x),
    error: (x) => console.log(x),
    complete: () => console.log("done")
});

Execution of a Subject

Unlike Observables, a Subject calls its own .next, .error, and .complete methods to push data to Observers.

// Push data to all observers
subject.next('first data');

// Push error to all observers
subject.error('oops something went wrong');

// Complete
subject.complete('done');

Subjects are Multicast

Subjects are multicast: multiple Observers share the same Subject and its execution path. It means all notifications are broadcasted to all the Observers. It is like watching a live program. All viewers are watching the same segment of the same content at the same time.

Example: let us create a Subject that emits 1 to 10 over 10 seconds. Then, subscribe to the Observable once immediately, and again after 5 seconds.

// Create a subject
const subject = new Subject();

let count = 1;
const interval = setInterval(() => {
    subscriber.next(count++);
    if (count > 10) {
        clearInterval(interval);
    }
}, 1000);

// Subscribe to the subjects
subject.subscribe(data => {
    console.log(`Observer 1: ${data}`);
});

// After 5 seconds subscribe again
setTimeout(() => {
    subject.subscribe(data => {
        console.log(`Observer 2: ${data}`);
    });
}, 5000);

/* OUTPUT

Observer 1: 1
Observer 1: 2
Observer 1: 3
Observer 1: 4
Observer 1: 5
Observer 2: 5
Observer 1: 6
Observer 2: 6
Observer 1: 7
Observer 2: 7
Observer 1: 8
Observer 2: 8
Observer 1: 9
Observer 2: 9
Observer 1: 10
Observer 2: 10

*/

In the output, you can notice that the second Observer started printing from 5 instead of starting from 1. This happens because the second Observer is sharing the same Subject. Since it subscribed after 5 seconds, the Subject has already finished emitting 1 to 4. This illustrates the multicast behavior of a Subject.

Subjects are both Observable and Observer

Subjects have the .next, .error and .complete methods. That means that they follow the structure of Observers. Hence, a Subject can also be used as an Observer and passed to the .subscribe function of Observables or other Subjects.

Example: let us create an Observable and a Subject. Then subscribe to the Observable using the Subject as an Observer. Finally, subscribe to the Subject. All the values emitted by the Observable will be pushed to the Subject, and the Subject will broadcast the received values to all its Observers.

// Create an Observable that emits data every second
const observable = new Observable(subscriber => {
   let count = 1;
   const interval = setInterval(() => {
       subscriber.next(count++);

       if (count > 5) {
            clearInterval(interval);   
       }
   }, 1000);
});

// Create a subject
const subject = new Subject();

// Use the Subject as Observer and subscribe to the Observable
observable.subscribe(subject);

// Subscribe to the subject
subject.subscribe({
    next: value => console.log(value)
});

/* Output

1
2
3
4
5

*/

Operators

Operators are what make RxJS useful. Operators are pure functions that return a new Observable. They can be categorized into 2 main categories:

1. Creation Operators
2. Pipeable Operators

Creation Operators

Creation Operators are functions that can create a new Observable.

Example: we can create an Observable that emits each element of an array using the from operator.

const observable = from([2, 30, 5, 22, 60, 1]);

observable.subscribe({
    next: (value) => console.log("Received", value),
    error: (err) => console.log(err),
    complete: () => console.log("done")
});

/* OUTPUTS

Received 2
Received 30
Received 5
Received 22
Received 60
Received 1
done

*/

The same can be an Observable using the marble diagram.

Pipeable Operators

Pipeable Operators are functions that take an Observable as an input and return a new Observable with modified behavior.

Example: let's take the Observable that we created using the from operator. Now using this Observable, we can to create a new Observable that emits only numbers greater than 10 using the filter operator.

const greaterThanTen = observable.pipe(filter(x => x > 10));

greaterThanTen.subscribe(console.log, console.log, () => console.log("completed"));

// OUTPUT
// 11
// 12
// 13
// 14
// 15

The same can be represented using the marble diagram.

There are many more useful operators out there. You can see the full operators list along with examples on the official RxJS documentation here.

It is crucial to understand all the commonly used operators. Here are some operators that I use often:

1. mergeMap
2. switchMap
3. exhaustMap
4. map
5. catchError
6. startWith
7. delay
8. debounce
9. throttle
10. interval
11. from
12. of

Redux Observables

As per the official website,

RxJS-based middleware for Redux. Compose and cancel async actions to create side effects and more.

In Redux, whenever an action is dispatched, it runs through all the reducer functions and a new state is returned.

Redux-observable takes all these dispatched actions and new states and creates two observables out of it - Actions observable action$, and States observable state$.

Actions observable will emit all the actions that are dispatched using the store.dispatch(). States observable will emit all the new state objects returned by the root reducer.

Epics

As per the official website,

It is a function which takes a stream of actions and returns a stream of actions. Actions in, actions out.

Epics are functions that can be used to subscribe to Actions and States Observables. Once subscribed, epics will receive the stream of actions and states as input, and it must return a stream of actions as an output. Actions In - Actions Out.

const someEpic = (action$, state$) => { 
    return action$.pipe( // subscribe to actions observable
        map(action => { // Receive every action, Actions In
            return someOtherAction(); // return an action, Actions Out
        })
    )
}

It is important to understand that all the actions received in the Epic have already finished running through the reducers.

Inside an Epic, we can use any RxJS observable patterns, and this is what makes redux-observables useful.

Example: we can use the .filter operator to create a new intermediate observable. Similarly, we can create any number of intermediate observables, but the final output of the final observable must be an action, otherwise an exception will be raised by redux-observable.

const sampleEpic = (action$, state$) => {
    return action$.pipe(
        filter(action => action.payload.age >= 18), // can create intermediate observables and streams
        map(value => above18(value)) // where above18 is an action creator
    );
}

Every action emitted by the Epics are immediately dispatched using the store.dispatch().

Setup

First, let's install the dependencies.

npm install --save rxjs redux-observable

Create a separate folder named epics to keep all the epics. Create a new file index.js inside the epics folder and combine all the epics using the combineEpics function to create the root epic. Then export the root epic.

import { combineEpics } from 'redux-observable';
import { epic1 } from './epic1';
import { epic2 } from './epic2';

const epic1 = (action$, state$) => {
 ...   
}

const epic2 = (action$, state$) => {
 ...   
}

export default combineEpics(epic1, epic2);

Create an epic middleware using the createEpicMiddleware function and pass it to the createStore Redux function.

import { createEpicMiddleware } from 'redux-observable';
import { createStore, applyMiddleware } from 'redux';
import rootEpic from './rootEpics';

const epicMiddleware = createEpicMiddlware();

const store = createStore(
    rootReducer,
    applyMiddleware(epicMiddlware)
);

Finally, pass the root epic to epic middleware's .run method.

epicMiddleware.run(rootEpic);

Some Practical Usecases

RxJS has a big learning curve, and the redux-observable setup worsens the already painful Redux setup process. All that makes Redux observable look like an overkill. But here are some practical use cases that can change your mind.

Throughout this section, I will be comparing redux-observables with redux-thunk to show how redux-observables can be helpful in complex use-cases. I don't hate redux-thunk, I love it, and I use it every day!

1. Make API Calls

Usecase: Make an API call to fetch comments of a post. Show loaders when the API call is in progress and also handle API errors.

A redux-thunk implementation will look like this,

function getComments(postId){
    return (dispatch) => {
        dispatch(getCommentsInProgress());
        axios.get(`/v1/api/posts/${postId}/comments`).then(response => {
            dispatch(getCommentsSuccess(response.data.comments));
        }).catch(() => {
            dispatch(getCommentsFailed());
        });
    }
}

and this is absolutely correct. But the action creator is bloated.

We can write an Epic to implement the same using redux-observables.

const getCommentsEpic = (action$, state$) => action$.pipe(
    ofType('GET_COMMENTS'),
    mergeMap((action) => from(axios.get(`/v1/api/posts/${action.payload.postId}/comments`).pipe(
        map(response => getCommentsSuccess(response.data.comments)),
        catchError(() => getCommentsFailed()),
        startWith(getCommentsInProgress())
    )
);

Now it allows us to have a clean and simple action creator like this,

function getComments(postId) {
    return {
        type: 'GET_COMMENTS',
        payload: {
            postId
        }
    }
}

2. Request Debouncing

Usecase: Provide autocompletion for a text field by calling an API whenever the value of the text field changes. API call should be made 1 second after the user has stopped typing.

A redux-thunk implementation will look like this,

let timeout;

function valueChanged(value) {
    return dispatch => {
        dispatch(loadSuggestionsInProgress());
        dispatch({
            type: 'VALUE_CHANGED',
            payload: {
                value
            }
        });

        // If changed again within 1 second, cancel the timeout
        timeout && clearTimeout(timeout);

        // Make API Call after 1 second
        timeout = setTimeout(() => {
            axios.get(`/suggestions?q=${value}`)
                .then(response =>
                      dispatch(loadSuggestionsSuccess(response.data.suggestions)))
                .catch(() => dispatch(loadSuggestionsFailed()))
        }, 1000, value);
    }
}

It requires a global variable timeout. When we start using global variables, our action creators are not longer pure functions. It also becomes difficult to unit test the action creators that use a global variable.

We can implement the same with redux-observable using the .debounce operator.

const loadSuggestionsEpic = (action$, state$) => action$.pipe(
    ofType('VALUE_CHANGED'),
    debounce(1000),
    mergeMap(action => from(axios.get(`/suggestions?q=${action.payload.value}`)).pipe(
        map(response => loadSuggestionsSuccess(response.data.suggestions)),
        catchError(() => loadSuggestionsFailed())
    )),
    startWith(loadSuggestionsInProgress())
);

Now, our action creators can be cleaned up, and more importantly, they can be pure functions again.

function valueChanged(value) {
    return {
        type: 'VALUE_CHANGED',
        payload: {
            value
        }
    }
}

3. Request Cancellation

Usecase: Continuing the previous use-case, assume that the user didn't type anything for 1 second, and we made our 1st API call to fetch the suggestions.

Let's say the API itself takes an average of 2-3 seconds to return the result. Now, if the user types something while the 1st API call is in progress, after 1 second, we will make our 2nd API. We can end up having two API calls at the same time, and it can create a race condition.

To avoid this, we need to cancel the 1st API call before making the 2nd API call.

A redux-thunk implementation will look like this,

let timeout;
var cancelToken = axios.cancelToken;
let apiCall;

function valueChanged(value) {    
    return dispatch => {
        dispatch(loadSuggestionsInProgress());
        dispatch({
            type: 'VALUE_CHANGED',
            payload: {
                value
            }
        });

        // If changed again within 1 second, cancel the timeout
        timeout && clearTimeout(timeout);

        // Make API Call after 1 second
        timeout = setTimeout(() => {
            // Cancel the existing API
            apiCall && apiCall.cancel('Operation cancelled');

            // Generate a new token
            apiCall = cancelToken.source();


            axios.get(`/suggestions?q=${value}`, {
                cancelToken: apiCall.token
            })
                .then(response => dispatch(loadSuggestionsSuccess(response.data.suggestions)))
                .catch(() => dispatch(loadSuggestionsFailed()))

        }, 1000, value);
    }
}

Now, it requires another global variable to store the Axios's cancel token. More global variables = more impure functions!

To implement the same using redux-observable, all we need to do is replace .mergeMap with .switchMap.

const loadSuggestionsEpic = (action$, state$) => action$.pipe(
    ofType('VALUE_CHANGED'),
    throttle(1000),
    switchMap(action => from(axios.get(`/suggestions?q=${action.payload.value}`)).pipe(
        map(response => loadSuggestionsSuccess(response.data.suggestions)),
        catchError(() => loadSuggestionsFailed())
    )),
    startWith(loadSuggestionsInProgress())
);

Since it doesn't require any changes to our action creators, they can continue to be pure functions.

Similarly, there are many use-cases where Redux-Observables actually shines! For example, polling an API, showing snack bars, managing WebSocket connections, etc.

To Conclude

If you are developing a Redux application that involves such complex use-cases, it is highly recommended to use Redux-Observables. After all, the benefits of using it are directly proportional to the complexity of your application, and it is evident from the above mentioned practical use-cases.

I strongly believe using the right set of libraries will help us to develop much cleaner and maintainable applications, and in the long term, the benefits of using them will outweigh the drawbacks.

‘Observables’, ‘Observables’, ‘Observables’...Yes! Today, we will talk about this often discussed word of the market. We'll also learn how they are different from Promises (haven't heard about Promises? Not to worry! You will know more soon). Let’s start!

I first encountered the term Observable when I started learning Angular. Although it’s not an Angular-specific feature, it’s a new way of handling async requests. Async request? You know it, right? No! It’s ok. Let’s first understand what’s an async request is then.

Async Requests

Well! You must have read about asynchronous features in the JavaScript world. 'Asynchrony' in the computer world means that the flow of the program occurs independently. It does not wait for a task to get finished. It moves to the next task.

Now, you might be thinking - what happens to the task that is not finished? The co-worker handles those unfinished tasks. Yes! In the background, a co-worker works and handles those unfinished tasks and once they are done, it sends data back.

This can bring up another question of how we handle the data that is returned. The answer is Promises, Observables, callbacks and many more.

We know that these asynchronous operations return responses, either some data after success or an error. To handle this, concepts like Promises, callbacks, observables came into the market. Well! I will not get into them now as we have deviated from our sub topic i.e 'async' request. (Don't worry! These topics will be discussed soon).

After discussing the above points, you might ha have got a rough picture of what the async request is. Let’s clear it up. An Async request is one where the client does not wait for the response. Nothing is blocked. Let’s understand this concept by looking at a very common scenario.

In the web world, it's quite common to hit the server to get data like the details of a user, a list, and so on. We know it will take time and anything can follow (success/failure).

this case, instead of waiting for data to come, we handle it asynchronously (no waiting) so that our application does not get blocked. Such requests are asynchronous requests. I think now we are clear with it. So let's see how we can actually handle these async requests.

As I already told you, Observables gave us a new way of handling async requests. The other ways are promises, callbacks, and async/await. These are the popular ways. Let’s have a look at two of them which are callbacks and promises.

Callbacks

Callbacks are quite a common one. Callback functions (as their name suggests) are called at the back. That is when the request gets completed and returns the data or error, these functions get called. Have a look at the code for a better understanding:

const request = require(‘request’);
request('https://www.example.com', function (err, response, body) {
  if(error){
    // Error handling
  }
  else {
    // Success 
  }
});

This is one way of handling an async request. But what happens when we want to again request to the server for data after the success of the first request? What if we want to make a third request after that successful second request? Horrible!

At this point, our code will become messy and less readable. This is called ‘callback hell’. To overcome it, promises came around. They offer a better way of handling an async request that improves code readability. Let’s understand a bit more.

Promises

Promises are objects that promise they will have value in the near future - either a success or failure. Promises have their own methods which are then and catch. .then() is called when success comes, else the catch() method calls. Promises are created using the promise constructor. Have a look at code to better understand.

function myAsyncFunction(name){
     return new Promise(function(resolve, reject){
          if(name == ‘Anchal’){
               resolve(‘Here is Anchal’)
         }
         else{
              reject(‘Oops! This is not Anchal’)
        }

     }
} 

myAsyncFunction(‘Anchal’)
.then(function(val){
      // Logic after success
      console.log(val)     // output -  ‘Here is Anchal’
})
.catch(function(val){
    //Logic after failure
     console.log(val)     // output - ‘Oops! This is not Anchal’
})

As you can see, myAsyncFunction is actually promising that it will have some value in near future. .then() or .catch() is called according to the promise's status.

Promises improve code readability. You can see how readable the code is by using promises. Better handling of async operations can be achieved using Promises. This is a brief introduction of what promises are, how they handle data and what beauty promises carry.

Now, it's time to learn about our main topic: Observables.

What are Observables?

Observables are also like callbacks and promises - that are responsible for handling async requests. Observables are a part of the RXJS library. This library introduced Observables.

Before understanding what an observable actually is, you must understand two communication models: pull and push. These two concepts are protocols of how producers of data communicate with the consumers of data.

Pull & Push Model

As I have already told you, Push and Pull are communication protocols between data producers and consumers. Let’s understand both one by one.

Pull Model: In this model, the consumer of data is king. This means that the consumer of data determines when it wants data from the producer. The producer does not decide when the data will get delivered. You can better understand if you relate functions to it.

As we know, functions are responsible for doing some task. For example, dataProducer is a function which is simply returning a string, like "Hi Observable".

function dataProducer(){
   return ‘Hi Observable’;
}

Now, you can see that the above function is not going to decide when it will deliver the ‘Hi Observable’ string. It will decide by the consumer, that is code that calls this function. Consumer is king. The reason why it is called a pull model is that pull task is determining the communication. This is the pull Model. Now, let’s go into the Push Model.

Push Model: In this model, the producer of data is king. Producer determines when to send data to consumer. The Consumer does not know when data is going to come. Let’s understand it by taking an example:

I hope you remember promises. Yes, Promises follow the push model. A Promise (producer) delivers data to the callback (.then() - consumer). Callbacks do not know when data is going to come. Here, promise (producer) is king. It is determining the communication. That’s why it's called Push Model as the producer is in charge.

Like promises, Observables also follow the push model. How? You will get the answer once I elaborate on observables. Let’s get back to observables then.

Observables as functions

To simply understand, you can think of observables as functions. Let’s have a look at the below examples:

function dataProducer(){
    return ‘Hi Observable’
}

var result = dataProducer();
console.log(result) // output -  ‘Hi Observable’

You can get the same behaviour using an observable:

var observable = Rx.Observable.create((observer: any) =>{

   observer.next(‘Hi Observable’);

})

observable.subscribe((data)=>{
   console.log(data);    // output - ‘Hi Observable’
})

From above, you can see both functions and observables show the same behaviour. This may bring a question to your mind - are observables the same as functions? No. I'll clarify in a minute why the answer is no. Have a look at an elaborate version of the above example.

function dataProducer(){
    return ‘Hi Observable’;
    return ‘Am I understandable?’ // not a executable code.
}

var observable = Rx.Observable.create((observer: any) =>{

   observer.next(‘Hi Observable’);
   observer.next( ‘Am I understandable?’ );

})

observable.subscribe((data)=>{
   console.log(data);    
})

Output :
‘Hi Observable’
‘Am I understandable?’

I hope you can now see what difference I wanted to address. From above, you can see, both functions and observables are lazy. We need to call (functions) or subscribe (observables) to get the results.

Subscriptions to observables are quite similar to calling a function. But where observables are different is in their ability to return multiple values called streams (a stream is a sequence of data over time).

Observables not only able to return a value synchronously, but also asynchronously.

var observable = Rx.Observable.create((observer: any) =>{
   observer.next(‘Hi Observable’);
    setTimeout(()=>{
        observer.next(‘Yes, somehow understandable!’)
    }, 1000)   

   observer.next( ‘Am I understandable?’ );
})

output :

‘Hi Observable’
‘Am I understandable?’ 
Yes, somehow understandable!’.

In short, you can say observables are simply a function that are able to give multiple values over time, either synchronously or asynchronously.

You now have an outline about observables. But let’s understand them more by looking into different phases of observables.

Observable Phases

We have already seen from the above example how observables create and execute and come into play by subscription. Hence, there are four stages through which observables pass. They are:

1. Creation
2. Subscription.
3. Execution
4. Destruction.

Creation of an observable is done using a create function.

var observable = Rx.Observable.create((observer: any) =>{
})

To make an observable work, we have to subscribe it. This can be done using the subscribe method.

observable.subscribe((data)=>{
   console.log(data);    
})

Execution of observables is what is inside of the create block. Let me illustrate with the help of an example:

var observable = Rx.Observable.create((observer: any) =>{

   observer.next(‘Hi Observable’);        
    setTimeout(()=>{
        observer.next(‘Yes, somehow understandable!’)
    }, 1000)   

   observer.next( ‘Am I understandable?’ );

})

The above code inside the create function is observable execution. The three types of values that an observable can deliver to the subscriber are:

observer.next(‘hii’);//this can be multiple (more than one)

observer.error(‘error occurs’) // this call whenever any error occus.

Observer.complete(‘completion of delivery of all values’) // this tells the subscriptions to observable is completed. No delivery is going to take place after this statement.

Let’s have a look below to understand all three values:

var observable = Rx.Observable.create((observer: any) =>{
try {
   observer.next(‘Hi Observable’);                                       
    setTimeout(()=>{
        observer.next(‘Yes, somehow understandable!’)
    }, 1000)   

   observer.next( ‘Am I understandable?’ );

   observer.complete();

   observer.next(‘lAST DELIVERY?’ );  
   // above block is not going to execute as completion notification is      already sent.
   }
catch(err){
     observer.error(err);    
  }

})

Last phase that comes into the market is destruction. After an error or a complete notification, the observable is automatically unsubscribed. But there are cases where we have to manually unsubscribe it. To manually do this task, just use:

var subscription = observable.subscribe(x => console.log(x)); // Later: subscription.unsubscribe();

This is all about the different phases through which an observable passes.

I think, now, we know what observables are? But what about the other question which is - how observables are different from promises? Let’s find the answer to it.

Promises vs observables

As we know, promises are for handling async requests and observables can also do the same. But where do they differ?

Observables are lazy whereas promises are not

This is pretty self-explanatory: observables are lazy, that is we have to subscribe observables to get the results. In the case of promises, they execute immediately.

Observables handle multiple values unlike promises

Promises can only provide a single value whereas observables can give you multiple values.

Observables are cancelable

You can cancel observables by unsubscribing it using the unsubscribe method whereas promises don’t have such a feature.

Observables provide many operators

There are many operators like map, forEach, filter etc. Observables provide these whereas promises does not have any operators in their bucket.

These are features that makes observables different from promises.

Now, it's time to end. I hope you have a better understanding of the hot topic of observables!

Thanks for reading!

A Subject is a “special” type of observable that allows us to broadcast values to multiple subscribers. The cool thing about Subjects, is that it provides a real-time response.

For instance, if we have a subject with 10 subscribers, whenever we push values to the subject, we can see the value captured by each subscriber

This introduces a couple challenges; what if we push some values, and then subscribe, or vice-verse? Timing plays an important role, if we subscribe late, we won’t be able to access the values, similar to if someone enters a live-sport event on TV 30 minutes later.

Luckily, we have 4 types of Subjects that allows us to “time travel” in which we can access values even though we subscribe late or not.

Topics we’ll cover:

1. What is a Subject with a practical example
2. BehaviorSubject: Get the last message
3. ReplaySubject: Time travel
4. AsyncSubject: Once completed, get the last message

1. What is a Subject?

As mentioned, a Subject is nothing more like an observable with a few more characteristics. An observable is by definition an invokable collection that emits data once subscribed. Meanwhile, a Subject is where we control the state of “when to emit data” to multiple subscribers.

A Subject allows us to invoke methods like .next(), .complete() and .error() outside, while in an observable, we invoke these methods as callbacks.

// Creating an Observable
const observable = new Observable((observer) => {
    observer.next(10);
    observer.next(5);
    observer.complete();
});

// Creating a Subject
const subject = new Subject();
subject.next(10);
subject.next(5);
subject.complete();

Practical example: Let’s build a simple chat group using a Subject

Let’s imagine we are building a simple chat app where people can post messages to the chat group. The first step is to create an instance of the Subject, and then assign it to a chatGroup.

// Create subject "Observable"
const chatGroup = new Subject();

Now that our chat group (Subject) is created, the next thing to do is add messages. Let’s create a typical conversation between two friends.

// Push values to the stream
chatGroup.next('David - Hi, which hot series do you recommend?');
chatGroup.next('Peter - Game of Thrones, Bodyguard or Narcos are few of the good ones');
chatGroup.next('David - Interesting, which one is the hottest?');
chatGroup.next('Peter - Game of Thrones!');

So far so good — now we have 4 messages posted in our chat group, so what happens if we subscribe? Or let’s say a new friend named John wants to join the conversation. Is he be able to see the old messages?

// Print messages
chatGroup.subscribe((messages) => {
    console.log(messages)
})

Unfortunately not, John misses the conversation because he’s subscribes late. This is a perfect example of how reactive programming works — the idea of values passing over time, and thus, we must subscribe in the right time to access the values.

To further elaborate on the previous example, what if John enters in the middle of the conversation?

// Push values to the stream
chatGroup.next('David - Hi, which hot series do you recommend?');
chatGroup.next('Peter - Game of Thrones, Bodyguard or Narcos are few of the good ones');

// John enters the conversation 
chatGroup.subscribe((messages) => {
    console.log(messages)
});

chatGroup.next('David - Interesting, which one is the hottest?');
chatGroup.next('Peter - Game of Thrones!');

// OUTPUT
// David - Interesting, which one is the hottest?
// Peter - Game of Thrones!

Once John subscribes, he sees the two last messages. The Subject is doing what it’s intended to do. But what if we want John to view all messages, or just the last one, or get notified when a new message is posted?

In general, these Subjects are mostly similar, but each one provides some extra functionality, let’s describe them one by one.

2. BehaviorSubject: Get last message

The BehaviorSubject is similar to a Subject except it requires an initial value as an argument to mark the starting point of the data stream. The reason is because when we subscribe, it returns the last message. This is similar concept when dealing with arrays; where we do array.length-1 to get the latest value.

import {BehaviorSubject } from "rxjs";

// Create a Subject
const chatGroup = new BehaviorSubject('Starting point');

// Push values to the data stream
chatGroup.next('David - Hi, which hot series do you recommend?');
chatGroup.next('Peter - Game of Thrones, Bodyguard or Narcos are few of the good ones');
chatGroup.next('David - Interesting, which one is the hottest?');
chatGroup.next('Peter - Game of Thrones!');

// John enters the conversation
chatGroup.subscribe((messages) => {
    console.log(messages)
})

// OUTPUT
// Peter - Game of Thrones!

3. ReplaySubject: Time travel

The ReplaySubject, as the name implies, once subscribed it broadcasts all messages, despite if we subscribed late or not. It’s like time travel, where we can access all the values that were broadcast.

import { ReplaySubject } from "rxjs";

// Create a Subject
const chatGroup = new ReplaySubject();

// Push values to the data stream
chatGroup.next('David - Hi, which hot series do you recommend?');
chatGroup.next('Peter - Game of Thrones, Bodyguard or Narcos are few of the good ones');
chatGroup.next('David - Interesting, which one is the hottest?');
chatGroup.next('Peter - Game of Thrones!');

// John enters the conversation
chatGroup.subscribe((messages) => {
    console.log(messages)
})

// OUTPUT
// David - Hi, which hot series do you recommend?'
// Peter - Game of Thrones, Bodyguard or Narcos are few of the good ones'
// David - Interesting, which one is the hottest?'
// Peter - Game of Thrones!'

4. AsyncSubject: Once completed, get last message

The AsyncSubject is similar to BehaviorSubject in terms of emitting the last value once subscribed. The only difference is that it requires a complete() method to mark the stream as completed. Once that is done, the last value is emitted.

import { AsyncSubject } from "rxjs";

// Create a Subject
const chatGroup = new AsyncSubject();

// Push values to the data stream
chatGroup.next('David - Hi, which hot series do you recommend?');
chatGroup.next('Peter - Game of Thrones, Bodyguard or Narcos are few of the good ones');
chatGroup.next('David - Interesting, which one is the hottest?');
chatGroup.next('Peter - Game of Thrones!');

chatGroup.complete(); // <-- Mark the stream as completed

// John enters the conversation
chatGroup.subscribe((messages) => {
    console.log(messages)
})

// OUTPUT
// Peter - Game of Thrones!'

Summary

Back to our previous example with John, we can now decide if we want John to access the whole conversation (ReplaySubject), the last message (BehaviorSubject), or the last message once the conversation completes (AsyncSubject).

If you ever struggle to identify if a Subject is the right way to go, the article “To Use a Subject or Not to Use a Subject” by Dave Sixton describes when to use Subjects based on two criteria:

1. Only when one want to convert a cold Observable into a hot observable.
2. Generate a hot observable that passes data continuously.

In short, only creativity limits the potential use of reactive programming. There will be some scenarios where Observables do most of the heavy-lifting, but understanding what Subjects are, and what type of Subjects exists, will definitely improve your reactive programming skills.

If you are interested to learn more about the web-ecosystem, here are few articles I’ve written, enjoy.

• A comparison between Angular and React
• A practical guide to ES6 modules
• [How to perform HTTP requests using the Fetch API](http://A practical ES6 guide on how to perform HTTP requests using the Fetch API)
• Important web concepts to learn
• Boost your skills with these JavaScript methods
• Create custom bash commands

You can find me on Medium where I publish on a weekly basis. Or you can follow me on Twitter, where I post relevant web development tips and tricks.

One of the most challenging things for new developers to learn is the observer pattern. Understanding how to effectively use it with RxJS to deal with asynchronous data such as user events, HTTP requests or any other events that require waiting for something to complete is tricky.

What most people struggle with is the new approach. It requires a different mindset where visualization plays an important role. We think of data as a sequence of values that passes over time rather than as one single value that is retrieved once. This mindset is known as reactive programming.

Since the Observer pattern is a fairly large ecosystem consisting of many important parts, I’ve chosen to narrow it down by focusing on Observables only. I’ll share other articles soon that cover the rest of the Observer pattern, such as how to deal with RxJS.

Topics we’ll cover:

1. What does asynchronous really mean?
2. Which pattern to use (Observer or Promise)
3. How to create an Observable (code examples start here)
4. How to subscribe to an Observable
5. How to unsubscribe to an Observable

1. What does asynchronous really mean?

One of the things with the web, and the majority of languages, is that once you ask for data such as requesting a list of users from the server, you can’t guarantee that the data will be returned. There is an uncertainty issue.

One of the reasons may be that the data is not present, the server may be broken, or the HTTP URL is not valid because someone has changed the query string.

For that reason, along with a few others, we need to deal with such data asynchronously. We request the list of users, and wait until it is retrieved, but don’t stop the whole application for a simple operation.

It’s like telling a coworker to solve a task instead of sending the whole team; that would be an expensive and not a wise approach to take.

Let’s clarify a misconception: the terms synchronous or asynchronous have nothing to do with multi-threading, where operations are executed at the same time. It simply means the operations are either dependent on or independent from each other, that’s it.

Let’s compare the difference between synchronous and asynchronous to better understand how they really work.

What is Synchronous?

With Synchronous events, you wait for one to finish before moving on to another task.

Example: You are in a queue to get a movie ticket. You cannot get one until everybody in front of you gets one, and the same applies to the people queued behind you. Answered by themightysapien.

What is Asynchronous?

With asynchronous events, you don’t wait, you can move on to the next task until the data is available.

Example: You are in a restaurant with many other people. You order your food. Other people can also order their food, they don’t have to wait for your food to be cooked and served to you before they can order. In the kitchen, restaurant workers are continuously cooking, serving, and taking orders. People will get their food served as soon as it is cooked. Answered by themightysapien.

Alright, so in short, this allows us to either wait for operations to happen before we can move on, or not wait until the data is ready.

2. Which pattern to use (Observer or Promise)

First of all, both the observer pattern and the promise pattern deal with asynchronous operations. Operations such as user events or HTTP requests, or any other events that execute independently.

The majority of operations today need some type of asynchronous/synchronous handling, and understanding how it works plays an important role when building robust apps.

It’s not meant to make your life harder, but easier. However, it thus requires a learning-curve which may be a painful approach, but the reward at the end is well worth it.

Stay with one pattern

The difference lies in the complexity of the application. If you deal with a small app where the task is to simply get a list of users from the server, or to show active members, then promises with the Fetch API (read more) work fine.

But if you deal with a large application with many asynchronous operations that require changing the data, performing multiple operations on a data stream, or reusing it in multiple places, then the observer pattern works great.

Can I use both patterns in one project?

Yes, but it’s not recommended that you mix between two architectures which basically do the same thing (handle asynchronous events). Instead, stick with one, and learn more about it.

Boost your skills with RxJS

With RxJS, you have access to 189 operators with documentation + other great resources. And each of these operators are simply callbacks that do something on the data stream.

If you are familiar with JavaScript’s functional prototypes (methods) such as map(), filter(), and reduce(), you’ll find them in RxJS. Note, the concept is the same but the written code is not.

So what is the difference between these two patterns?

Observable vs Promise

Here’s a quick comparison between the observer pattern and the promise pattern. The key points are that a promise emits a single value(s) once the .then() callback is used, while an Observable emits multiple values as a sequence of data that passes over time. Another important point is that an Observable can be canceled or retried while a promise cannot. However, there are external packages that make it possible to cancel a promise.

3. How do we create an Observable?

Here are a couple of ways one can create an Observable:

• create an Observable from scratch
• turn a promise into an Observable
• or use a framework that does it for you behind the scenes, such as Angular.

Did you know that Angular uses the Observer pattern extensively? All asynchronous operations such as HTTP GET or listening for events or value changes follow the observer pattern.

If you ever want to mimic (test) a real-world scenario, so to say pass values over time, I highly recommend using the interval function. This passes values after x time in milliseconds. So if you have an interval where x is 2000ms — it passes each value (increments) after 2 seconds.

4. How do we subscribe to an Observable?

An Observable is simply a collection of data that waits to be invoked (subscribed) before it can emit any data. If you’ve worked with promises, then the way to access the data is to chain it with the then() operator or use the ES6 async/await.

So to follow the previous example, how does one access the data?

As shown above, when we subscribe, we tell the Observable to pass us whatever it holds. It can be an array, a collection of events, or a sequence of objects and so forth.

A common beginner-mistake I’ve seen among developers is that they do many operations on the Observable but get frustrated because they can’t see any results. You are not alone! I’ve made this mistake a couple of times and as a thumb-rule — always remember to subscribe.

5. How do we unsubscribe to an Observable?

It is important to unsubscribe, otherwise we end up with a memory leak which slows down the browser. If you’ve worked with Angular, there is a pipe named asyncPipe which subscribes and unsubscribes automatically for you.

The way we unsubscribe is that we create a reference to each Observable that is subscribed by creating a variable to preserve its current state. And then, for each variable, we chain it with the unsubscribe() method. Remember that you can only unsubscribe after you’ve subscribed. It’s fairly simple but often forgotten.

Notice, if you unsubscribe here, Observable_1 and Observable_2 will output the data before it is unsubscribe because these are cold Observables (not time-dependent), while Observable_3 and Observable_4 will not output anything because these are hot Observables (time-dependent).

Summary

As mentioned above, the most challenging part of learning the observer pattern is the mindset. A mindset where we look at values differently, such as a sequence of data that emits over time. In this article, we’ve covered types of ways we can create an Observable, as well as how to subscribe and unsubscribe.

I recommend using the observer pattern because it provides everything that the promise pattern offers, and much more. It also provides a few great operators to prevent users from sending thousands of unnecessary requests to the backend.

One of them is debonceTime which gives the user enough time to write a complete word, and then send one request instead of a request for every character. You can, of course, achieve this with a simple promise, but that requires some lines of code.

I’ll cover more about reactive programming in the near future, stay tuned!

If you are interested to learn more about the web-ecosystem, here are few articles I’ve written to boost your web skills, enjoy :)

• Boost your skills with these JavaScript methods
• A comparison between Angular and React
• A practical guide to ES6 modules
• [How to perform HTTP requests using the Fetch API](http://A practical ES6 guide on how to perform HTTP requests using the Fetch API)
• Important web concepts to learn

If you want to become a better web developer, start your own business, teach others, or improve your development skills, you can find me on Medium where I publish on a weekly basis. Or you can follow me on Twitter, where I post relevant web development tips and tricks.

P.S. If you enjoyed this article and want more like these, please clap ❤ and share with friends that may need it, it’s good karma.

Remember: .pipe() is not .subscribe()!

_Look! A lightning tip! (Original ? by M[ax Bender)](https://unsplash.com/photos/iF5odYWB_nQ?utm_source=unsplash&utm_medium=referral&utm_content=creditCopyText" rel="noopener" target="blank" title=")

This article is directed at the beginners trying to increase their RxJs knowledge but can also be a quick refresh or a reference to show to beginners for more experienced developers!

Today we are going to keep it short and straight to the point!

Currently I am working in a rather large organization quite a few teams and projects (more than 40 SPAs) that are in the process of migration to Angular and therefore also RxJs.

This represents a great opportunity to get in touch with the confusing parts of RxJs which can be easy to forget once one masters the APIs and focuses on the implementation of the features instead.

The “.subscribe()” function

RxJs observable represent a “recipe” of what we want to happen. It’s declarative which means that all the operations and transformations are specified in their entirety from the get go.

An example of an observable stream could look something like this…

Example of the RxJs observable stream declaration

This RxJs observable stream will do literally nothing by itself. To execute it we have to subscribe to it somewhere in our codebase!

This subscription will log our greetings every odd minute

In the example above, we provided a handler only for the values emitted by the observable. The subscribe function itself accepts up to three different argument to the handle next value, error or complete event.

Besides that we could also pass in an object with the properties listed above. Such an object is an implementation of the Observer interface. The advantage of observer is that we don’t have to provide implementation or at least a null placeholder for the handlers we are not interested in.

Consider the following example…

In the code above, we are passing an object literal which contains only complete handler, the normal values will be ignored and errors will bubble up the stack.

And in this example, we are passing the handler of the next error and complete it as direct arguments of the subscribe function. All unimplemented handlers have to be passed as a null or undefined until we get to the argument we’re interested in.

As we can see, the inline argument style of implementation of a .subscribe() function call is positional.

In my experience, the inline arguments style is the one which is most common in various projects and organizations.

Unfortunately, many times we may encounter implementation like the following…

Example of redundant handlers often encountered in the “wild”

The example above contains redundant handlers for both next and error handlers which do exactly nothing and could have been replaced by null.

Even better would be to pass the observer object with the complete handler implementation, omitting other handlers altogether!

The “.pipe()” and the operators

As beginners are used to providing three arguments to subscribe, they often try to implement a similar pattern when using similar operators in the pipe chain.

RxJs operators, which are often confused with the .subscribe() handlers, are catchError and finalize. They both serve a similar purpose too — the only difference being that they are used in the context of the pipe instead of the subscription.

In case we would like to react to the complete event of every subscription of the RxJs observable stream, we could implement finalize operator as a part of the observable stream itself.

That way we don’t have to depend on the developers to implement complete handlers in the every single .subscribe() call. Remember, the observable stream can be subscribed to more than once!

Use the finalize operator to react to the complete event of the stream independently from the subscription. (Similar to tap)

This brings us to the final and arguably most problematic pattern we may encounter when exploring various code bases: redundant operators added when trying to follow .subscribe() pattern in the .pipe() context.

Also, we might encounter its even more verbose cousin…

Stuff might get verbose…

Notice we have progressed from the original single line to the full nine lines of code which we have to read and understand when we want to fix a bug or add a new feature.

Stuff might get even more complex when combined with more complex generic Typescript types, which can make the whole code block even more mysterious (and hence waste more of our time).

Recapitulation

1. The .subscribe() method accepts both the observer object and inline handlers.
2. The observer object represents the most versatile and concise way to subscribe to an observable stream.
3. In case we want to go with the inline subscribe arguments (next, error, complete) we can provide null in place of a handler we don’t need.
4. We should make sure that we don’t try to repeat the .subscribe() pattern when dealing with .pipe() and operators.
5. Always strive to keep the code as simple as possible and remove unnecessary redundancies!

That’s it! ✨

I hope you enjoyed this article and will now have better understanding of how to subscribe to RxJs observables with clean, concise implementation!

Please support this guide with your ??? using the clap button and help it spread to a wider audience ? Also, don’t hesitate to ping me if you have any questions using the article responses or Twitter DMs @tomastrajan.

And never forget, the future is bright

_[avier Coiffic)](https://twitter.com/tomastrajan" rel="noopener" target="_blank" title="">Obviously the bright future! (? by X<a href="https://unsplash.com/photos/WV4B_aVj0aQ?utm_source=unsplash&utm_medium=referral&utm_content=creditCopyText" rel="noopener" target="blank" title=")

Starting an Angular project? Check out Angular NgRx Material Starter!

Angular NgRx Material Starter with built in best practices, theming and much more!

If you made it this far, feel free to check out some of my other articles about Angular and frontend software development in general…

?‍?️ The 7 Pro Tips To Get Productive With Angular CLI & Schematics ?
Ang_ular Schematics is a workflow tool for the modern web — official introduction articlehac_kernoon.com The Best Way To Unsubscribe RxJS Observable In The Angular Applications!
_There are many different ways how to handle RxJS subscriptions in Angular applications and we’re going to explore their…_blog.angularindepth.comTotal Guide To Angular 6+ Dependency Injection — providedIn vs providers:[ ] ?
L_et’s learn when and how to use new better Angular 6+ dependency injection mechanism with new providedIn syntax to make…m_edium.com The Ultimate Answer To The Very Common Angular Question: subscribe() vs | async Pipe
_Most of the popular Angular state management libraries like NgRx expose application state in a form of a stream of…_blog.angularindepth.com

Introduction

After getting to know RxJS (thanks to Angular!), I realized that it’s surprisingly a good fit for handling web scraping operations.

I tried it out in a side project and I would like to share my experience with you. Hopefully, this will open your eyes to how reactive programming can make your life simpler.

The codes can be found on https://github.com/jacobgoh101/web-scraping-with-rxjs

Requirements

• Node
• RxJS and intermediate understanding of it
• cheerio: it allows you to use jQuery-like syntax to extract information out of HTML code
• request-promise-native: for sending HTTP request

Hypothetical Goal

Everybody loves a good comedy movie.

Let’s make it our goal to scrape a list of good comedy movies from IMDB.

There are only 3 requirements that the target data needs to fulfill:

• it is a movie (not TV shows, music videos, etc)
• it is a comedy
• it has a rating of 7 or higher

Get Started

Let’s set our base URL and define a BehaviorSubject allUrl$ that uses the base URL as the initial value.

(A BehaviorSubject is a subject with an initial value.)

const { BehaviorSubject } =  require('rxjs');const  baseUrl  =  `https://imdb.com`;const  allUrl$  =  new  BehaviorSubject(baseUrl);

allUrl$ is going to be the starting point of all crawling operations. Every URL will be passed into allUrl$ and be processed on later.

Making sure that we scrape each URL only once

With the help of distinct operators and normalize-url, we can easily make sure that we never scrape the same URL twice.

// ...const { map, distinct, filter } =  require('rxjs/operators');const  normalizeUrl  =  require('normalize-url');

// ...

const  uniqueUrl$  =  allUrl$.pipe(    // only crawl IMDB url    filter(url  =>  url.includes(baseUrl)),    // normalize url for comparison    map(url  =>  normalizeUrl(url, { removeQueryParameters: ['ref', 'ref_']     })),    // distinct is a RxJS operator that filters out duplicated values    distinct());

It’s time to start scraping

We are going to make a request to each unique URL and map the content of each URL into another observable.

To do that, we use mergeMap to map the result of the request to another observable.

const { BehaviorSubject, from } =  require('rxjs');const { map, distinct, filter, mergeMap } = require('rxjs/operators');const rp = require('request-promise-native');const  cheerio  =  require('cheerio');

//...const urlAndDOM$ = uniqueUrl$.pipe(  mergeMap(url => {    return from(rp(url)).pipe(      // get the cheerio function $      map(html => cheerio.load(html)),      // add URL to the result. It will be used later for crawling      map($ => ({        $,        url      }))    );  }));

urlAndDOM$ will emit an object consisting of 2 properties, which are $ and url. $ is a Cheerio function where you can use something like $('div').text() to extract information out of raw HTML code.

Crawl all the URLs

const { resolve } =  require('url');//...

// get all the next crawlable URLsurlAndDOM$.subscribe(({ url, $ }) => {  $('a').each(function(i, elem) {    const href = $(this).attr('href');    if (!href) return;

// build the absolute url    const absoluteUrl = resolve(url, href);    allUrl$.next(absoluteUrl);  });});

In the code above, we scrape all the links inside the page and send them to allUrl$ to be crawled later.

Scrape and save the movies we want!

const  fs  =  require('fs');//...

const isMovie = $ =>  $(`[property='og:type']`).attr('content') === 'video.movie';const isComedy = $ =>  $(`.title_wrapper .subtext`)    .text()    .includes('Comedy');const isHighlyRated = $ => +$(`[itemprop="ratingValue"]`).text() > 7;

urlAndDOM$  .pipe(    filter(({ $ }) => isMovie($)),    filter(({ $ }) => isComedy($)),    filter(({ $ }) => isHighlyRated($))  )  .subscribe(({ url, $ }) => {    // append the data we want to a file named "comedy.txt"    fs.appendFile('comedy.txt', `${url}, ${$('title').text()}\n`);  });

Yup, we just created a web scraper

In around 70 lines of code, we have created a web scraper that

• automatically crawled URLs without unnecessary duplicates
• automatically scrapes and saves the info we want in a text file

You may see the code up to this point in https://github.com/jacobgoh101/web-scraping-with-rxjs/blob/86ff05e893dec5f1b39647350cb0f74efe258c86/index.js

If you have ever tried writing a web scraper from scratch, you should be able to see now how elegant it is to write one with RxJS.

But we are not done yet…

In an ideal world, the code above should work forever without any problems.

But in reality, silly errors happen.

Handling Errors

Limit the number of active concurrent connections

If we send too many requests to a server in a short period of time, it’s likely that our IP will be temporarily blocked from making any further requests, especially for an established website like IMDB.

It’s also considered rude/unethical to send so many requests at once because it would create a heavier load on the server and in some cases, crash the server.

mergeMap has built-in functionality to control concurrency. Simply add a number to the 3rd function argument and it will limit the active concurrent connection automatically. Graceful!

const maxConcurrentReq = 10;//...const urlAndDOM$ = uniqueUrl$.pipe(  mergeMap(    //...    null,    maxConcurrentReq  ));

Code Diff: https://github.com/jacobgoh101/web-scraping-with-rxjs/commit/6aaed6dae230d2dde1493f1b6d78282ce2e8f316

Handle and Retry Failed Request

Requests may fail randomly due to dead links or server-side rate limiting. This is crucial for web scrapers.

We can use catchError, retry operators to handle this.

const { BehaviorSubject, from, of } =  require('rxjs');const {  // ...  retry,  catchError} = require('rxjs/operators');//...

const maxRetries = 5;// ...

const urlAndDOM$ = uniqueUrl$.pipe(  mergeMap(    url => {      return from(rp(url)).pipe(        retry(maxRetries),        catchError(error => {          const { uri } = error.options;          console.log(`Error requesting ${uri} after ${maxRetries} retries.`);          // return null on error          return of(null);        }),        // filter out errors        filter(v => v),        // ...      );    },

Code Diff: https://github.com/jacobgoh101/web-scraping-with-rxjs/commit/3098b48ca91a59aa5171bc2aa9c17801e769fcbb

Improved Retry Failed Request

Using retry operator, the retry would happen immediately after the request failed. This is not ideal.

It’s better to retry after a certain period of delay.

We can use the genericRetryStrategy suggested in learnrxjs to achieve this.

Code Diff: https://github.com/jacobgoh101/web-scraping-with-rxjs/commit/e194f4ff128a573241055ffc0d1969d54ca8c270

Conclusion

To recap, in this post, we discussed:

• how to crawl a web page using Cheerio
• how to avoid duplicated crawl using RxJS operators like filter, distinct
• how to use mergeMap to create an observable of request’s response
• how to limit concurrency in mergeMap
• how to handle error
• how to handle retry

I hope this has been helpful to you and has deepened your understanding of RxJs and web scraping.

Originally published at dev.to.

We are all likely familiar with the concept of a Joystick.

We start holding the handle of the Joystick, we move the handle around, and when we release it, the handle gently goes back to its initial position.

Now, what if we want to build some sort of software component that simulates the behaviour of a Joystick in the browser?

Well, with RxJS this turns out to be pretty simple. And it is also an interesting exercise to prove your Reactive thinking. You can jump directly to the code here if you want, otherwise keep reading and see what we can do.

Which are the events we are interested in?

The behaviour of the Joystick can be seen a series of events combined together in some way.

The first event we are interested in is when the user presses a mouse on the handle (mousedown) - the handle is just the central part of the Joystick image.

If you hold the mouse pressed, then you can move around and you see the handle move accordingly — the mousemove events of the mouse are therefore the second series of events we want to capture.

Last, we need to consider when the user releases the mouse (mouseup) since this is the event that causes the Joystick handle to go back to its initial position.

Relevant events for the Joystick case

The whole sequence can be repeated after the handle is released. The mouse is pressed on the handle, then it is moved, then it is released. Again and again.

This repetition can be seen as a stream of events. We can say that the behaviour of a joystick is governed by this stream of events.

The stream of events of a Joystick

If we are able to build such stream of events, we are in a good position to reach our objective — that is, to implement a Joystick software component for the browser using RxJS.

The building blocks with RxJS

The browser actually provides us with the notification of the events we are interested in: the mousedown event on the DOM element representing the handle of the Joystick, and the mousemove and mouseup events at DOM document level.

RxJS, on its side, comes with the function fromEvent that allows us to create an Observable from a browser event.

Create an Observable with fromEvent RxJS function

Using this mechanism we can create the three streams of events which are going to be the building blocks of our solution: mouse_DOWN_Obs, mouse_MOVE_Obs, mouse_UP_Obs.

The streams of events which are our building blocks

But these are just our building blocks. We need to do something with them in order to get what we want: we need to ignore all mousemove events which occur before the first mousedown and then ignore all the mousemove events which occur after the next mouseup. Then we repeat all this again when a new mousedown event occurs. These compose the “stream of events for the Joystick”.

Joystick event stream built from building blocks

The transformation of Observables via composition of operators

Everything starts when the user presses the mouse on the handle of the Joystick, i.e. mouse_DOWN_Obs. We can call it the source Observable.

Once we are notified of an event from mouse_DOWN_Obs we have to switch and start listening to mouse_MOVE_Obs.

The first transformation with switchMap

It may seem like we have not achieved much, but in fact we are now in a position where we can take the mouse_MOVE_Obs notifications until we hear from mouse_UP_Obs. At this point we stop just to restart at the next notification from mouse_DOWN_Obs.

The second transformation with takeUntil

Notice that we apply takeUntil to mouse_MOVE_Obs, because this is the Observable we want to complete. If we had applied one level higher, to mouse_DOWN_Obs, this is what would have happened:

Just the first sequence of move events would have been notified, and then the stream of events would have been closed down. No more events for the Joystick.

Now is the time of side effects

We have learned how to build a stream of all the events relevant for a Joystick. To do something useful with this stream, we need to link the events to some sort of action we want to do. More specifically:

• when we sense a mousemove event we have to change the position of the handle on the browser
• when we sense a mouseup event we have to gently move the handle back to its original position, setting some transition style
• when we sense a mousedown event we have to reset the transition style

But careful. Not all are mousemove events, not all are mouseup events, and not all are mousedown events. Only those which belong to the set of “relevant events for the Joystick”. For instance, we are not interested in all mousemove events which happen before the Joystick has been activated (pressing the mouse on the handle) or after the Joystick handle has been released.

Where we need side effects

Let’s get back to our main track of reasoning. We need to do something at the occurrence of some events. Something that changes the state of the system. In our case, this is the position of the handle on the browser. In functional programming terms these are called side effects, i.e. functions which change the state of the system.

Side effects as functions — Functions as side effects

RxJS gives us two ways to implement side effects.

The first one is the subscribe method of Observable. The second one is the tap operator, formerly know as do, which “performs a side effect for every emission on the source Observable, but returns an Observable that is identical to the source” — the side effect is determined by the function passed to tap as a parameter. tap is the method we are going to use.

Eventually this is the core of the code that implements our Reactive Joystick

const handle = document.getElementById("handle");const mouse_DOWN_Obs = rxjs.fromEvent(handle, 'mousedown');const mouse_MOVE_Obs = rxjs.fromEvent(document, 'mousemove');const mouse_UP_Obs = rxjs.fromEvent(document, 'mouseup');

function activateJoytick() {  mouse_DOWN_Obs.pipe(    rxjs.operators.tap(() => joystickActivated()),    rxjs.operators.switchMap(() => mouse_MOVE_Obs.pipe(      rxjs.operators.takeUntil(mouse_UP_Obs.pipe(        rxjs.operators.tap(() => joystickReleased())      )),    )),    rxjs.operators.tap(event => showHandleInNewPosition(event))  )  .subscribe();}

Example code

You can find the example code here, where you can compare the RxJS implementation with one built using pure JavaScript.

I loved DragonBall Z as a kid, and still love it as an adult.

Among the ludicrous number of transformations, the original Super Saiyan remains my favorite.

Nothing quite like the original

I’m also loving RxJS the more I level up with it, so why not combine these two for the ultimate showdown?

Let’s Go Super Saiyan

With four sprite sheets and a bit of HTML, CSS, and RxJS, we can recreate this legendary transformation!

This is what we’ll be making. Exciting, right?! ?

Setup

Everything’s on my GitHub.

cd ./wherever-you-want
git clone [https://github.com/yazeedb/dbz-rxjs](https://github.com/yazeedb/dbz-rxjs)
cd dbz-rxjs

Open index.html in your favorite browser, and the project in your favorite text editor, and you’re ready to go!

No npm installs today ?

And going forward, I’ll use the acronym “SSJ” instead of “Super Saiyan” for brevity.

First Day of Training

You’ll notice that Goku’s already moving. Since we’re focusing on RxJS, we’ll just skim the project’s starting point.

Here’s the main HTML:

<div id="root">
  <div id="meter-container">
    <span>Hold any key to POWER UP!</span>
    <div id="meter"></div>
  </div>

  <div id="sprite" class="base"></div>
</div>

The bottom div has class="base", which corresponds to this CSS:

.base,
.ssj {
  width: 120px;
  height: 250px;
  animation: stand 0.8s steps(2) infinite;
}

.base {
  background-image: url('img/goku-standing-sheet.png');
}

This sets Goku’s width, height, and standing animation.

If you look at his base/ssj sprite sheets, it’s two different positions and we’re switching between them every 0.8 seconds.

The switching’s handled towards the bottom of style.css:

@keyframes stand {
  from {
    background-position: 0px;
  }
  to {
    background-position: -255px;
  }
}

Same thing for power up:

@keyframes powerup {
  from {
    background-position: 0px;
  }
  to {
    background-position: -513px;
  }
}

We’ll cover the power up meter when we manipulate it.

Mastering the DOM Elements

index.html already includes RxJS@6.2.1 via CDN, so you’re covered.

In app.js, let’s capture the DOM elements we’re interested in:

const sprite = document.querySelector('#sprite');
const meterContainer = document.querySelector('#meter-container');
const meter = document.querySelector('#meter');

I prefer to alias document.querySelector so using it doesn’t cause me wrist pain.

const $ = document.querySelector.bind(document);**
const sprite = $('#sprite');
const meterContainer = $('#meter-container');
const meter = $('#meter');

Next, we’ll create a main function and immediately call it.

// ...

const main = () => {
  // do something
};
main();

Powering Up

Here is main’s first code snippet:

const main = () => {
  const { fromEvent } = rxjs;

  const begin = fromEvent(document, 'keydown');
  const end = fromEvent(document, 'keyup');
};

Goku should power up when a key is held down, and stop when that key is let go. We can use the fromEvent operator to create two observables:

• begin: Notifies when the user presses a key down.
• end: Notifies whenever the user lets go of a key.

Then we can subscribe to these emissions and act upon them. To get the power up animation, give sprite the powerup class name.

begin.subscribe(() => {
  sprite.classList.add('powerup');
});

It works, but pressing a key causes him to power up forever…

We must also subscribe to the end observable, so we know when the key has been let go.

end.subscribe(() => {
  sprite.classList.remove('powerup');
});

Now he powers up and down at your command.

Building a Scouter

Any DBZ fan has seen a scouter, the little eyewear used to track power levels (until like episode 20…).

Obligatory > 9000 joke

As Saiyans power up, their power level grows. Inconceivable, right?

We need a way to track Goku’s power level as he ascends, and trigger the SSJ transformation after say, 100 points.

We can start his power off at 1, and increase it while the user holds a key down.

RxJS Operators

Operators are where RxJS really shines. We can use pure functions to describe how data should transform through the stream.

When the user holds a key down, let’s transform those emissions into a number that increases over time.

Scan

The scan operator is perfect for this. It’s like Array.reduce, but it emits as it’s reducing.

For example, if you have an array of numbers:

nums = [1, 2, 3, 4, 5];

And wish to add them up, reduce is a great choice.

nums.reduce((a, b) => a + b, 0);
// 15

What if you want to see each addition as it happens?

Enter scan. You can run this in our app’s console.

const { from } = rxjs;
const { scan } = rxjs.operators;

from([1, 2, 3, 4, 5])
  .pipe(scan((a, b) => a + b, 0))
  .subscribe(console.log);

// 1 (0 + 1)
// 3 (1 + 2)
// 6 (3 + 3)
// 10 (6 + 4)
// 15 (10 + 5)

See how the emissions increase over time? We can do that with Goku as he powers up!

const { fromEvent } = rxjs;
const { scan, tap } = rxjs.operators;

const begin = fromEvent(document, 'keydown');
const end = fromEvent(document, 'keyup');

begin
  .pipe(
    scan((level) => level + 1, 1),
    tap((level) => {
      console.log({ level });
    })
  )
  .subscribe(() => {
    sprite.classList.add('powerup');
  });

We start his level at 1 and increase it by 1 every time the keydown event fires.

And the tap operator operator lets us quickly log the value without disturbing the pipeline.

My power infinitely approaches MAXIMUM!

Going Super Saiyan

We’ve trained hard, it’s time to transform.

The scan operator tracks Goku’s power level. Now we need to go SSJ when it emits 100.

I built a map of levels: transformations. You can put it right above main.

const powerLevels = {
  100: {
    current: 'base',
    next: 'ssj'
  }
};

const main = () => {
  // ...
};

It’s overkill, but should simplify adding future transformations.

When the power level reaches a number in that powerLevels map, we’ll remove its current class from sprite and add the next class.

This lets us smoothly go from one transformation to the next.

Here’s the code.

const { fromEvent } = rxjs;
const { filter, map, scan, tap } = rxjs.operators;

const begin = fromEvent(document, 'keydown');
const end = fromEvent(document, 'keyup');

begin
  .pipe(
    scan((level) => level + 1, 1),
    tap((level) => {
      console.log({ level });
      sprite.classList.add('powerup');
    }),
    map((level) => powerLevels[level]),
    filter((level) => level && level.next)
  )
  .subscribe(({ current, next }) => {
    sprite.classList.remove(current);
    sprite.classList.add(next);
  });

Map and Filter

Adding the powerup class now happens inside of tap, because it should always happen. The SSJ transformation however, shouldn’t always happen.

Using map, the latest power level becomes an entry in the powerLevels map. We use filter to check if the entry exists and has a .next property.

If it does, that means Goku can go even further beyond! Our .subscribe will swap current and next as class names on sprite.

The end result?

Power Meter

You’re having as much fun as I am, right? Unfortunately, our user won’t.

They can’t see how high Goku’s power level is! They won’t know how to open the DevTools console. We must remedy this!

Let’s improve our UX by filling the power meter. You can put this above main.

const fillMeter = (level) => {
  const limit = 100;

  if (level >= limit) {
    return;
  }

  const containerWidth = meterContainer.offsetWidth;
  const newWidth = (level / limit) * containerWidth;

  meter.style.width = `${newWidth}px`;
};

And call it inside tap.

tap((level) => {
  console.log({ level });
  sprite.classList.add('powerup');
  fillMeter(level);
});

And here we go:

Going Even Further Beyond

Unlocking more transformations is just a matter of adding sprites, and updating our powerLevels map. If you’re interested, submit a PR on the repo and we’ll definitely talk.

Here’s the original sprite sheet. Enjoy!

In my previous article, How to think reactively and animate moving objects using RxJs, I described how to build a MobileObject class that simulates the movement of an object subject to accelerations imposed on it by an external controller.

Now I want to show you a simple distributed system that allows a Controller app to remotely control the movement of a MobileObject. A second remote app, the Monitor, shows the movement of the object on a two-dimensional plan. At the center of the system lays a MobileObjectServer, which is the place where the MobileObjects live.

The goal of this article is to explain how Reactive thinking can progressively produce a design which maps the requirements very naturally and produces a neat solution. We will end up solving the problem subscribing to just ONE Observable.

We’ll focus on the server part, which is the most intriguing from this standpoint.

For the implementation, we’ll use RxJs and TypeScript. The server runs on Node. All the components communicate using Web-Sockets.

The full code base, comprised of the Server Controller and Monitor, can be found here.

Schema of the distributed system

The logical schema of the distributed system is represented in the following diagram:

Schema of the distributed system

At the center lays the MobileObjectServer where the instances of the MobileObjets run. Each MobileObject is controlled by its Controller, that is a Web app through which we can issue commands (like accelerate, brake) to the MobileObject. The movement of all MobileObjects can be seen on one or more Monitors. Each Monitor is again a Web app.

The following diagram shows a sample interaction flow between one Controller, one Monitor, and the MobileObjectServer.

Sample interaction flow

The Server Requirements in terms of events

We can express the requirements for the server part of our distributed system in terms of events:

• Event1 — when a Controller connects => create a MobileObject
• Event2 — when a Controller receives a command => forward the command to the MobileObject controlled by the Controller
• Event3 — when a Controller disconnects => delete the MobileObject controlled by the Controller
• Event4 — when a Monitor connects => start sending dynamics data of all running MobileObjects to the newly connected Monitor
• Event5 — when a MobileObject is added => start sending its dynamics data to all the Monitors connected
• Event6 — when a Monitor disconnects => stop sending the streams of dynamics data for all MobileObjects to that Monitor

Reactive thinking will produce a design which naturally maps the requirements expressed in this way.

The elements composing the server

The server component of the distributed application is made up of two main elements:

• the MobileObject class, which implements the dynamic movement logic using RxJs Observables — this has been described in detail here
• the MobileObjectServer class, which manages the web-socket protocol, receiving commands from the Controller and sending out to the Monitors all information about the dynamics of MobileObject. This implementation has been inspired by this article from Luis Aviles.

MobileObject APIs

Let’s have a brief overview of the MobileObject class — all details can be found here while the code can be found in this repository.

The MobileObject offers two families of APIs.

The first one is the set of methods through which an external Controller can issue commands that affect the dynamics of the object (for example, accelerate, brake).

The second are streams of readonly data which communicate to external clients, the Monitors, the relevant data about the dynamic behaviour of the object (that is, its position and velocity over time).

APIs of MobileObject

In order to move an instance of a MobileObject, a Controller has to turn it on (with the turnOn() method), apply the desired acceleration (with the methods accelerateX(acc: number) and accelerateY(acc: number)), and then maybe brake (with the method brake()).

When a Monitor connects to the MobileObjectServer, the MobileObjectServer subscribes to the dynamicsObs and the observable of the MobileObjects running in the server. It then starts sending the data related to their movement to the connected Monitors.

For the purpose of this article, this is all you need to know about the MobileObject.

Sockets as Observables

The MobileObjectServer starts doing something when a client, either a Controller or a Monitor, opens a websocket connection. Over the course of time, the MobileObjectServer can receive many requests to open a connection from many clients.

Sequence of sockets opened over time

This looks like an Observable of sockets. This is how to obtain it using the socket.io library:

import { Server } from 'http';

import { Observable } from 'rxjs';
import { Observer } from 'rxjs';

import * as socketIoServer from 'socket.io';

import {SocketObs} from './socket-obs';

export function sockets(httpServer: Server, port) {
    httpServer.listen(port, () => {
        console.log('Running server on port %s', port);
    });
    return new Observable<SocketObs>(
        (subscriber: Observer<SocketObs>) => {
            socketIoServer(httpServer).on('connect', 
                socket => {
                    console.log('client connected');
                    subscriber.next(new SocketObs(socket));
                }
            );
        }
    );
}

Via the function sockets, we create an Observable of SocketObs (we will see the implementation of this class later). Any time the websocket server receives a connect request and creates a new socket, the Observable returned by this function emits an instance of SocketObs which wraps the socket just created.

Messages over sockets as Observables

Sockets can be used to send messages from the client to the server and vice versa. With the socket.io library, we can send messages using the emit method.

SocketIO.Socket.emit(event: string, …args: any[]): SocketIO.Socket

The parameter event can be seen as an identifier of the type of message we want to send. The …args parameters can be used to send data specific to a single message.

Whoever is interested in a certain type of message (or event, to use the socket.io terminology) can start listening on the socket using the method on.

SocketIO.Emitter.on(event: string, fn: Function): SocketIO.Emitter

Sequence of messages received over time

Again, the sequences of messages received by the Receiver look like Observables. This is how we can create Observables that actually emit any time a message of a certain type is received.

The onMessageType method is the one that does the trick. It returns an Observable, which emits any time a message of type messageType is received.

import { Observable, Observer } from 'rxjs';

export class SocketObs {
    constructor(private socket: SocketIO.Socket) {}

    onMessageType(messageType): Observable<any> {
        return new Observable<any>((observer: Observer<any>) => {
            this.socket.on(messageType, data => observer.next(data));
        });
    }
}

In this way, sockets events, or messages as we call them here, have been transformed into Observables. These are going to be the foundations of our design.

Determine the nature of the Client

There are two types of clients which can connect with the MobileObjectServer. One is the Controller and one is the Monitor. The MobileObjectServer first needs to determine which type of client it is going to deal with on a specific socket.

The way we have chosen to implement such logic is to have the Controller and the Monitor send different message types as their first message.

• Controller sends a message of type BIND_CONTROLLER
• Monitor sends a message of type BIND_MONITOR

Depending on the type of the first message received on a socket, the MobileObjectServer is able to identify whether it is communicating with a Controller or a Monitor.

How to determine the nature of the client on a socket

As soon as a socket is created, the MobileObjectServer has to start listening to both types of messages, BIND_CONTROLLER and BIND_MONITOR. The first to occur will win. It is a race between the two Observables which map the two different types of messages.

Such logic has to be repeated any time a new socket is created, that is any time the Observable returned by the function [sockets](https://gist.github.com/EnricoPicci/35f3c3a2a2a3f96cfdf7b89d46a5d499) emits. Therefore, we need to merge all the events that win the race. We need to use the mergeMap operator, which merges all the events raised by the Observables involved, and flatten the results into a new Observable (mergeMap was formerly know as flatMap).

Observable that emits specific events depending on the nature of the Client

The code to obtain this result is the following:

startSocketServer(httpServer: Server) {
    sockets(httpServer, this.port).pipe(
        mergeMap(socket =>
            race(
                socket.onMessageType(MessageType.BIND_MONITOR),
                socket.onMessageType(MessageType.BIND_CONTROLLER)
            )
        )
    )
    .subscribe();
}

Now that we know how to differentiate Controllers and Monitors, we can focus on what to do in these two cases.

Events relevant for a Monitor

A Monitor shows the movement of all MobileObjects which are running on the MobileObjectServer. So the MobileObjectServer has to send the right information to the monitors at the right times. Let’s see first what those times are, that is which are the relevant events that the MobileObjectServer has to be aware of in order to fulfill its job.

Adding and removing MobileObjects

The first relevant events are:

• a MobileObject has been added => the MobileObject is shown on the Monitor
• a MobileObject has been removed => the MobileObject is removed from the Monitor

MobileObjects are added or removed over time, so such events can be modeled with two Observables:

• an Observable which emits when a MobileObject is added
• an Observable which emits when a MobileObject is removed

Once a Monitor is connected, the MobileObjectServer starts being interested in both of those Observables, so it has to merge them:

Merging MobileObject added and removed events for a Monitor

Similar to what we have seen before, we need to repeat such logic any time a Monitor is added. Therefore we need to mergeMap all the Observables which are the result of the merge of the ‘mobile object added’ Observable with the ‘mobile object removed’ Observable.

Events for MobileObject added and removed for all Monitors

This is the code to obtain an Observable which emits any time a MobileObject has to be added to or removed from every Monitor:

import {sockets} from './socket-io-observable';
import {SocketObs} from './socket-obs';

class MobileObjectServer {
    private mobileObjectAdded = new Subject<{mobObj: MobileObject, mobObjId: string}>();
    private mobileObjectRemoved = new Subject<string>();

    startSocketServer(httpServer: Server) {
        sockets(httpServer, this.port).pipe(
            mergeMap(socket =>
                race(
                    socket.onMessageType(MessageType.BIND_MONITOR)
                    .pipe(
                        map(() => (socketObs: SocketObs) => this.handleMonitorObs(socketObs))
                    ),
                    socket.onMessageType(MessageType.BIND_CONTROLLER)
                    // something will be added here soon to make this logic work
                )
                .pipe(
                    mergeMap(handler => handler(socket))
                )
            )
        )
        .subscribe();
    }

    handleMonitorObs(socket: SocketObs) {
        const mobObjAdded = this.mobileObjectAdded;
        const mobObjRemoved = this.mobileObjectRemoved;
        return merge(mobObjAdded, mobObjRemoved);
    }
}

We have introduced a few things with this code which are worth commenting on here.

We have created the MobileObjectServer class, which will be the place where we will code all our server logic from now on.

The method handleMonitorsObs, which we are going to enrich later on, returns simply the merge of two Observables, mobileObjectAdded and mobileObjectRemoved, which are Subjects. This is the “inner” merge shown in the picture above.

Subjects are Observables, and therefore can be merged as we do here. But Subjects are also Observers, so we can emit events through them. As we will see later in the code, there will be a time when we will use these Subjects to emit the events their names suggest.

The last point is related to the code we have added in the startSocketServer method:

race(
   socket.onMessageType(MessageType.BIND_MONITOR)
   .pipe(
      map(() => (sObs: SocketObs) => this.handleMonitorObs(sObs))
   ),
   socket.onMessageType(MessageType.BIND_CONTROLLER)
   // something will be added here soon to make this logic work
)
.pipe(
   mergeMap(handler => handler(socket))
)

This is basically a way to say: any time a BIND_MONITOR message is received, return the function

(socketObs: SocketObs) => this.handleMonitorObs(socketObs)

which will be executed within the mergeMap operator piped into the result of the race function. This mergeMap operator is the external mergeMap shown in the picture above.

Another way to read the code is the following: any event corresponding to a message of type BIND_MONITOR gets transformed by the logic of

mergeMap(() => this.handleMonitorObs(socket))

where socket is the instance of type SocketsObs emitted by the race function.

Soon we will add something similar for the BIND_CONTROLLER case to make this whole logic work.

Handle MobileObject dynamics Observables

Let’s consider one Monitor which connects to the MobileObjectServer. After the connection, a couple of MobileObjects are added to the MobileObjectServer.

MobileObjects added for one Monitor

Now for each MobileObject, we have to start considering the dynamics Observables they offer as part of their APIs. These Observables emit, at regular intervals of time, data about the dynamics (position and velocity) of the MobileObject. If mobileObject stores a reference to a MobileObject, we can obtain its dynamics Observable via mobileObject.dynamicsObs (see MobileObject APIs).

First we have to transform each event representing the fact that a MobileObject has been added into the series of events emitted by its dynamicsObs. Then we mergeMap all these series into a new single Observable which emits all dynamic events for all MobileObjects which are added.

Dynamics events for a single Monitor

Then we apply all this jazz to all the Monitors which connect to the MobileObjectServer. So we end up with a new Observable which emits dynamics data for all Monitors and all MobileObjects (plus all events related to the fact that a MobileObject has been removed).

Relevant events for all Monitors

Per each time interval, we have groups of four events related to the emission of data about the dynamics of our MobileObjects. Why? This makes sense if we think that we have two Monitors and two MobileObjects. Each MobileObject has to send its dynamics data to each Monitor per every time interval. Therefore it is correct to see four events per each time interval.

Once this is clear, the code is very simple:

import {sockets} from './socket-io-observable';
import {SocketObs} from './socket-obs';

class MobileObjectServer {
    private mobileObjectAdded = new Subject<{mobObj: MobileObject, mobObjId: string}>();
    private mobileObjectRemoved = new Subject<string>();


    startSocketServer(httpServer: Server) {
        sockets(httpServer, this.port).pipe(
            mergeMap(socket =>
                race(
                    socket.onMessageType(MessageType.BIND_MONITOR)
                    .pipe(
                        map(() => (socketObs: SocketObs) => this.handleMonitorObs(socketObs))
                    ),
                    socket.onMessageType(MessageType.BIND_CONTROLLER)
                    // something will be added here soon to make this logic work
                )
                .pipe(
                    mergeMap(handler => handler(socket))
                )
            )
        )
        .subscribe();
    }

    handleMonitorObs(socket: SocketObs) {
        const mobObjAdded = this.mobileObjectAdded
                              .pipe(
                                mergeMap(data => data.mobileObject.dynamicsObs)
                              );
        const mobObjRemoved = this.mobileObjectRemoved;
        return merge(mobObjAdded, mobObjRemoved);
    }

}

We have just introduced one simple change. We changed the handleMonitorObs method to add the mergeMap operator. This transforms the mobileObjectAdded Observable so that the new Observable emits the dynamics data we are looking for.

The rest has remained untouched.

Summary so far

What have we done so far? We have just transformed Observables to obtain new Observables which emit all the events MobileObjectServer is interested in when it has to deal with a Monitor. Nothing else.

Transformations of Observables relevant for Monitors

You can see how these transformations are reflected in the code in the following image:

The only thing we need to do now is to add the desired side effects to the relevant events. This will eventually allow us to achieve what we want, that is to communicate to the Monitor the right information at the right time.

But before moving to side effects, let’s cover what MobileObjectServer needs to do when interacting with a Controller, the other client in our distributed system.

Events relevant for a Controller

When a Controller connects to the MobileObjectServer there are fewer things that the server needs to care about. At least there are fewer nested relevant events happening.

The things that the MobileObjectServer needs to care about are:

• A Controller has connected, which in our simple logic means that we have to create a brand new MobileObject
• The Controller has sent commands for its MobileObject
• The Controller has disconnected. In our implementation, this means that we somehow have to delete the MobileObject controlled by the Controller (we have a 1 to 1 relationship between MobileObject and its Controller)

We already know the first event: it is the one emitted by the Observable returned by socket.onMessageType(BIND_CONTROLLER).

Commands are sent by the Controller to the MobileObjectServer in the form of messages. So we can create an Observable of commands received over a certain _socket (_received from a certain Controller) since each Controller has its own socket. We do this by simply using the onMessageType method of SocketObs

socket.onMessageType(CONTROLLER_COMMAND)

SocketObs also offers a method, onDisconnect, which returns an Observable that emits when the socket is disconnected. This is what we need in order to deal with the third event.

Events relevant when a Controller connects to MobileObjectServer

Since we are dealing with more than one Controller potentially connecting to the MobileObjectServer, it should not surprise you to learn that we need to mergeMap the result of the merge. This is the same type of transformation we have already done a few times.

The code should be no surprise as well.

startSocketServer(httpServer: Server) {
        sockets(httpServer, this.port).pipe(
            mergeMap(socket =>
                race(
                    socket.onMessageType(MessageType.BIND_MONITOR)
                    .pipe(
                        map(() => (socketObs: SocketObs) => this.handleMonitorObs(socketObs))
                    ),
                    socket.onMessageType(MessageType.BIND_CONTROLLER)
                    .pipe(
                        map(() => (socketObs: SocketObs) => this.handleControllerObs(socketObs))
                    ),
                )
                .pipe(
                    mergeMap(handler => handler(socket))
                )
            )
        )
        .subscribe();
}

handleMonitorObs(socket: SocketObs) {
        const mobObjAdded = this.mobileObjectAdded
                              .pipe(
                                mergeMap(data => data.mobileObject.dynamicsObs)
                              );
        const mobObjRemoved = this.mobileObjectRemoved;
        return merge(mobObjAdded, mobObjRemoved);
}

handleControllerObs(socket: SocketObs) {
        const commands = socket.onMessageType(MessageType.CONTROLLER_COMMAND);
        const disconnect = socket.onDisconnect();

        return merge(commands, disconnect);
}

We have simply added an handleControllerObs method that deals with commands received and the disconnect of a Controller. We apply the mergeMap transformation to it as we have already done with handleMonitorObs.

Summary of the transformations applied to Controllers

The following diagram illustrates all transformations we have applied starting from the Observable that emits when a Controller connects.

Transformations of event streams (Observables) relevant for Controllers

The Final Observable

If we put together the transformations we have done for both the Monitors and the Controllers, what we obtain is the following final Observable.

The tree of events produced subscribing the Final Observable

Just by subscribing to this one final Observable, the whole tree of events gets unfolded.

Side effects

The beautiful tree of events we have created by subscribing to the Final Observable does not do anything. But it does a good job of mapping the Events we identified while describing the requirements of the Server at the beginning of this article.

Basically it tells us clearly when we have to do something.

This something is what we call a side effect.

Side effects

When a Controller connects and disconnects, we respectively create or delete a MobileObject. As side effect of these actions is that we raise “MobileObject added” and “MobileObject deleted” events using the mobileObjectAdded and mobileObjectRemoved Subjects we introduces some paragraphs ago.

How to implement side effects

In RxJs there are different ways to implement side effects.

Observers is one. We can add Observers while we subscribe using the tap operator (formerly know as do).

Another way is to inject them in any function we pass to any RxJs operator.

We are mainly going to use tap, since it allows us to place side effects throughout the entire tree of events. But we are also going to place side effects directly inside functions we pass to RxJs operators.

The only place we do not put side effects is subscribe. The reason is that, given how we built it, the Final Observer emits many different types of events. Therefore subscribe, which works the same for all events, is not the right place to put behavior which depends on certain types of events.

Hopefully at this point the code sort of speaks for itself.

Implementation of Side Effects in the code

Last but not least: completion of Observables

There is one thing that we still need to do to complete our design: stop the streams of events, or complete the Observables, when either a Controller or a Monitor disconnects.

When a Controller disconnects

When a Controller disconnects, we delete the MobileObject it controls. As part of the deletion, it is important to make sure that the MobileObjectServer stops sending dynamics data related to this MobileObject to the connected Monitors. This means that we must complete the following Observable:

mobObjInfo.mobObj.dynamicsObs
.pipe(
  tap(dynamics => socket.send(MessageType.DYNAMICS_INFO, dynamics)),
)

We can easily achieve this just using the takeUntil operator together with the mobileObjectRemoved Observable we already know:

mobObjInfo.mobObj.dynamicsObs
.pipe(
  tap(dynamics => socket.send(MessageType.DYNAMICS_INFO, dynamics)),
  takeUntil(this.mobileObjectRemoved.pipe(
    filter(id => id === mobObjInfo.mobObjId)
  ))
)

takeUntil ensures that an Observable completes when the Observable passed as a parameter to takeUntil emits.

mobileObjectRemoved emits every time a MobileObject is removed. What we want, though, is to stop sending dynamics info when a specific MobileObject, identified by its id, is removed. So we add the filter logic.

When a Monitor disconnects

In this case, we can also use takeUntil.

We know when a Monitor disconnects because the socket, of type SocketObs, associated to it emits via the socket.onDisconnect() Observable. So what we need to do is stop sending dynamics info when socket.onDisconnect() emits.

So the final logic to govern the completion of the Observable is

mobObjInfo.mobObj.dynamicsObs
.pipe(
  tap(dynamics => socket.send(MessageType.DYNAMICS_INFO, dynamics)),
  takeUntil(this.stopSendDynamics(socket, mobObjInfo.mobObjId))
)

where

private stopSendDynamics(socket: SocketObs, mobObjId: string){
  return merge(
            this.mobileObjectRemoved.pipe(
                                       filter(id => id === mobObjId)
                                     ),
            socket.onDisconnect()
  );
}

And this is how the core of the code implementing our logic looks:

import {sockets} from './socket-io-observable';
import {SocketObs} from './socket-obs';

class MobileObjectServer {
    private mobileObjectAdded = new Subject<{mobObj: MobileObject, mobObjId: string}>();
    private mobileObjectRemoved = new Subject<string>();


        public startSocketServer(httpServer: Server) {
        sockets(httpServer, this.port).pipe(
            mergeMap(socket =>
                race(
                    socket.onMessageType(MessageType.BIND_MONITOR)
                    .pipe(
                        map(() => (socketObs: SocketObs) => this.handleMonitorObs(socketObs))
                    ),
                    socket.onMessageType(MessageType.BIND_CONTROLLER)
                    .pipe(
                        map(() => (socketObs: SocketObs) => this.handleControllerObs(socketObs))
                    ),
                )
                .pipe(
                    mergeMap(handler => handler(socket)) 
                )
            )
        )
        .subscribe();
    }


    private handleMonitorObs(socket: SocketObs) {
        const mobObjAdded = this.mobileObjectAdded
                                .pipe(
                                    tap(mobObjInfo => socket.send(MessageType.MOBILE_OBJECT, mobObjInfo.mobObjId)),
                                    mergeMap(mobObjInfo => mobObjInfo.mobObj.dynamicsObs
                                                    .pipe(
                                                        tap(dynamics => socket.send(MessageType.DYNAMICS_INFO, dynamics)),
                                                        takeUntil(this.stopSendDynamicsInfo(socket, mobObjInfo.mobObjId))
                                                    )
                                    )
                                );
        const mobObjRemoved = this.mobileObjectRemoved
                                .pipe(
                                    tap(mobObjId => socket.send(MessageType.MOBILE_OBJECT_REMOVED, mobObjId)),
                                );
        return merge(mobObjAdded, mobObjRemoved);
    }

    private handleControllerObs(socket: SocketObs) {
        const {mobObj, mobObjId} = this.newMobileObject();

        this.mobileObjectAdded.next({mobObj, mobObjId});

        const commands = socket.onMessageType(MessageType.CONTROLLER_COMMAND)
                        .pipe(
                            tap(command  => this.execute(command, mobObj))
                        );

        const disconnect = socket.onDisconnect()
                        .pipe(
                            tap(() => this.mobileObjectRemoved.next(mobObjId)),
                        );

        return merge(commands, disconnect);
    }

    private stopSendDynamicsInfo(socket: SocketObs, mobObjId: string) {
        return merge(this.mobileObjectRemoved.pipe(filter(id => id === mobObjId)), socket.onDisconnect());
    }

}

Conclusion

It has been a pretty long journey. We have seen some reasoning driven by Reactive Thinking and some implementations of this reasoning.

We started transforming WebSockets events into Observables. Then, applying incremental transformations, we ended up creating a single Observable that, once subscribed, unfolds all the events we are interested in.

At this point, adding the side effects that allow us to achieve our goal has been straightforward.

This mental process of design, which is incremental in itself, is the meaning I give to “Reactive Thinking”.

The full code base, comprising Server Controller and Monitor, can be found here.

Dealing with asynchronous non-blocking processing has always been the norm in the JavaScript world, and now is becoming very popular in many other contexts. The benefits are clear: an efficient use of resources. But the benefits come at a cost: a non-trivial increase in complexity.

Over time, vendors and the open source community have tried to find ways to reduce such complexity without compromising the benefits.

Asynchronous processing started with ‘callbacks’, then came Promise and Future, async and await. Recently another kid has come to town — ReactiveX with its various language implementations — bringing the developers a new powerful tool, the Observable.

In this article, we want to show how Observables implemented by RxJs (the JavaScript embodiment of ReactiveX) can simplify code to be executed with Node.js, the popular server-side JavaScript non-blocking environment.

A simple use case — Read, Transform, Write, and Log

To make our reasoning concrete, let’s start from a simple use case. Let’s assume we need to read the files contained in **Source Dir**, transform their content and write the new transformed files in a **Target Dir**, while keeping a log of the files we have created.

Read — Transform — Write — Log

Synchronous implementation

The synchronous implementation of this use case is pretty straightforward. In a sort of pseudo code representation, we could think of something like:

read the names of the files of Source Dir
   for each file name
      read the file
      transform the content
      write the new file in Target Dir
      log the name of the new file
   end for
console.log('I am done')

There is nothing special to comment here. We can just say that we are sure of the sequence of execution of each line and that we are sure that things will happen as described by the following flow of events. Each circle corresponds to the completion of an I/O operation.

The sequence of events in a synchronous world

What happens in an asynchronous non-blocking environment like Node.js

Node.js is an asynchronous non-blocking execution environment for JavaScript. Non-blocking means that Node.js does not wait for I/O or Network operations to complete before moving to the execution of the next line of code.

Processing one file

Reading and writing files are I/O operations where Node.js shows its non-blocking nature. If a Node.js program asks for a file to read, it has to provide a function to be executed when the file content is available (the so called callback) and then immediately move on to the next operation to execute.

Let’s consider the case of just one file. Reading, transforming, writing one file and updating the log in Node.js looks something like this:

import * as fs from 'fs'; // Node module to access file system
const fileName = 'one-file.txt';
fs.readFile(fileName, callback(err, data) => {
   const newContent = transform(data);
   const newFileName = newFileName(fileName); // calculate new name
   fs.writeFile(newFileName, newContent, err => {
      if(err) {// handle error};
      fs.appendFile('log.txt', newFileName  + ' written', err = {
         if (err) {// handle error}
      });
   });
})

The syntax may look a bit convoluted with 2 levels of indentation, but if we think of what happens in terms of events we can still precisely foresee the sequence:

Sequence of events in Node while transforming 1 file

The paradise of Promise

This is the use case where JavaScript Promise shines. Using Promise we can make the code look again sequential, without interfering with the asynchronous nature of Node.js.

Assuming we can access functions that perform read and write operations on file and return a Promise, then our code would look like:

const fileName = 'my-file.txt';
readFilePromise(fileName)
.then(data => {
   const newContent = transform(data);
   const newFileName = newFileName(fileName); // build the new name
   return writeFilePromise(newFileName, newContent)
})
.then(newFileName => appendFile('log.txt', newFileName))
.then(newFileName => console.log(newFileName + ' written'))
.catch(err => // handle error)

There are several ways to transform Node.js functions in Promise based functions. This is one example:

function readFilePromise(fileName: string): Promise<Buffer>{
   return new Promise(function(resolve, reject) {
      fs.readFile(fileName, function(err, data: Buffer) {
         if(err !== null) return reject(err);
         resolve(data);
      });
   });
}

Processing many files

If we return to the original use case, where we have to transform all the files contained in a Directory, the complexity increases and Promises start showing some limits.

Let’s look at the events the Node.js implementation needs to manage:

Sequence of events when transforming many files in parallel

Each circle represents the completion of one I/O operation, either read or write. Every line represents the processing of one specific file, or a chain of Promises.

Given the non-blocking nature of Node.js, there is no certainty on the sequence in time of such events. It is possible that we will finish writing **File2** before we finish reading **File3**.

The parallel processing of each file makes the use of Promises more complex (at the end of this article, a Promise based implementation is provided). This is the scenario where ReactiveX — RxJs in particular — and Observable shine and allow you to build elegant solutions.

What are Observables and what can you do with them?

There are many places where formal definitions of Observables are detailed, starting from the official site of ReactiveX.

Here I just want to remind you of a couple of properties that have always gotten my attention:

• Observable models a stream of events
• Observable is the “push” brother of Iterable, which is “pull”

As the “push” brother of Iterable, Observable offers developers many of the cool features provided by Iterables such as:

• Transform “streams of events” or Observables, via operators such as map, filterand skip
• Apply functional programming style

One additional very important thing that Observable offers is subscription. Via subscription, the code can apply “side effects” to events and perform specific actions when specific events happen, such as when errors occur or the stream of events completes.

The subscribe interface of Observable

As you can see, the Observable interface gives developers the possibility to provide three different functions which define what to do respectively when: an event is emitted with its data, an error occurs, or the stream of events completes.

I guess all of the above may sound very theoretical to those who have not yet played with Observable, but hopefully the next part of the discussion, which is focused on our use case, will make these concepts more concrete.

Implementation of the Read, Transform, Write, and Log use case via Observable

Our use case starts with reading the list of files contained in **Source Dir**. So, let’s start from there.

Read all the file names contained in a Directory

Let’s assume we have access to a function which receives as input the name of a directory and returns an Observable which emits the list of file names of the directory once the directory tree structure has been read.

readDirObservable(dirName: string) : Observable<Array<string>>

We can subscribe to this Observable and when all file names have been read, start doing something with them:

Subscription to an Observable which emits when the directory has been read

Read a list of files

Let’s assume now that we can access a function which receives as input a list of file names and emits each time a file has been read (it emits the content of the file Buffer, and its name string).

readFilesObservable(fileList: Array<string>) 
   : Observable<{content: Buffer, fileName: string}>

We can subscribe to such Observable and start doing something with the content of the files.

Combining Observables — switchMap operator

We have now two Observables, one that emits a list of file names when the directory has been read and one that emits every time a file is read.

We need to combine them to implement the first step of our use case, which is: when readDirObservable emits, we need to switch to readFilesObservable .

switchMap operator

The trick here is performed by the switchMap operator. The code looks like:

readDirObservable(dirName)
.switchMap(fileList => readFilesObservable(fileList))
.subscribe(
      data => console.log(data.fileName + ‘ read’), // do stuff with the data received
      err => { // manage error },
      () => console.log(‘All files read’)
)

We must mention that the switchMap operator is more powerful than this. Its full power though can not be appreciated in this simple use case, and its full description is outside the scope of this post. If you are interested, this is an excellent article that describes in detail switchMap.

Observable generating a stream of Observables

We have now a stream of events representing the completion of a read operation. After the read we need to do a transformation of the content that, for sake of simplicity, we assume to be synchronous, and then we need to save the transformed content in a new file.

But writing a new file is again an I/O operation, or a non-blocking operation. So every ‘file-read-completion’ event starts a new path of elaboration that receives as input the content and the name of the source file, and emits when the new file is written in the Target Dir (the event emitted carries the name of the file written).

Again, we assume that we’re able to access a function that emits as soon as the write operation is completed, and the data emitted is the name of the file written.

writeFileObservable(fileName: string, content: Buffer) :            Observable<string>

In this case, we have different “write-file” Observables, returned by the writeFileObservable function, which emits independently. It would be nice to merge them into a new Observable which emits any time each of these “write-file” Observables emit.

A stream of Observables we would like to merge

With ReactiveX (or RxJs in JavaScript) we can reach this result using the mergeMap operator (also known as a flatMap). This is what the code looks like:

readDirObservable(dir)
.switchMap(fileList => readFilesObservable(fileList))
.map(data => transform(data.fileName, data.content))
.mergeMap(data => writeFileObservable(data.fileName, data.content))
.subscribe(
      file => console.log(data.fileName + ‘ written’),
      err => { // manage error },
      () => console.log(‘All files written’)
)

The mergeMap operator has created a new Observable, the writeFileObservable as illustrated in the following diagram:

Observable returned by mergeMap operator

So what?

Applying the same approach, if we just imagine that we have a new function of writeLogObservable, that writes a line on the log as soon as the file is written and emits the file name as soon as the log is updated, the final code for our use case would look like:

readDirObservable(dir)
.switchMap(fileList => readFilesObservable(fileList))
.map(data => transform(data.fileName, data.content))
.mergeMap(data => writeFileObservable(data.fileName, data.content))
.mergeMap(fileName => writeLogObservable(fileName))
.subscribe(
      file => console.log(fileName + ‘ logged’),
      err => { // manage error },
      () => console.log(‘All files have been transformed’)
)

We do not have indentations introduced by the callbacks.

Time flows along the vertical axis only, so we can read the code line by line and reason about what is happening line after line.

We have adopted a functional style.

In other words, we have seen the benefits of Observable in action.

Create Observable from functions with callbacks

I hope you now think that this looks pretty cool. But even in this case you may have one question. All the functions that make this code cool just do not exist. There is no readFilesObservable or writeFileObservable in standard Node.js libraries. How can we create them?

bindCallback and bindNodeCallback

A couple of functions provided by Observable, namely bindCallback(and bindNodeCallback) come to our rescue.

The core idea behind them is to provide a mechanism to transform a function f which accepts a callback cB(cBInput) as input parameter into a function which returns an Observable obsBound which emits cBInput. In other words, it transforms the invocation of the cB in the emission of cBInput.

The subscriber of obsBound can define the function which will process cBInput (which plays the same role as cB(cBInput)). The convention applied is that the callback function cB(cBInput) must be the last argument off.

It is probably easier to understand the mechanism looking at the following diagram:

From a function to an Observable

The starting point, the function f(x, cb) is the same in the two cases. The result (what is printed on the console) is the same in the two cases.

What is different is how the result is obtained. In the first case the result is determined by the callback function passed as input. In the second case it is determined by the function defined by the subscriber.

Another way of considering how bindCallback works is to look at the transformation it performs, as illustrated in the diagram below.

Transformation performed by bindCallback

The first argument of f becomes the value passed to the new function fBound. The arguments used as parameters of the callback cb become the values emitted by the new Observable returned by fBound.

bindNodeCallback is a variation of bindCallback based on the convention that the callback function has an error parameter as the first parameter, along with the Node.js convention fs.readFile(err, cb).

Create Observables from non-callback functions

bindNodeCallback has been designed to work with functions which expect a callback as the last argument of their input, but we can make it work also with other functions.

Let’s consider the standard Node.js function readLine. This is a function used to read files line by line. The following example shows as it works:

readLine function

Each line read is pushed into the lines array. When the file is completely read, the function processLinesCb is called.

Imagine now that we define a new function,_readLines, which wraps the logic defined above as shown by the following snippet:

readLine function wrapped by a callback function

Once all lines are read, they are processed by the function processLinesCb, which is the last input parameter of _readLines. _readLines is therefore a function that can be treated by bindCallback. Through this trick we can transform the Node.js function fs.readLine into an Observable using the usual bindCallback function as follows:

readLine as Observable

Conclusion

Asynchronous non-blocking processing is complex by nature. Our minds are used to think sequentially — this is true at least for those of us who started coding few years ago. We often find it challenging to reason about what is really happening in these environments. The callback-hell is just around the corner.

Promises and Futures have simplified some of the most frequent cases such as ‘one time’ asynchronous events, the ‘request now — respond later’ scenario typical of HTTP requests.

If we move from ‘one time’ events to ‘event streams’ Promises start showing some limitations. In such cases we may find ReactiveX and Observables a very powerful tool.

As promised: the Promise-based implementation of our use case

This is an implementation of the same use case based on Promises:

const promises = new Array<Promise>();
readDirPromise(dir)
.then(fileList => {
   for (const file of fileList) {promises.push(
         readFilePromise(file)
         .then(file_content => transform(file_content))
         .then(file => writeLogPromise(file))
      );
   }
   return promises;
}
.then(promises => Promise.all(promises))
.then(() => console.log(‘I am done’))
.catch(err => { // manage error })