Using Observables to Transform Data

One of the most intriguing, somewhat difficult, but fun parts of JavaScript programming is based around the handling of events. Whenever we issue an asynchronous call, we are essentially waiting for an event to occur that will trigger our callback processing. We generally have no control over how long an asynchronous call might take, especially if this call is to a server somewhere via an API call. In a similar vein, when running a JavaScript application on the browser, we are often waiting for a user to interact with our software and generate an event via a button click, or a keypress. Again, we have no control over how long the user will take to click that button, or indeed the order in which buttons could be clicked. Think of a small child playing with a keyboard. They may hit a group of buttons at the same time by using their fist to smash the keyboard, or they may hit a completely random combination of keys. It is up to our software to handle...

Introduction to Observables

To begin the discussion on Observables, let's first install the RxJS library as follows:

npm install rxjs

The RxJS library already includes the declaration files that are needed by TypeScript, so there is no need to install them separately using @types.

To generate an Observable, we can use the of function as follows:

import { of, Observable } from "rxjs";
const emitter : Observable<number> = of(1, 2, 3, 4);

Here, we start by importing the of function and the Observable type from the rxjs library. We are then defining a constant variable named emitter, which is using generic syntax to define its type as an Observable of type number. We then assign the result of the of function to the emitter variable, which will create an Observable from the numbers 1 through 4. We can now create an Observer as follows:

emitter.subscribe((value: number) => {
    console.log(`value: ${value}`)
});

Here, we are calling...

Observable errors

So what happens when something goes wrong within an Observable stream? Obviously, we will need a mechanism to catch these errors, so that we can do something sensible with them. As an example of a faulty Observable stream, consider the following code:

interface IValue {
    value: number
}
interface INestedObj {
    id?: IValue;
}
const objEmit : Observable<INestedObj> = of(
    { id: { value: 1 } },
    {},
    { id: { value: 2 } }
);

Here, we start with two interfaces, named IValue and INestedObj. The IValue interface has a property named value of type number, and the INestedObj has a single optional parameter named id of type IValue. We then create an Observable named objEmit that emits three values. The first value has the nested structure described by the INestedObj interface, and the second is a blank object. Now consider the following Observable stream:

const returnIdValue = objEmit.pipe(
    map((value: INestedObj) => {
        return...

Observables returning Observables

Quite often, when working with Observables, we need to return a new Observable stream while already dealing with an Observable stream. In other words, for each Observable value in a stream, create a new Observable stream. While this might sound complicated, in the real world, it can happen fairly regularly.

Suppose that we are working with a REST API that tells us what products are sold within a sales catalog. This particular API call returns an array of product IDs that are associated with a particular catalog. For each of these product IDs, we then need to initiate a new REST API call to retrieve the information for this particular product, such as its name and description.

Let's assume that we are using an HTTP client that returns an Observable for each API call. This means that the first Observable stream will be the list of products within a catalogue, say, [1,2,3]. For each value in this stream, we then need to initiate a new API...

Observable Subject

Thus far, we have worked with Observables that emit values and seen how to subscribe to Observable streams. The Observable itself is responsible for emitting values, and the subscribers react to values being emitted. When an Observable stream is complete, all subscribers complete their processing, and their execution stops. In essence, subscribers are alive as long as the Observable stream is emitting values.

So what if we want to keep an Observable stream open and register one or more subscribers that will wait around until a new value is emitted? Think in terms of an event bus, where multiple subscribers register their interest in a topic on an event bus, and then react as and when an event is raised that they are interested in. RxJS provides the Subject class for this express purpose.

A Subject maintains a list of listeners that have registered their interest. A Subject is also an Observable stream, and therefore listeners can subscribe to the stream...

Summary

In this chapter, we have explored parts of the RxJS library and the fundamental concept of Observables that it provides. We have seen how we can create Observables easily using the of and from functions, as well as how we can transform values emitted from an Observable stream using the pipe, map, take, toArray, mergeMap, and contactMap functions. We also explored error handling through the error function available on a subscription, and how to handle errors within an Observable stream using the catchError function. We then explored how to wait for multiple Observable streams to complete with the forkJoin function and rounded out the chapter with a discussion on Subjects, and an implementation of a simple event bus.

We have only just scratched the surface of the functionality available in the RxJS library, however, but have covered some of the main topics and concepts. Feel free to head over to the RxJS website to view the wealth of available operators and functions that...