RxJS recipes: ‘forkJoin’ with the progress of completion for bulk network requests in Angular

As is often the case, this piece grew out of my activity on StackOverflow.

The question:

Can we combine a list of observables to get a single cumulative result the way forkJoin does, while also receiving updates on how far along the process is?

There is a clear opportunity to build a custom RxJS utility here. Let's do it.

Understanding forkJoin

The official docs describe it like this:

__****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.

So in essence, forkJoin for observables behaves equivalently to how Promise.all functions for promises.

Here's the classical marble diagram:

A typical application is firing off many independent network calls — for instance, pulling the details for each user separately when we only have their IDs.

Consider this sample:

See the snippet for a cleaner look.

There is also a codepen available to play with.
This approach works well for aggregated results, but it offers no visibility into intermediate states — say, how many requests have already finished. The stock forkJoin cannot tell us this. That's a gap we can fill with a custom operator.

Here’s the plan for forkJoinWithProgress:

Output contract:

1. It returns a higher-order Observable that emits a tuple 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 a numeric percentage of completed sources.
3. finalResult$ emits the accumulated results at the end (or an error if any source errors).

The intended usage is shown below:

link to the code snippet

Design details:

1. The forkJoinWithProgress function accepts an array of Observables and produces a higher-order Observable.
2. The returned Observable emits a tuple [finalResult$, percent$] (wrapped in of).
3. We inject side-effects into each of the input Observables: when one completes, we bump a counter, calculate completion percentage, and push it through the percent$ Subject (using the finalize operator).
4. The finalResult$ is simply the classic forkJoin aggregate.
5. Each subscriber gets an independent execution — this is achieved by wrapping the whole flow in the RxJS defer factory (more about this pattern here).
6. Errors from any of the input Observables propagate directly to the finalResult$ subscriber.

---

A comprehensive RxJS course is available courtesy of Packtpub. It covers both fundamental and advanced topics and offers practical solutions for everyday dev tasks. Check it out!

---

Building the Operator

#1–2: Accept an array of Observables; return a higher-order Observable of [finalResult$, percent$].

We start with the skeleton of the function:

view the snippet

Now we have our arrayOfObservables (our AJAX Observables). The next step is to loop through them and attach the necessary logic.

#3: Add the side-effects and compute the percentage.

Here is how we modify the inputs:

view the snippet

We are augmenting each observable in the array. By mapping over the input, we attach a finalize operator to each.

When an observable finishes, the finalize callback runs: we increment the counter, divide by the total, multiply by 100, and send the result via the percent$ Subject.

#4: Bring in forkJoin for the final aggregation.

view the snippet

Let's walk through this chunk:

• We pass the modified list into forkJoin, which subscribes to every observable and waits for all of them to emit and complete.
• When the aggregate result is ready, the tap callback emits the final value of 100 to percent$ and then completes the Subject. This is also a necessary memory-safety step.
• Finally, finalResult$ itself emits the combined result to whoever subscribes.

Wrapping the Whole Thing in defer

The defer creation function guarantees that a brand new observable is produced for every subscriber. Without it, our counter and our percent$ Subject would be shared across subscriptions, polluting state. With it, each subscriber gets a fresh run with a fresh counter.

view the snippet

Here is the flow, step by step:

1. The array of observables is accepted as input.
2. Everything downstream of defer runs once for each subscriber (indicated by the callback at the top).
3. We create the fresh counter variable and the percent$ Subject inside that factory, guaranteeing isolation.
4. We map the original sources into a new array, decorating each with a finalize hook that computes the percentage and pushes it through the Subject as each source completes.
5. The decorated array is passed to the standard RxJS forkJoin. The tap after it ensures that once forkJoin resolves, we emit a final 100 to percent$ and close it.
6. The function returns a higher-order Observable that emits [finalResult$, percent$].

Summing Up

You can see the complete working example in this codepen.

forkJoinWithPercent

Leveraging It Inside an Angular Application

The utility has been distributed as an npm package — rxjs-toolbox — so integrating it into your own projects is straightforward.

For a practical demonstration, check out this Stackblitz Angular demo. It employs the package to drive a loading bar's percentage indicator:

Within the Angular framework, the integration is seamless.

Further References

If you wish to dive deeper into various RxJS operator use cases, the following resources are recommended:

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.

Enjoyed the read? Feel free to connect 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!