Rxjs + For + Await… What?

If you have been making use of TypeScript/ES7+ for a while, you are undoubtedly well-acquainted with the async/await keywords and the convenience they bring to managing asynchronous operations. Take a look at this elementary example to refresh your recollection:

async function request() {
  const result = await fetch('some-api-url');
  const data = await result.json();
  return data.items; // for example a nested array;
}

const list = await request();
list.map(/*do stuff with the array*/);

Note that I've employed top-level await, a feature that has been accessible since TypeScript v3.8.

Within this example, we construct a simple Promise and deal with it in a synchronous-looking manner, steering clear of .then and callbacks. As a result, the code takes on a more sequential appearance, and the absence of callbacks simplifies the reasoning process significantly.

So, what exactly is for-await?

Naturally, Promises symbolize the outcome of one asynchronous action, but what scenario arises when we deal with a continuous operation that dispatches events over a period of time? This might ring a bell, and it's precisely where for await steps in:

for await (const event of events) {
  handleEvent(event);
}

Here, we're presented with an event stream (crucial point: this is not an array; rather, it's a stream that dispatches events asynchronously). The twist? We're not handing over a callback to manage events as time progresses. Instead, we harness for await to tackle these events in a much more sequential manner. This approach, combined with conventional async/await, enables us to completely eradicate callbacks from our codebase!

But does it really?…

RxJS Observables hold greater power than Promises

Without a doubt, RxJS Observables are not just capable of streaming a vast number of events, in contrast to the single output from a Promise, but they also boast a rich set of operators that process data in an elegant, functional style.

However, is there a way to combine async/await with RxJS? It appears not. Or, at least, not in a straightforward manner.

Granted, when our Observable is designed to emit just a single value before completing, we have the option to use toPromise and subsequently apply async/await. And before you question the need for an Observable that yields only one value, bear in mind that Angular's HttpClient is fundamentally constructed around Observables that emit once and then terminate. Thus, with HttpClient, we can transform the resulting Observable into a Promise and reap the benefits of both async/await and RxJS operators!

Alright, that seems fine up to this point. But how do we handle streams of events?

import { fromEvent } from 'rxjs'; 

const clicks$ = fromEvent(document.body, 'click');

Here, I've established a stream that captures every click event on the document's body. Alone, this doesn't accomplish much without handling those events—logging them to the console at the very least. Let's see how we might approach this:

const clicks$ = fromEvent(document.body, 'click').subscribe(event => console.log(`Event ${event} occured`));

This, however, introduces a callback within the subscribe function. Is there a way to integrate async/await here to sidestep callbacks? Not exactly, since toPromise depends on the source Observable completing, and this particular Observable never finishes. So, are we forced to stick with the callback?

?Presenting rxjs-for-await! A library that offers multiple strategies for subscribing to #RxJS observables using async/await and `for await` loops

Now available on npm pic.twitter.com/l9tLlseOUR

— Ben Lesh (@BenLesh) March 18, 2020

So, what does it entail? As shown in the screenshot, this library supplies the eachValueFrom function, which takes a source Observable and turns it into an AsyncInterable (check out more details about AsyncIterables here). This, in turn, enables us to employ for await on an Observable stream! Here's how we might refactor our code:

import { fromEvent } from 'rxjs'; 
import { eachValueFrom } from 'rxjs-for-await';

const clicks$ = fromEvent(document.body, 'click');

async function handleClicks() {
  for await (const event of eachValueFrom(clicks$)) {
    console.log(event);
  }
}

handleClicks();

Now, writing callbacks is no longer necessary—we can readily use for await with RxJS Observables.

Where can it be beneficial?

Naturally, this doesn't suggest you should rush to refactor every application immediately; however, there are scenarios where this handy utility proves useful.

One such case is in unit testing. Crafting tests demands they remain as straightforward and lean as possible. Introducing subscribe calls with callbacks adds needless complexity and nested structures to unit tests (keep in mind—tests should read like plain language to accurately mirror what the application is doing). Take, for instance, this snippet:

describe('AppComponent', () => {
  beforeEach(async(() => {
    TestBed.configureTestingModule({
      imports: [
        RouterTestingModule
      ],
      declarations: [
        AppComponent
      ],
    }).compileComponents();
  }));

  it(`should have a source$ Observable emit numbers`, () => {
    const fixture = TestBed.createComponent(AppComponent);
    const app = fixture.debugElement.componentInstance as AppComponent;
    app.start();
    let counter = 1;
    app.source$.subscribe(data => {
      expect(data).toBe(counter);
      counter++;
    });
  });
});

This test invokes a start method on the AppComponent, which should generate a stream of numbers incrementing by one. Following that, we verify that the stream indeed outputs such numbers. Clearly, we resorted to subscribe to capture the emitted values, resulting in a third level of callback nesting. If additional checks with a callback inside that final callback were needed, a common occurrence in unit tests, we'd find ourselves in callback hell. rxjs-for-await steps in to resolve this:

it(`should have a source$ Observable emit numbers`, async () => {
  const fixture = TestBed.createComponent(AppComponent);
  const app = fixture.debugElement.componentInstance as AppComponent;
  app.start();
  let counter = 1;
  for await (const data of eachValueFrom(app.source$)) {
    expect(data).toBe(counter);
    counter++;
  }
});

Here, we've leveraged eachValueFrom to perform the same assertions using for await, thereby eliminating nested callbacks. Moreover, if further asynchronous checks are necessary, async/await comes into play, all without any callbacks.

Wrapping up

While it's not necessary to rewrite all our codebases overnight, this library grants us the capability to incorporate the latest ES features while continuing to benefit from the strength of RxJS.