A forkJoin alternative for uncompleted observables?

Question

constructor(     private route: ActivatedRoute,     private http: Http ){     // Observe parameter changes     let paramObs = route.paramMap;      // Fetch data once     let dataObs = http.get('...');      // Subscribe to both observables,     // use both resolved values at the same level } 

Is there something similar to forkJoin that triggers whenever a parameter change is emitted? forkJoin only works when all observables have completed.

I just need to avoid callback hell, any alternative that complies is welcome.

Answer 1

There are several options:

1. Use take(1) with forkJoin() to complete each source Observable:

   forkJoin(o1$.take(1), o2$.take(1)) 

2. Use zip() and take(1) to emit only when all source Observables have emitted the same number of items:

   zip(o1$, o2$).take(1) 

3. Use combineLatest() to emit when any of the source Observables emit:

   combineLatest(o1$, o2$) 

Jan 2019: Updated for RxJS 6

Answer 2

A small trick to avoid breaking of observable subscriptions if any one of the observable fails.

 import { throwError, of, forkJoin } from "rxjs"; import { catchError, take } from "rxjs/operators";  //emits an error with specified value on subscription const observables$ = []; const observableThatWillComplete$ = of(1, 2, 3, 4, 5).pipe(take(1));  const observableThatWillFail$ = throwError(   "This is an error hence breaking the stream" ).pipe(catchError((error) => of(`Error Catched: ${error}`)));  observables$.push(observableThatWillComplete$, observableThatWillFail$);  forkJoin(observables$).subscribe(responses => {   console.log("Subscribed");   console.log(responses); });