Extension for rxjs library

Table of Contents

• Installation
• Usage
• Methods
• Operators

Installation

First you need to install the npm module:

1npm install async-await-rx --save

In plain rxjs we have a lot of operators and we can do the same actions in different ways. Often we do repeatable actions such as fetching data parallely, one by one etc. and every time we have to build it with many operators. This extension can simplify code by introducing only few operators for repeatable actions. Thanks to this extension code is more clear and we can focus on action to do, not which operator we should choose.

This extension can be useful when:

• you need data from each actions,
• you want to call actions parallely or one by one and you want to wait on result - awaitAll
• you want to define dynamic actions - asyncPipeFrom

Usage

Comparizon with plain rxjs

code with async pipe extension

1asyncPipe(
2  awaitAction(() => action1()),
3  awaitAll(() => [action2_1(), action2_2(), action2_3(), action2_4()]),
4  awaitAction(() => action3()),
5  action((value, index, [_, items]) =>
6    console.log({ value, items /* data[1] */ }) // { 'action3', ['action2_1', 'action2_2', 'action2_3', 'action2_4']}
7  ),
8  awaitAction((value) => action4(value))
9)
10  .pipe(
11    takeUntil(cancel$),
12    finalize(() => (isLoading = false))
13  )
14  // Async pipe return value from last action and data from each action. It is array.
15  // we can use destructurization for example: [value, [action1Val, action2Val, action3Val]]
16  // first item is value from last action and the second is data from all actions
17  .subscribe(([value, [val1, val2, val3]]) => {
18    console.log(val1); // Display value from 1st action
19    console.log(val2); // Display items from 2nd action
20    console.log(val3); // Display items from 3rd action
21  });

code without extension (in plain rxjs)

1action1()
2  .pipe(
3    switchMap((data) =>
4      combineLatest([action2_1(), action2_2(), action2_3(), action2_4()]).pipe(
5        map((items) => [data, items])
6      )
7    ),
8    switchMap((data) => action3().pipe(map((value) => [...data, value]))),
9    tap(([_, items, value]) => console.log({ value, items })),
10    switchMap((data) =>
11      action4(data[2]).pipe(map((value) => [...data, value]))
12    ),
13    takeUntil(cancel),
14    finalize(() => (isLoading = false))
15  )
16  .subscribe(([val1, val2, val3]) => {
17    console.log(val1); // Display value from 1st action
18    console.log(val2); // Display items from 2nd action
19    console.log(val3); // Display items from 3rd action
20  });

Methods

• asyncPipe

It takes list of operators as params and return Observable. asyncPipe return array where first item is value from last action and the second is data from all actions. It is array to handle data in easier way (we can use destructurization for example: [value, [action1Val, action2Val, action3Val]])

Example

1asyncPipe(
2  awaitAction(() => action1()),
3  awaitAction(() => action2()),
4  awaitAction(() => action3())
5)
6  .pipe(
7    finalize(() => (console.log('finalize')))
8  )
9  .subscribe(([value, [val1, val2, val3]]) => {
10    console.log(val1); // action1
11  });

• asyncPipeFrom (factory)

It takes list of functions as params (it convert each function to awaitAction) and return Observable

Example 1

1    const something = 5;
2    const actions: any[] = [() => action1()];
3
4    if (something === 1) {
5      actions.push((value: string) => action2(value));
6    } else {
7      actions.push((value: string) => action3(value));
8    }
9
10    asyncPipeFrom(actions).subscribe(([_, data]) => { console.log(data) }); // [action1, action3]

Example 2

1asyncPipe(
2  awaitAction(() => asyncPipeFrom(actions)),
3  awaitAction(() => action2())
4)
5  .pipe(
6    finalize(() => (console.log('finalize')))
7  )
8  .subscribe(([value, [val1, val2]]) => {
9    console.log(val1);
10  });

• omit

It takes Observable and condition as inputs. Condition determine that observable should be subscribed or should be omited. When we will use omit in awaitAction and condition is met then action will be omitted (it won't be put in the asyncPipe result)

Example

1const something = 5;
2
3asyncPipe(
4    awaitAll(() => [
5        omit(action1(), something == 5), // this action will be omitted, because condition is met
6        action2(),
7    ]),
8    awaitAction(() => omit(action3(), something == 5)), // this action will be omitted, because condition is met
9    action((value) => console.log(value)), // [action1]
10    awaitAction(() => action4())
11).subscribe(([_, data]) => console.log(data)); // [[action1], action4]

Operators

Each operator takes project method as input with structure (value, index, data) => type value - value from previous action index - value index data - array with values from each action type - returned type

• awaitAction - (value, index, data) => Observable Action will be completed when observable have fulfilled.

• awaitAll - (value, index, data) => Observable[], strategy: AwaitAllStrategy Action will be completed when all of the observables have fulfilled. It breaks immediately upon any of the observables throwing an error.

As second param it determine execution strategy: AwaitAllStrategy:

• parallel (default) - actions are processing parallel

• oneByOne - actions are processing one by one

• action - (value, index, data) => void This acton has no impact on asyncPipe result. Used to perform side-effects.