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.