Learn RxJS
Search…
forkJoin
signature: forkJoin(...args, selector : function): Observable
forkJoin(...args, selector : function): ObservableWhen all observables complete, emit the last emitted value from each.
⚠ If an inner observable does not complete
forkJoin will never emit a value!Why use forkJoin?
forkJoin?This operator is best used when you have a group of observables and only care about the final emitted value of each. One common use case for this is if you wish to issue multiple requests on page load (or some other event) and only want to take action when a response has been received for all. In this way it is similar to how you might use
Promise.all.Be aware that if any of the inner observables supplied to
forkJoin error you will lose the value of any other observables that would or have already completed if you do not catch the error correctly on the inner observable. If you are only concerned with all inner observables completing successfully you can catch the error on the outside.It's also worth noting that if you have an observable that emits more than one item, and you are concerned with the previous emissions
forkJoin is not the correct choice. In these cases you may be better off with an operator like combineLatest or zip.Examples
Example 1: Using a dictionary of sources to make AJAX request
1
// RxJS v6.5+
2
import { ajax } from 'rxjs/ajax';
3
import { forkJoin } from 'rxjs';
4
5
/*
6
when all observables complete, provide the last
7
emitted value from each as dictionary
8
*/
9
forkJoin(
10
// as of RxJS 6.5+ we can use a dictionary of sources
11
{
12
google: ajax.getJSON('https://api.github.com/users/google'),
13
microsoft: ajax.getJSON('https://api.github.com/users/microsoft'),
14
users: ajax.getJSON('https://api.github.com/users')
15
}
16
)
17
// { google: object, microsoft: object, users: array }
18
.subscribe(console.log);
Copied!
Example 2: Observables completing after different durations
1
// RxJS v6+
2
import { interval, forkJoin, of } from 'rxjs';
3
import { delay, take } from 'rxjs/operators';
4
5
const myPromise = val =>
6
new Promise(resolve =>
7
setTimeout(() => resolve(`Promise Resolved: ${val}`), 5000)
8
);
9
10
/*
11
when all observables complete, give the last
12
emitted value from each as an array
13
*/
14
const example = forkJoin({
15
//emit 'Hello' immediately
16
sourceOne: of('Hello'),
17
//emit 'World' after 1 second
18
sourceTwo: of('World').pipe(delay(1000)),
19
//emit 0 after 1 second
20
sourceThree: interval(1000).pipe(take(1)),
21
//emit 0...1 in 1 second interval
22
sourceFour: interval(1000).pipe(take(2)),
23
//promise that resolves to 'Promise Resolved' after 5 seconds
24
sourceFive: myPromise('RESULT')
25
});
26
/*
27
* Output:
28
* {
29
* sourceOne: "Hello",
30
* sourceTwo: "World",
31
* sourceThree: 0,
32
* sourceFour: 1,
33
* sourceFive: "Promise Resolved: RESULT"
34
* }
35
*/
36
const subscribe = example.subscribe(val => console.log(val));
Copied!
Example 3: Making a variable number of requests (uses deprecated API)
1
// RxJS v6+
2
import { mergeMap } from 'rxjs/operators';
3
import { forkJoin, of } from 'rxjs';
4
5
const myPromise = val =>
6
new Promise(resolve =>
7
setTimeout(() => resolve(`Promise Resolved: ${val}`), 5000)
8
);
9
10
const source = of([1, 2, 3, 4, 5]);
11
//emit array of all 5 results
12
const example = source.pipe(mergeMap(q => forkJoin(...q.map(myPromise))));
13
/*
14
output:
15
[
16
"Promise Resolved: 1",
17
"Promise Resolved: 2",
18
"Promise Resolved: 3",
19
"Promise Resolved: 4",
20
"Promise Resolved: 5"
21
]
22
*/
23
const subscribe = example.subscribe(val => console.log(val));
Copied!
Example 4: Handling errors on outside
1
// RxJS v6+
2
import { delay, catchError } from 'rxjs/operators';
3
import { forkJoin, of, throwError } from 'rxjs';
4
5
/*
6
If any inner observables error, the error result
7
will be emitted by catchError.
8
*/
9
const example = forkJoin({
10
// emit 'Hello' immediately
11
sourceOne: of('Hello'),
12
// emit 'World' after 1 second
13
sourceTwo: of('World').pipe(delay(1000)),
14
// throw error
15
sourceThree: throwError('This will error')
16
}).pipe(catchError(error => of(error)));
17
18
// output: 'This will Error'
19
const subscribe = example.subscribe(val => console.log(val));
Copied!
Example 5: Getting successful results when one inner observable errors
1
// RxJS v6+
2
import { delay, catchError } from 'rxjs/operators';
3
import { forkJoin, of, throwError } from 'rxjs';
4
5
/*
6
Emit values from successfully completed
7
inner observables.
8
*/
9
const example = forkJoin({
10
// emit 'Hello' immediately
11
sourceOne: of('Hello'),
12
// emit 'World' after 1 second
13
sourceTwo: of('World').pipe(delay(1000)),
14
// throw error
15
sourceThree: throwError('This will error').pipe(catchError(error => of(error)))
16
});
17
18
/*
19
* Output:
20
* {
21
* sourceOne: "Hello",
22
* sourceTwo: "World",
23
* sourceThree: "This will error"
24
* }
25
*/
26
const subscribe = example.subscribe(val => console.log(val));
Copied!
Example 6: forkJoin in Angular
1
@Injectable()
2
export class MyService {
3
makeRequest(value: string, delayDuration: number) {
4
// simulate http request
5
return of(`Complete: ${value}`).pipe(
6
delay(delayDuration)
7
);
8
}
9
}
10
11
@Component({
12
selector: 'my-app',
13
template: `
14
<div>
15
<h2>forkJoin Example</h2>
16
<ul>
17
<li> {{propOne}} </li>
18
<li> {{propTwo}} </li>
19
<li> {{propThree}} </li>
20
</ul>
21
</div>
22
`,
23
})
24
export class App {
25
public propOne: string;
26
public propTwo: string;
27
public propThree: string;
28
constructor(private _myService: MyService) {}
29
30
ngOnInit() {
31
// simulate 3 requests with different delays
32
forkJoin({
33
requestOne: this._myService.makeRequest('Request One', 2000),
34
requestTwo: this._myService.makeRequest('Request Two', 1000),
35
requestThree: this._myService.makeRequest('Request Three', 3000)
36
})
37
.subscribe(({requestOne, requestTwo, requestThree}) => {
38
this.propOne = requestOne;
39
this.propTwo = requestTwo;
40
this.propThree = requestThree;
41
});
42
}
43
}
Copied!
Additional Resources
Last modified 1yr ago