RxJS Awaitable Observables

🧨    💥    💪    Use async/await with any RxJS stream and be happy    ✅    👨‍💻    😎

Zero third-party dependencies

 

Notice: If you have any propositions feel free to make an issue or create a pull request.

Installation

1. Install the package

yarn add @z-brain/rxjs-awaitable-observables
or
npm i -s @z-brain/rxjs-awaitable-observables

2. Add the import line of the package to your entry .ts files

Your main.ts:

import '@z-brain/rxjs-awaitable-observables';

That is all 🙂  Feel free to use async/await with observables 😉

Detailed description:

• In case of Angular project it can be main.ts
• In case of a server project it is the entry file of your server app (index.ts or server.ts in most cases)

3. Configure you testing framework

If you are going to use async/await with observables in any kind of your tests (unit / e2e) there are two ways to do it:

1. Add a setup file with the import to you testing framework. In case of Jest it can be done via setupFiles or setupFilesAfterEnv. (recommended)

// jest.config.js
module.exports = {
...
setupFiles: [ './jest.setup.ts' ],
...
};

// jest.setup.ts
import '@z-brain/rxjs-awaitable-observables';

2. Or you can manually import the package to each of your test (unrecommended)

Usage examples

1. Simple example: async/await works with completed observables

const value$ = $$.of(123);
const value = await value$;
expect(value).toBe(123);

2. Take the first value in the stream

const value$: Observable<number> = $$.of(111, 222, 333);
const value = await value$;
expect(value).toBe(111);

3. Takes the last value of a completed stream

const value$: Observable<number> = $$.of(111, 222, 333);
const value = await value$.pipe($.last());
expect(value).toBe(333);

4. Take the initial value of Subject + .startWith()

      const newValues$: Observable<number> = new Subject<number>();
      const value$ = newValues$.pipe($.startWith(123));

      const value = await value$;
      expect(value).toBe(123);

5. Take the first value of ReplaySubject

const value$ = new ReplaySubject<number>(2);
value$.next(111);
value$.next(222);
value$.next(333);
value$.next(444);

const value = await value$;

expect(value).toBe(333);

6. Error handling using try/catch

const MSG = 'Something went wrong';
const err$ = $$.throwError(new Error(MSG));

try {
  await err$;
} catch (e) {
  expect(e.message).toMatch(MSG);
}

7. throw an EmptyError on completing observable without a message

const emptyStream$ = $$.EMPTY;

try {
    await emptyStream$;
} catch (e) {
    expect(e).toBeInstanceOf(EmptyError);
}

8. Manually .then() call (Promises integration)

const value$: Observable<number> = $$.of(123);

const promise = Promise.resolve(10)
                       .then(multiplier => value$.then(v => v * multiplier));

return expect(promise).resolves.toBe(1230);

How does it work?

• This package patches Observables's prototype and adds .then() method to make all Observables and its children work with async/await as is without the necessary to call .toPromise()
• Yes, the package patches .prototype of another package (RxJS).
  In theory, it can be dangerous. However, this package does it enough carefully.
  We already saw some successful examples of patching .prototype even build-in functions. (Zone.js for example)
  I suggest don't afraid and make our daily work easier and our code more beautiful.

100% test coverage:

async/await tests
  ✅ async/await should work with completed observables
  ✅ Should take the first value in the stream
  ✅ Should take the last value of a completed stream
  ✅ Should take the current value of BehaviorSubject
  ✅ Should take the initial value of Subject + .startWith()
  ✅ Should take the first value of ReplaySubject
  ✅ try/catch should handle an error from the stream via async/await
  ✅ Should throw an EmptyError on completing observable without message
Check integration with Promises
  ✅ Should work with manually .then() call
  ✅ Empty .then() call should just return a promise with the value without any errors
  ✅ In case of error in the stream empty .then() call should do nothing and return rejected promise with the error

Development notes

Quick Start

cd /code/z-brain
git clone git@github.com:z-brain/rxjs-awaitable-observables.git
cd rxjs-awaitable-observables
yarn install

How to use NodeJS version from the .nvmrc

1. Install NVM

2. Use .nvmrc file one of the next ways:

   • Execute nvm use in the project root directory
   • Install NVM Loader and your .nvmrc will be loaded automatically when you open the terminal.

How to make a build

npm run build

How to run lint

• Just show problems npm run lint
• Fix problems if it is possible npm run lint:fix

How to run tests

• All tests

  npm run test
npm run test:watch

• Specific tests

  npm run test -- src/my.spec.ts
npm run test:watch -- src/my.spec.ts

How to build and publish NPM package

NPM Token: 3240...1349

CI configuration details here: .github/workflows/npmpublish.yml

npm run pre-push
&& npm version patch -m 'Update package version version to %s'
&& npm run gen-public-package.json
&& cp README.md dist/
&& npm publish dist --access public
&& git push --no-verify && git push --tags --no-verify

How to build package to local installation

1. yarn run build:local
2. Then you can install a local package build from path file:.../rxjs-awaitable-observables/dist.

Author

Anton Korniychuk