Chaining multiple observables and reusing results of already executed ones in the chain

Question

I want to combine the following functions a(), b() and c() to get one result.

function a() {
    return rx.of(42);
}

function b(value) {
    return rx.of(value*2);
}

function c(val1, val2) {
    return rx.of({
        original: val1,
        modified: val2
    })
}

These are simplified variants - they would actually be HTTP-requests. The result of a is needed for executing b, a and bs results are preconditions for executing c. The only thing I can think of is just wrapping them all in .subscribe() calls, but that just doesn't look right. Yet I cannot think of a better way to solve this:

// bad
new Observable(observer => {
    a().subscribe(
        result_a => {
            b(result_a).subscribe(
                result_b => {
                    c(result_a, result_b).subscribe(
                        result_c => {
                            observer.next(c);
                        }
                    )
                }
            )
        }
    )    
}).subscribe(
    result => console.log(result)
)

I do know that there are operators such as pipe, concat and more but I do not know how to apply them in this specific situation where the result of one Observable is still needed for another one that is executed later on. How could I simplify this by using the proper RxJS techniques?

user_5699993 · Answer

What you can try is something like this, comments are added inline

a().pipe(
  // concatMap ensures that the order of notification is maintained, i.e. a() emits its next value only after b() has finished processing the previous one
  concatMap(valA => b(valA).pipe(
    // via this map we can return both the result of b() and also the parameter received, i.e. the result of a()
    map(valB => {
       return {valA, valB}
    })
  ),
  // here again, with concatMap, we ensure the order of notification is maintained
  concatMap(({valA, valB}) => c(valA, valB))
)

This is a typical pattern that may be encountered many times while using RxJS with http.

If you want to see some more of such patterns, you can read this article.

user_135786 · Answer

The direct equivalent of your example there would be with mergeMap:

a().pipe(
  mergeMap(a =>
    b(a).pipe(
      mergeMap(b => c(a, b))
    )
  )
)
.subscribe(console.log)

If order is important (once these are something that is asynchronous and not just made with of), then you will want to use concatMap instead. But the overall structure is the same.

user_3738171 · Answer

Given the a and b observables complete, you could use the forkJoin operator like this:

const { of, forkJoin } = rxjs;
const { mergeMap } = rxjs.operators;

function a() {
    return of(42);
}

function b(value) {
    return of(value*2);
}

function c(val1, val2) {
    return of({
        original: val1,
        modified: val2
    })
}

const init = a();

forkJoin(
    init, 
    init.pipe(mergeMap(b))
).pipe(mergeMap(([a, b]) => c(a, b)))
.subscribe(console.log);

Chaining multiple observables and reusing results of already executed ones in the chain

3 Answers