Observables and observers in RxJS

Observables and observers are probably the first things you think of when someone mentions RxJS. They are the foundation of the library and the idea behind its functionalities.

What are Observable and Observer?

In short – an observable is something producing data, while the observer is interested in that data. If you want to receive data from an observable, you call its subscribe method. You can think of the observable as the source of the data and if you are interested in receiving it, you just subscribe to it.

The subscribe function is receiving as a parameter the observer, this is the function interested in the data, and also responsible for reading it:

const simpleObservable$ = new Observable().subscribe(
  (data => console.log(data)),
  (error => console.log(error)),
  (() => console.log("Complete"))
);

Everyone in the application, who wants to receive this data, is passed as a parameter to the subscribe method. This is the observer, and you can have as many observers as you want. An observer is an object implementing a simple interface, with just 3 methods next(), error(), and complete(). The observable is calling these methods depending on the current state of the data. If there is a new value produced, it will call the next() method, if there is an error – the error() method, and when there are no more values to produce, it will finally call complete().

Creating observables

Create from scratch with the observable constructor

Let’s suppose you want to create an observable that will produce a stream of data, containing all of the courses from an array. Let’s keep this data hardcoded for the simplicity of the example.

We will use the observable constructor and name the variable allCoursesObservable$. The $ sign in the observable variable names is a convention, giving readability to the code what exactly is observable. The observable constructor is accepting one optional parameter called subscribe. It is optional but it is necessary if we want our observable to produce values. It is the Observer:

// data
const allCourses = ["Architecture", "Business management", "Computer science"];

// observer
function subscribe(subscriber: Observer<string>) {
  for (let course of allCourses) {
    subscriber.next(course);
  }
}

// observable
const allCoursesObservable$ = new Observable(subscribe);

Use RxJS helper function to create the observable from another data type

RxJS comes with 3 helper functions for creating observables from existing data – of, from, and fromEvent.

of

The of function is very flexible. You may pass it a comma separated objects, and it will convert them into an observable array:

var mixedObservable$ = of("Architecture", true, "Business management", 14, "Computer science");
mixedObservable$.subscribe((data) => {
  console.log(data);
})

This will result in:

It is important to note that each comma-separated object will be emitted. So if you change the arguments, and pass an array and a number, like this:

var mixedObservable$ = of(["Architecture", true, "Business management", 14, "Computer science"], 19);
mixedObservable$.subscribe((data) => {
  console.log(data);
})

The function will emit two values – the whole array at once (because it is one object), and the number 19 (it is the next comma-separated object). The result will be:

from

The from function is similar to of, but it doesn’t accept a comma-separated values. It works with iteratable objects like arrays, and even with other observables or promises. If you use the above example, with the array, but passed to the from function, like this:

var observableUsingFrom$ = from(["Architecture", true, "Business management", 14, "Computer science"]);
observableUsingFrom$.subscribe(data => {
  console.log(data);
})

The output will be:

As you can see the values are emitted one by one, compared to the of function, where the whole array was emitted at once.

As I already mention you may also use from with another observable:

var data$ = of("Architecture", true, "Business management", 14, "Computer science", 19);
var fromAnotherObservable$ = from(data$);
fromAnotherObservable$.subscribe(data => {
  console.log(data);
})

The output will be the same as the previous example.

An with a promise:

const myPromise = new Promise((resolve, reject) => {
setTimeout(() => {
  resolve("Promise result");
}, 300);
});

var fromPromise$ = from(myPromise);
fromPromise$.subscribe(data => {
console.log(data);
})

And the result is:

fromEvent

DOM events are a great example of an Observable. They are a data stream, that continues over time. A button, for instance, will produce data on every click, and you can use observable to process this data. Every element on the page can produce events over time, and you can think of them as a stream of data. This stream of data will be available as long as the corresponding element is available on the page.

For instance, there is a button that will print a list of movies to the console, when clicked:

<div class="row">
  <div>
    <button id="btn-movies">Get movies</button>
  </div>
</div>

In plain JavaScript, you will probably use the addEventListener function to react to the button click. But this is working only for the button, it is a JavaScript function designed to listen for DOM events on specific elements. A much cleaner API for processing events will be available if you use Observable. You will have a single function for data processing that works with observable, regardless of whether the source is a DOM element, HTTP response, or another function.

To achieve it you can use the fromEvent function. It accepts two parameters – the button element, and the event name.

var moveis = ["Friends", "The Big Bang Theory", "Game Of Thrones"];

let moviesButton = document.getElementById("btn-movies") as HTMLButtonElement;
let moviesObservable$ = fromEvent(moviesButton, "click");

moviesObservable$.subscribe(data => {
  moveis.forEach(movie => {
    console.log(movie);
  });
})

fromEvent returns an observable, and you can easily listen for data, by using the subscribe method. Inside the subscribe method, you place the logic for processing the data. If you change the button with some other source in the future, the logic inside the subscribe will still work.

When the button is clicked, the values are printed to the console, as long as the button is available on the page, and we click on it, the movies will be printed:

Call a function which will return an observable

Larger application usually separates the frontend from the backend logic. The frontend can be written in any framework such as Angular, React, View, or any other. To visualize any data, this framework usually will make HTTP requests to get the data. There are server-side frameworks which can return an Observable, but there are many others which doesn’t. But don’t worry it doesn’t matter what is your server-side logic, because RxJS is providing a function to create an observable from HTTP response. It is call ajax().

It is accepting an endpoint as a parameter, and returns and observable, with the data, returned from that endpoint:

let moviesButton = document.getElementById("btn-movies") as HTMLButtonElement;
let moviesObservable$ = fromEvent(moviesButton, "click");
moviesObservable$.subscribe(() => {
  ajax("/api/movies").subscribe((data: AjaxResponse<any>) => {
    data.response.forEach((movie: string) => {
      console.log(movie);
    });
  })
})

In this example, we are making a server call to the ‘api/movies’ endpoint, then the function ajax is converting the response to an observable. And finally, you can easily subscribe to process the result.