I'm trying to use forkJoin on two Observables. One of them starts as a stream... If I subscribe to them directly I get a response, forkJoin isn't firing though. Any ideas?
private data$: Observable<any[]>;
private statuses$: Observable<any[]>;
private queryStream = new Subject<string>();
....
this.data$ = this.queryStream
.startWith('')
.flatMap(queryInput => {
this.query = queryInput
return this._companyService.getCompanies(this.queryRequired + ' ' + this.query, this.page, this.sort);
})
.share();
...
Observable.forkJoin(this.statuses$, this.companies$)
.subscribe(res => {
console.log('forkjoin');
this._countStatus(res[0], res[1]);
});
// This shows arrays in the console...
this.statuses$.subscribe(res => console.log(res));
this.companies$.subscribe(res => console.log(res));
// In the console
Array[9]
Array[6]
forkJoin emits only when all inner observables have completed.
If you need an equivalent of forkJoin that just listens to a single emission from each source, use combineLatest + take(1)
combineLatest(
this.statuses$,
this.companies$,
)
.pipe(
take(1),
)
.subscribe(([statuses, companies]) => {
console.log('forkjoin');
this._countStatus(statuses, companies);
});
As soon as both sources emit, combineLatest will emit and take(1) will unsubscribe immediately after that.
A very common problem with forkJoin is that it requires all source Observables to emit at least one item and all of them have to complete.
In other words if this.statuses$ or this.companies$ doesn't emit any item and until they both complete the forkJoin won't emit anything.
this.statuses$.subscribe(
res => console.log(res),
undefined,
() => console.log('completed'),
);
The forkJoin didn't work so I used the below code to solve my problem. With mergeMap you can map the result of the outer subscription to the inner subscription and subscribe to it as you wish
this.statuses$.pipe(
mergeMap(source => this.companies$.pipe(
map(inner => [source , inner])
)
)
).subscribe(([e , r]) => {
console.log(e , r);
})