Wait observable inside another observable

Question

I'm having a problem where I'm returning an observable and sometimes inside that observable I should get a value from another observable. I simplified my problem to much simpler case but main problem is still there. Consider following code that works:

 public dummyStream(): Observable<number> { return of(true).pipe( switchMap(isTrue => iif(() => isTrue === true, combineLatest([of([1,2,3,4,5]), of(2)]).pipe( map(([arrayOfNumbers, multiplier]) => { const results = arrayOfNumbers.map(num => { if (num !== 5) return num; else return 4; }); return results.reduce((prev, curr) => prev + curr, 0); }) ), combineLatest([of([1,2,3,4,5]), of(2)]).pipe( map(([arrayOfNumbers, multiplier]) => { return 0; }) ) ) ) ); }

So little explanation. It all starts with of(true). For simplicity in this example it's always true so once we get to iif() condition is always true.

Inside there I have combineLatest that combines two observables. Then eventually I do arrayOfNumbers.map and simply return that number again except in a case where number is 5 then I return 4.

Now what I'd like to do is to return of(num * multiplier) but then of course return type from map would by either number or Observable<number> and that's not working.

So this is same code but instead of returning number I'm returning Observable<number> in else block

 public dummyStream(): Observable<number> { return of(true).pipe( switchMap(isTrue => iif(() => isTrue === true, combineLatest([of([1,2,3,4,5]), of(2)]).pipe( map(([arrayOfNumbers, multiplier]) => { const results = arrayOfNumbers.map(num => { if (num !== 5) return num; else of(num * multiplier); }); return results.reduce((prev, curr) => prev + curr, 0); }) ), combineLatest([of([1,2,3,4,5]), of(2)]).pipe( map(([arrayOfNumbers, multiplier]) => { return 0; }) ) ) ) ); }

Now how should I change this so that dummyStream() return type would still be Observable<number> but so that inside my else block I'm using another observable?

You don't need of, just do return num * multiplier. Inside the map, you don't want to return Observable, you are modifying the emission, so just return the plain value. — BizzyBob
– BizzyBob, Commented Jul 26, 2022 at 17:06
In a real world case I definitely need to use observable at some point because I don't have a plain value at that point. Using of(num * multiplier) is just an example here to keep things simple. — kivikall
– kivikall, Commented Jul 26, 2022 at 17:15
oh, I see, then you need to use a Higher Order Mapping Operator, like switchMap instead of your plain map. — BizzyBob
– BizzyBob, Commented Jul 26, 2022 at 17:16

Tobias S. · Accepted Answer · 2022-07-26 17:17:56Z

I would restructure dummyStream like this:

function dummyStream() { return of(true).pipe( switchMap(isTrue => iif(() => isTrue === true, combineLatest([of([1,2,3,4,5]), of(2)]).pipe( switchMap(([arrayOfNumbers, multiplier]) => { return forkJoin(arrayOfNumbers.map(num => { if (num !== 5) return of(num); else return of(num * multiplier); })); }), map((results) => results.reduce((prev, curr) => prev + curr, 0)) ), combineLatest([of([1,2,3,4,5]), of(2)]).pipe( map(([arrayOfNumbers, multiplier]) => { return 0; }) ) ) ) ); }

So instead of just return num in the map, you return of(num) so that you end up with an array of Observable<number>. Change the outer map to a switchMap and wrap the resulting array in a forkJoin to wait for all inner observables to finish. You can put the reduce in its own map afterwards.

Playground

Thanks! This was definitely what I was looking for! I don't know why after all these years I'm still pretty lost sometimes with RxJS. Anyway, now I just need to apply this to my real problem. After all it's more or less the same, just much bigger spaghetti :D

Eddy Lin · Accepted Answer · 2022-08-19 06:20:03Z

Is this what you need?

const data1$ = of([1, 2, 3, 4, 5]).pipe(concatAll()); // const data1$ = from([1, 2, 3, 4, 5]); const data2$ = of(2).pipe(repeat()); const source$ = zip(data1$, data2$).pipe( map(([num, multiplier]) => num !== 5 ? num : num * multiplier), reduce((prev, curr) => prev + curr, 0), ); source$.subscribe(console.log); const falseResult$ = of(0); iif(() => true, source$, falseResult$).subscribe(console.log);

Collectives™ on Stack Overflow

Wait observable inside another observable

2 Answers 2

1 Comment

Comments

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

1 Comment

Comments

Related