From Promises to Observables
Tags: rxjs, javascript, typescript, angular
Published: 2019-08-20 19:09:33

---
published: true
title: "From Promises to Observables"
cover_image: ""
description: "How I switched my site from a Promises to a Reactive implementation with Observables"
tags: rxjs, javascript, typescript, angular
series:
canonical_url:
---

Recently I attended the Angular Denver Conference in Denver, Colorado. It was an awesome experience and one of the biggest takeaways I brought home was the power of RxJS.

While I was at the conference I attended an RxJS workshop led by [Jan-Niklas Wortmann](https://twitter.com/niklas_wortmann) and [Michael Hladky](https://twitter.com/Michael_Hladky). I have previously used RxJS in some of my Angular projects, but wanted to learn more of the fundamentals and really understand how the technology works. During the workshop I learned how to think in terms of streams, and how the basic fundamentals of RxJS work. I also learned about the concepts behind Higher Order Observables, and how you can use them to increase performance in your applications.

I was so impressed with RxJS that I went home and actually used Observables to improve performance for one of the pages on my site [andrewevans.dev](https://www.andrewevans.dev).

In this post I'm going to cover how I was able to use RxJS to increase performance in my site. Specifically I'm going to show how I was able to use RxJS to manage multiple HTTP calls at once, and how this significantly improved my user experience.

I'm also going to go over some basics, but I highly recommend the official RxJS documentation at [rxjs.dev](https://rxjs.dev).

I've created a small Angular application that showcases what I did. You can view it on [Stackblitz](https://stackblitz.com/edit/learning-rxjs-with-angular) or at my [GitHub repo (https://github.com/andrewevans0102/learning-rxjs-with-angular).

This post also assumes you have a working knowledge of Angular. The example that I’m going to be showcasing is a traditional promise based approach compared with a reactive approach using RxJS.

## IMPERATIVE VS DECLARATIVE

Two big words you often see with RxJS are **imperative** and **declarative**.

**Imperative** refers to code that you have to manually write yourself. This is code that you specifically write to act in a specific way. For synchronous tasks this is perfect, but for handling application events this can be cumbersome.

**Declarative** refers to letting RxJS do the work for you. Basically, by leveraging the library you define event stream flow. Instead of having to specifically build code around handling different events, RxJS enables you to use **observables** and **operators** to do the work for you.

This all will be easier to understand as we go through the next sections. I'm just introducing these topics first.

## THE BASICS

At its core, RxJS is a library that utilizes streams for handling asynchronous activities. RxJS is a safe way to handle events in your code through the predefined behaviors and contracts that come with the **observables**.

> I know I’m just hitting the high points here, but for a more in depth explanation I highly recommend the documentation at [rxjs.dev](https://rxjs.dev).

RxJS has [observables](https://rxjs.dev/guide/observable) and [operators](https://rxjs.dev/guide/operators). RxJS also makes use of [subjects](https://rxjs.dev/guide/subject) for multicasting events in your applications.

Most people will first encounter RxJS through observables. An observable will typically look something like this:

```ts
import { Observable } from 'rxjs';

const observable = new Observable(function subscribe(subscriber) {
  try {
    subscriber.next(1);
    subscriber.complete();
  } catch (err) {
    subscriber.error(err);
  }
});
```

If you notice there are the following calls:

- **next**
- **complete**
- **error**

These are based around the observable model or contract. **Next** is what handles emitting events in the stream. **Complete** frees up the observables resources and essentially ends the stream. **Error** will return an error to anything that has **subscribed**.

What is a subscription? **Subscriptions** in RxJS are what starts a stream's execution. Whatever is defined in the **next** value will emitted as soon as a subscription is started. When a call is made to **complete**, the resources are freed and this observable is essentially finished.

You can also end a stream with **unsubscribe** or **complete**. If you use **unsubscribe**, you end a stream manually meaning that the resources are freed and there will be no more events. If you use **complete** then it marks the stream as finished. To clarify, when thinking of **unsubscribe** and **complete** just remember:

- **unsubscribe** means "the stream is not interested in new values"
- **complete** means "the stream is finished"

When you see **operators**, they are static functions that provide all of these same services we see in **observables** out of the box. Operators can be intimidating because there is a large number. However, most of them are wrapped around core behaviors. I highly recommend the workshop I mentioned earlier with [Jan-Niklas Wortmann](https://twitter.com/niklas_wortmann) and [Michael Hladky](https://twitter.com/Michael_Hladky) for a more in depth explanation using what they call the "algebraic approach" to operators.

## MY PROJECT

In my example I’m going to use both observables and operators.

The challenge that I wanted to resolve was that the blog page on my site [andrewevans.dev](https://www.andrewevans.dev) required retrieving several RSS feeds. I originally had coded it to take in all of the HTTP calls to the RSS feeds with the `promise.all()` approach. This basically tried to run all of them as promises in parallel, and when the requests completed then I could return all of the data. The code in my API endpoint looked like the following:

```js
const output = [];
// feed addresses to use in call to rss parser
let feedInput = [
  {
    sourceURL: 'https://medium.com/feed/@Andrew_Evans',
  },
  {
    sourceURL: 'https://rhythmandbinary.com/feed',
  },
  {
    sourceURL: 'https://dev.to/feed/andrewevans0102',
  },
];
const promises = [];
feedInput.forEach(feed => {
  // add all rss-parser calls as promises
  promises.push(
    parser
      .parseURL(feed.sourceURL)
      .then(response => {
        response.items.forEach(item => {
          let snippet = '';
          if (item.link.includes('dev.to')) {
            snippet = striptags(item['content']);
          } else {
            snippet = striptags(item['content:encoded']);
          }

          if (snippet !== undefined) {
            if (snippet.length > 200) {
              snippet = snippet.substring(0, 200);
            }
          }

          const outputItem = {
            sourceURL: feed.sourceURL,
            creator: item.creator,
            title: item.title,
            link: item.link,
            pubDate: item.pubDate,
            contentSnippet: snippet,
            categories: item.categories,
          };
          output.push(outputItem);
        });
      })
      .catch(error => console.log(error)),
  );
});

try {
  await Promise.all(promises);
} catch (error) {
  res.status(500).send(error);
}
```

If you notice the code here basically wraps a set of URLs that are called by the `callRSS` function. That function just uses the [rss-parser](https://www.npmjs.com/package/rss-parser) to call the RSS feeds and parse the values. This code looks like the following:

The resulting experience from this code was that (in the Angular client) I had to provide a traditional spinner on the page to show until all of the promises completed. This actually ended up being several seconds and was not optimal.

In the example, if you go over to the `functions` folder in the `index.js` file you can see the endpoint's code that wraps the promises in the `traditional` endpoint.

In the same example project, if you go over to the `traditional` Angular Component you'll see the client code calling the endpoint with the following:

```ts
  async load() {
    // show spinner while fetching posts
    this.showSpinner = true;

    // retrieve the posts by calling the endpoint that uses promise.all for fetching
    // all of the rss feeds and waiting (synchronously) for them to complete
    this.posts = [];
    const response = await axios.get(environment.traditionalEndpoint);
    response.data.forEach((entry) => {
      const inputDate = new Date(entry.pubDate);
      entry.pubDate = inputDate.toLocaleDateString('en-us') + ' at ' + inputDate.toLocaleTimeString('en-us');

      this.posts.push({
        ...entry,
        sortDate: inputDate.getTime()
      });
    });
    this.posts = response.data;

    // retrieve the manual entries
    const manualEntries: any
      = await axios.get(environment.manualEntries);
    manualEntries.data.forEach((entry: any) => {
      const inputDate = new Date(entry.pubDate);
      entry.pubDate = inputDate.toLocaleDateString('en-us') + ' at ' + inputDate.toLocaleTimeString('en-us');
      if (entry.contentSnippet.length > 200) {
        entry.contentSnippet = entry.contentSnippet.substring(0, 200);
      }

      this.posts.push({
        ...entry,
        sortDate: inputDate.getTime()
      });
    });

    // sort by date here
    this.posts.sort((a: any, b: any) => {
      return b.sortDate - a.sortDate;
    });

    // stop showing spinner when fetch completes
    this.showSpinner = false;
  }
```

Here I've created a `load` method that uses [axios](https://www.npmjs.com/package/axios) to make a GET call to retrieve the data. I also call an endpoint for an additional set of manual entries. When both complete I stop showing the spinner.

## Enter RxJS

So after my experiences from Angular Denver, I started thinking of ways that RxJS could improve this whole setup. I really didn't like the spinner and several second wait time, so thought this was a great opportunity to improve the site.

I decided that it would help if I could break down the HTTP calls to be handled individually as streams. Then the user would see results immediately on their page, and it would update as the calls completed. Since its only a matter of seconds, this didn't make the page jump around too much and made the site feel very responsive.

> I should note that typically with frontend engineering you don't want a page to jump around. In this situation it was only a matter of seconds and the way that I built the flow made it hardly noticeable. It was more important for me to show the output immediately. In a larger application this approach may need to be reworked for a better experience.

I refactored the HTTP calls to be done in one endpoint. If you look in the example project, you can see this [in the functions/index.js file](https://github.com/andrewevans0102/learning-rxjs-with-angular/blob/master/functions/index.js) here.

The code here is pretty straightforward, based on the "source" parameter it makes a call to the matching RSS feed. The results are gathered from the HTTP call and returned in the output value.

Now for the RxJS implementation, I wrapped each of the HTTP calls to this endpoint in a separate observable. This enabled each HTTP call to start at the same time. As soon as the individual HTTP calls completed, their responses were shown on the page.

```ts
  load() {
    const medium =
      this.http.get(environment.reactiveEndpoint + '/medium')
      .pipe(
        catchError(err => {
          throw 'error in source observable. Message: ' + err.message;
        })
      );

    const wordpress =
      this.http.get(environment.reactiveEndpoint + '/wordpress')
      .pipe(
        catchError(err => {
          throw 'error in source observable. Message: ' + err.message;
        })
      );

    const devto =
      this.http.get(environment.reactiveEndpoint + '/devto')
      .pipe(
        catchError(err => {
          throw 'error in source observable. Message: ' + err.message;
        })
      );

    const manualEntries =
      this.http.get(environment.manualEntries)
      .pipe(
        catchError(err => {
          throw 'error in source observable. Message: ' + err.message;
        })
      );

    this.posts$ =
      merge(medium, wordpress, devto, manualEntries)
        .pipe(
          scan((output: Post[], response: []) => {
            response.forEach((post: Post) => {
              const inputDate = new Date(post.pubDate);
              post.pubDate = inputDate.toLocaleDateString('en-us') + ' at ' + inputDate.toLocaleTimeString('en-us');
              post.sortDate = inputDate.getTime();

              if (post.sourceURL === 'https://blog.angularindepth.com/feed') {
                post.sourceURL = 'Angular-In-Depth';
              } else if (post.sourceURL === 'https://itnext.io/feed') {
                post.sourceURL = 'ITNext';
              } else if (post.sourceURL === 'https://medium.com/feed/@Andrew_Evans') {
                post.sourceURL = 'Medium';
              } else if (post.sourceURL === 'https://rhythmandbinary.com/feed') {
                post.sourceURL = 'Rhythm and Binary';
              } else if (post.sourceURL === 'https://dev.to/feed/andrewevans0102') {
                post.sourceURL = 'DEV.TO';
              }
              output.push(post);
            })

            output.sort((a: any, b: any) => {
              return b.sortDate - a.sortDate;
            });

            return output;
        }, []),
        catchError(err => {
          throw 'error in source observable. Message: ' + err.message;
        }),
        takeUntil(this.unsubscribe)
      );
  }
```

Here I am taking advantage of Angular's [HttpClient](https://angular.io/guide/http) which wraps HTTP calls in an observable.

I then use the [merge](https://rxjs.dev/api/index/function/merge) operator to subscribe to all of the HttpClient calls and combine them into one output.

The [scan](https://rxjs.dev/api/operators/scan) operator then takes the merged observables and appends the response to one common output.

I include the [catchError](https://rxjs.dev/api/operators/catchError) operator to handle any errors in the stream, if one of the calls fails etc.

I also use **pipe** to take the output of one observable and passes it into another. This is a common pattern with RxJS.

The last operator that was passed into the **pipe** also references a [takeUntil](https://rxjs.dev/api/operators/takeUntil) operator. This is a very powerful RxJS operator that will unsubscribe an observable based on an event you pass in. Here I've created a [subject](https://rxjs.dev/guide/subject) which handles unsubscribing this main observable when the code finishes running. This is a fairly common pattern when handling observables. RxJS **subjects** also can be used for multicasting and doing observable like actions. I'm just using it here because it provides the behavior I wanted and makes a simple `unsubscribe` call clear out the resources. If I didn't do this it could cause `memory leaks` and potentially freeze up my browser session. You can see this behavior in the `reactive` component's `clear` method:

```ts
  clear() {
    this.unsubscribe.next();
    this.unsubscribe.complete();
    this.posts$ = null;
  }
```

Also note that I make the observable `null`. This is not necessary, but for the basic example application I wanted to visually show the data disappearing when `clear` was called.

> As a disclaimer here, there are also 100+ other ways to do this same type of event handling. Additional implementations include using **from** to cast promises, **fromFetch** using the `fetch` API, or even an ajax implementation with `rxjs`. I chose the implementation here because it was easy to follow and I was happy with the output. RxJS is very flexible and gives you the power to control the streams with (almost) any behavior.

You can see this code in the `reactive` Angular Component in my project.

The `load` method does the subscription and starts the stream.

The `clear` method stops the stream and clears the array that is shown on screen.

## Marble Diagrams

The code that I've written here resulted in a streamed approach to the RSS calls that I made. It made my application more responsive, and I didn't need to show a spinner while the application was waiting for a response.

To understand this behavior, it might be helpful to also have a basic marble diagram. Marble diagram's are great ways to graphically represent RxJS behavior.

Here's a Marble Diagram explained:

![marble diagram](https://raw.githubusercontent.com/rxjs-blog/blog/master/blog-posts/from-promises-to-observables/assets/marble-diagram-anatomy.svg)

The following is a copy of the `merge` marble diagram from the RxJS documentation:

![merge operator](https://raw.githubusercontent.com/rxjs-blog/blog/master/blog-posts/from-promises-to-observables/assets/merge.png)

The following is a copy of the `scan` marble diagram from the RxJS documentation:

![scan operator](https://raw.githubusercontent.com/rxjs-blog/blog/master/blog-posts/from-promises-to-observables/assets/scan.png)

To see all this in action, look at my application in stackblitz. The application flow is very intuitive. The `traditional` tab makes the HTTP calls in the traditional (imperative) approach, and the `reactive` tab makes the HTTP calls using the RxJS observables and operators I've been discussing.

{% stackblitz learning-rxjs-with-angular %}

## Closing Thoughts

So here I've introduced some RxJS concepts and shown a working example.

I've shown how you can change your project from using Promises to Observables with RxJS.

Reactive extensions are a big shift in traditional software development. Streams make our applications more responsive and are actually easier to build.

I recommend checking out the [RxJS documentation](https://rxjs.dev) and my example project for more.

Hope you enjoyed my post! Feel fee to leave comments and connect with me on Twitter at [@AndrewEvans0102](https://twitter.com/AndrewEvans0102) and at [andrewevans.dev](https://www.andrewevans.dev).