How to get results from functions that use concatMap in rxjs?

Question

I calling to two functions: fn1 and fn2. I use concatMap to invoke other after the other.

I don't use exhustMap and switchMap because they lead to nested "callback-hell".

  exhaustMap(() => 
     fn1().pipe(
       switchMap(() => fn2()))))

The one problem is how to get the results of fn1 and fn2 into next function that happens after fn2 is invoked?

stackblitz.com

import { of } from 'rxjs';
import { concatMap, tap } from 'rxjs/operators';
console.clear();

const fn1 = () => {
  console.log('in fn1');
  return of('fn1');
};

const fn2 = () => {
  console.log('in fn2');
  return of('fn2');
};

of(1)
  .pipe(
    concatMap(() => fn1()),
    concatMap(() => fn2()),
    tap({
      next: (a) => {
        console.log({ a }); /// <---- here I want to get fn1, fn2 from the functions.
        console.log('in tap!!!');
      },
      error: () => {
        console.log('in tap error!!!');
      },
      complete: () => {
        console.log('in tap complete');
      },
    })
  )
  
  .subscribe({
    next: (r) => {
      console.log({ r });
    },
    error: (e) => {
      console.log({ e });
    },
    complete: () => {
      console.log('complete');
    },
  });

Is it possible to run the functions in parallel? Then you can use [forkJoin](https://rxjs-dev.firebaseapp.com/api/index/function/forkJoin) operator to call them like `forkJoin([fn1(), fn2()]` instead of `concatMap`. The result array contains the result of both functions then. — daflodedeing, May 10 '22 at 10:55
no, they need to be execute one after the other. if the first failed then the second should not execute. — Jon Sud, May 10 '22 at 10:59

score 2 · Answer 1 · answered May 10 '22 at 11:57

I don't think there's an existing operator that could avoid the "operators-hell", but maybe you can build one that would?

Something like

import { from, ObservableInput, Observable } from "rxjs";
import { concatMap, map } from 'rxjs/operators';

const appendMap =
  <T, R extends ObservableInput<any>>(mapFn: (value: T) => R) =>
  (source$: Observable<T>) =>
    source$.pipe(
      concatMap((value) =>
        from(mapFn(value)).pipe(map((result) => [value, result] as const))
      )
    );

So now you can use it and avoid "operators-hell":

import { of } from 'rxjs';
import { tap } from 'rxjs/operators';

const fn1 = () => of('fn1');

const fn2 = () => of('fn2');

of(1)
  .pipe(
    appendMap(() => fn1()),
    map(([_, fn1Result]) => fn1Result),
    appendMap(() => fn2()),
    tap({
      next: ([fn1Result, fn2Result]) => {
        console.log({ fn1Result, fn2Result });
        console.log('in tap!!!');
      },
      error: () => {
        console.log('in tap error!!!');
      },
      complete: () => {
        console.log('in tap complete');
      },
    })
  )
  .subscribe({
    next: (r) => {
      console.log({ r });
    },
    error: (e) => {
      console.log({ e });
    },
    complete: () => {
      console.log('complete');
    },
  });

score 0 · Answer 2 · answered May 10 '22 at 10:59

0

You could just pipe in the 2^nd concatMap to the fn1() and use map to emit both the values.

of(1)
  .pipe(
    concatMap(() => 
      fn1().pipe(
        concatMap((fn1Value) => fn2().pipe(
          map((fn2Value) => ({
            fn1: fn1Value,
            fn2: fn2Value
          }))
        ))
      )
    )

I've adjusted your Stackblitz

answered May 10 '22 at 10:59

ruth

29,535
4
30
57

mmmmm, can I avoid "operators-hell" structure? – Jon Sud May 10 '22 at 11:02
@JonSud: It depends, if the observables `fn1()` and `fn2()` are independent of each other, you wouldn't even need the `concatMap`. You could use something `forkJoin` or `combineLatest`. They would emit all the emissions as an array. Please see [here](https://stackoverflow.com/a/63685990/6513921) for more info. – ruth May 10 '22 at 11:52

score 0 · Answer 3 · answered May 10 '22 at 11:15

0

The closest I can think of is here: Stackblitz

const fn1$ = of(1);
const fn2$ = of(2);

fn1$.pipe(
    concatWith(fn2$)
  ).pipe(
      bufferCount(2)
).subscribe(console.log);

answered May 10 '22 at 11:15

daflodedeing

319
3
11

score 0 · Answer 4 · answered May 10 '22 at 11:45

use zip:

stackblitz

import { of, zip } from 'rxjs';
import { concatMap, tap } from 'rxjs/operators';
console.clear();

const fn1$ = of('fn1');
const fn2$ = of('fn2');

zip(fn1$, fn2$)
  .pipe(
    tap({
      next: (a) => {
        console.log({ a }); // a === [fn1,fn2]
        console.log('in tap!!!');
      },
      error: () => {
        console.log('in tap error!!!');
      },
      complete: () => {
        console.log('in tap complete');
      },
    })
  )
  .subscribe({
    next: (r) => {
      console.log({ r });
    },
    error: (e) => {
      console.log({ e });
    },
    complete: () => {
      console.log('complete');
    },
  });

score 0 · Answer 5 · answered May 12 '22 at 12:14

I would do something like this

of(1)
  .pipe(
    concatMap(() => fn1()),
    concatMap((resp_1) => fn2().pipe(
      map(resp_2 => ({resp_1, resp_2})
    )),
    tap({
      next: ({resp_1, resp_2}) => {
        console.log(resp_1, resp_2); 
        console.log('in tap!!!');
      },
      error: () => {
        console.log('in tap error!!!');
      },
      complete: () => {
        console.log('in tap complete');
      },
    })
  )

A bit of operators hell but maybe not too much

How to get results from functions that use concatMap in rxjs?

5 Answers5