Observable map property based on another observable

Question

Lets say i have an observable that emits employees.

$employees

I wish to manipulate a sub property of each employee based on an Observable. For the example lets say display name.

Currently im performing the task like this.

const getDisplayName = (emp) => {}; //returns an observable

const mapFn = async (emp) => {
    emp.displayName = await getDisplayName(emp).toPromise();

    return emp;
}

$employees
    .pipe(mergeMap(mapFn));

I think my confusion is that, to my understanding, we have two streams. The root $employees and the getDisplayName. My understanding is with the various merge operators the root value would be replaced by the value of the secondary stream. Not merged with.

Is there a better way to do this where i don't need to convert to a promise but can also just map a property of the employee?

Thanks.

Does this answer your question: https://stackoverflow.com/questions/44004144/how-to-wait-for-two-observables-in-rxjs? — DaggeJ, Apr 22 '20 at 08:56
I think you do not need to convert to Promise. This `$employees .pipe(mergeMap(getDisplayName))` should work — Picci, Apr 22 '20 at 12:49

score 1 · Answer 1 · answered Apr 22 '20 at 12:58

This should work:

const emp$ = from([1, 2, 3]);

function getName(emp: number): Observable<string> {
    return of(emp.toString());
}

const empNameTuple$ = emp$.pipe(mergeMap(emp => {
    return getName(emp).pipe(map(name => [emp, name]));
}))

Here I'm just returning a tuple of [number, string], but you can map it however you wish.

bryan60 · Accepted Answer · 2020-04-22T13:27:13.237

you want to do:

$employees
    .pipe(mergeMap(employees => {
      return forkJoin(employees.map(emp => getDisplayName(emp).pipe(
        map(displayName => ({...emp, ...{displayName}}))
      )))
    }));

if you really want a broken out mapFn:

const mapFn = employees => {
  return forkJoin(employees.map(emp => getDisplayName(emp).pipe(
    map(displayName => ({...emp, ...{displayName}}))
  )));
}

$employees
    .pipe(mergeMap(mapFn))

always advise against mixing rxjs and async / await. They're different methods of handling async operations and don't play nice. mergeMap needs you to return an observable, forkJoin executes observables in parralel, so you join all your employees mapped into their getName functions and then map the name into the original employee and return it.

EDIT: the above is if $eployees is emitting an array of employees. if it's just a single employee, do:

const mapFn = emp => {
  return getDisplayName(emp).pipe(
    map(displayName => ({...e, ...{displayName}}))
  );
}

however, if it's emitting a single employee (or an array of employees) multiple times, it's important to understand the implications of using mergeMap vs switchMap vs concatMap. Let me know if that is the case.

Observable map property based on another observable

2 Answers2