What can I do with Observables?

Previously in "What is the difference between a Promise and an Observable?", I covered the basics of what Observables can do above and beyond what Promises can do. That barely scratches the surface of what Observables and RxJs really offer and their capabilities are vast.

Reactive Programming

I've previously used the term "reactive" and "reactivity" in the article mentioned above. You may be wondering what that means and how observables allow "reactive" programming. This is where observables and RxJs get really interesting and overshadow promises. Observables and the capabilities they offer don't really shine until you bring in the rest of RxJs and really get into fully reactive programming.

To put it another way: Observables are the gateway into the wonderful world of reactive programming which itself leads to functional programming without the need to explicitly worry about asynchrony in JS.

Using RxJs for Reactive Programming

RxJs is the Reactive Extensions library for JavaScript. It is complete with TypeScript typings and is used internally by Angular. So as an Angular developer, you already have RxJs at your fingertips. While Observables allow us to handle asynchronous events and streams of data very easily they are only the most overt and commonly used concept wrapped within the library. RxJs provides not just observables but also subjects, operators and a host of utilities to create rich, functional and reactive applications.

Observables are just the starting point for all of this reactive capability. Once an Observable is in hand, the real fun begins. An Observable, as previously mentioned, represents a way to watch (observe) streams of Notifications which represent values, errors and completions in a stream. From an Observable we can create a pipe of discrete operators that handle, transform, switch, fork and join, etc. the data moving through the stream. Subject and its derivatives give us the power to easily create sources for streams of observable data and feed data into them. Operators like map, tap, forkJoin, catchError and many more allow us to watch, augment, switch and branch and join streams of data.

Transforming Data

Revisiting our previous example of a WebSocket feeding us data. It may be that we simply want to grab the data and display it, however, it may also be that we want to trace that data, transform it, combine it with other data, maybe even split some of that data off into separate streams.

We use the pipe() function to do this. We can actually access the RxJs pipe in one of two ways: by calling .pipe() on an actual Observable or by importing and using the pipe() function to create our own functional, reactive operators that can be used in the observable pipeline.

The Observable to Subscriber part of our previous diagram for our WebSocket example; this time with some pipeable operators in play to transform the data in the stream to our needs:

In this example, whenever we receive a Notification of data (next) from Obs1, we proceed to tap the stream (which can be used to perform side effects, such as logging) then we map the data in the stream (say extract more deeply nested data from a wrapper object or to transform the data, or both, etc.). We include a catchError at the end to handle any errors that occur in the stream. With catchError we may choose to re-throw the error, or a transformation of it, with throwError or perhaps transform or replace the error into a non-error with of or from.

Combining Streams

Now, what if we needed additional data? What if we need additional data from another asynchronous process? Both are possible but the latter case is the more interesting case! We could potentially need to combine streams of data from two different WebSockets and process data from both in a certain way. Observables can do this!

In this next example, whenever we receive a Notification of data (next) from Obs1, we also grab the latest data from the last successful (non-error*, non-complete) notification from Obs2 using withLatestFrom. We then proceed with the same pipeline from our previous example.

This is only one way to combine data with RxJs and observables. We have many other options with RxJs, as it provides a host of both operators (which operate on data within a pipe) as well as factory functions (which operate on observables). We could also use the combineLatest factory function to combine data from our two source observables preemptively, then pipe off of the stream that combines the two.

Switching Streams

In addition to combining streams, with observables we may also switch from one stream to another. Switching, unlike combining, will change the data in the stream from the original data from the original source to alternative data from an alternative source.

Using switchMap exposes the original data in a child context so that data may be used to call functions or methods to get the Observable you wish to switch to. In addition to switchMap there are several other stream mapping operators that each have unique semantics, including: exhaustMap, concatMap, mergeMap and flatMap.

Forking and Joining

We can also fork and join a single stream into multiple streams and back into a single stream again. Instead of covering this useful and often complex topic here, take a look at another one of my articles for more:

What's a Fork and why would I Join one?

More RxJs!

This is still just barely scratching the surface of what Observables, and RxJs, can do and what they really provide. There are many factory functions like of, from, combineLatest, fromEvent and more, which allow the creation of new observable streams from a diverse set of event or data sources.

There are also an extensive set of operators that may be used to augment data streams, including: map, filter, reduce, scan, take, skip, first, debounce, distinctUntilChanged and many, many more!