As it often happens to me — this post is just a prolongation of my StackOverflow activity:-).

Question:

Is there a way to gather a list of observables together to get cumulative result like forkJoin does, but get some kind of a progress while they finishes?

A nice task for creating custom RxJS function. Let's implement it!

What is forkJoin?

Let's take a look at official documentation:

__****forkJoin(…sources: any\[\]): Observable<any>****__

Accepts an Array of [ObservableInput](https://rxjs-dev.firebaseapp.com/api/index/type-alias/ObservableInput) or a dictionary Object of [ObservableInput](https://rxjs-dev.firebaseapp.com/api/index/type-alias/ObservableInput) and returns an [Observable](https://rxjs-dev.firebaseapp.com/api/index/class/Observable) that emits either an array of values in the exact same order as the passed array, or a dictionary of values in the same shape as the passed dictionary.

In other words, **forkJoin** for **Observables** works in the same way as **Promise.all** works for **Promises.**

Here is a marble diagram:

The possible use-case for it is running many parallel networks requests — for example, fetching some usersList details (if API supports only per user fetch).

Here is an example:

snippet link

Here is a codepen link to play with.
It works nice but what if we need to know some intermediate information like how many requests are resolved already? With current forkJoin implementation, it is impossible but we can create our own version of it ?.
OK, so brief agenda how this forkJoinWithPercent should work:
Output:
1. It returns higher-order(HO) Observable ( ) that emits an array of two Observables: [finalResult$, percent$].

*Higher-order (HO) Observable - observable that emits other observables, so data flow should be handled with special flattening operators like mergeMap, switchMap, concatMap, etc... you can read more about it here and here.

2. percent$ — emits percentage of completion (number);
3. finalResult$ — emits a final array of values at the end (or error if some argument observable emits error value).
Preliminarily usage looks like this:

snippet link

**Implementation details:**

1. Our **forkJointWithProgress** function accepts an array of Observables and should return higher-order Observable.
2. This returned Observable should emit value **[finalResult$, percent$] (**we will use **of****([finalResult$, percent$])** for that**)**
3. Side effects will be added to each item in an array of argument Observables to calculate the percentage of completion and emit this value with special **percent$** (we will use **finalize****** operator for percentage calculation and use Subject as **percent$**).
4. **finalResult$** provides standard RxJS **forkJoin** result.
5. Result Observable should work independently for each subscriber (wrap functionality with RxJS **defer** function to provide a clean run for each subscriber — you can read more about this case here).
6. If some of the argument Observables emits error — it will be propagated to **finalResult$** subscribers.

Packtpub.com and I prepared a whole **RxJS course****** with many other details of how you can solve your every-day developer’s tasks with this amazing library. It can be interesting for beginners but also contains advanced topics. **Take a look!**

Implementation

**#1–2 We accept an array of Observables and should return higher-order Observable with [finalResult$, percent$].**

Ok, just a footprint of our future function:

snippet link%250A%257D)

Given we got arrayOfObservables (our ajax observables), so let’s iterate over and add requested logic:

**#3 Adding side effects to argument Observables and calculating percentage.**

Here is what we’ve got:

snippet link

We will add more functionality to argument observables in array. We iterate over an array and apply **finalize** operator for each observable.

**finalize** operator waits till specific observable completes, then calculate the percentage of completed observables and emit value with **percent$**Subject.

**#4 Let's call forkJoin for getting a final result value**

snipper link

Ok, let's go through this code:
forkJoin gets an array of argument observables, subscribes to all of them and waits till they are complete.
Once a result is ready — tap operator emits last percentage value (100%) and completes percentage$ Subject (to prevent memory leaks).
Final results will be emitted to subscribers.

5 Wrapping all this function in ‘defer’

RxJS defer function provides new Observable instance (a result of its callback function) for each subscriber. This means that each of our subscribers will get clean run (and counter = 0).

snippet link

Lets review how it works:

1. We get an array of observables (line1).
2. Wrap result with RxJS **defer** to provide a clean run for each subscriber (line 3)
3. Create **counter** to calculate the percentage of completion and instantiate **percent$** Subject to emit percentage value (lines 5–6).
4. We create a new array by iterating over the original array and adding new functionality with **finalize** – calculating percentage value and emitting it if some of the observables completes. (lines 8–15)
5. Call Rx **forkJoin** and apply **tap** operator to be able to send percentage 100 when forkJoin gets a final result. Assign result to **finalResult$**variable (lines 17–22).
6. Return higher-order observable that will emit **[finalResult$, percent$].**

To wrap up

Here is how it works in a codepen:

forkJoinWithPercent

What about using it in Angular?

This function is published as an npm package as well — **rxjs-toolbox** so you can use it in your projects.

And here is a **Stackblitz Angular demo** that uses that package to feed percentage value to load bar:

In Angular

Additional reading.

You can read more about RxJS operators uses cases here:

1. “Retry vs Repeat”
2. “RxJS: Managing Operator State” by Nicholas Jamieson.
3. “RxJS ‘repeat’ operator — beginner necromancer guide”
4. “Throttling notifications from multiple users with RxJS”
5. rxjs-toolkit — RxJS Everyday Custom Operators by Jason Awbrey.
6. backoff-rxjs — A collection of helpful RxJS operators to deal with backoff strategies by Alex Okrushko.

Like this article? Let’s keep in touch on Twitter!

Starting from section 4 of my **RxJS video course****** advances staff is reviewed — so if you familiar with RxJS already — you can find something useful for you as well: higher-order observables, anti-patterns, schedulers, unit testing, etc! **Give it a try********!**

***Special thanks to Lars Gyrup Brink Nielsen, Nicholas Jamieson, Tim Deschryver and Michael Karén for reviewing this post and making many valuable remarks to make it better!**