Modify each item in an observable list and return the modified list again as an observable

user_401025 0
Viewed 889

A find method has following signiture 

find(): Observable<User[]>

It returns an observable that emits a list of users, that may look like this:

[{ user_1 }, { user_2 }]

My goal is to run an async operation on each user from the list. That async operation should modify the element by adding some fields. At the end the list of updated users should be emitted. This is what I have tried:

return this.service.find()
    .mergeMap((users: User[]) => Observable.from(users))
    .mergeMap((user: User) => {
        return this.anotherService.getAdditionalData(user.id)
            .map((res: any) => {
                // return edited user
            })
    })
    // How to return User[] here?

With the first mergeMap an observable for each user in the list is created. Then for each user observable another async operation is made. The async operation returns the edited user.

My question is, at the end, how can I return the user list with the edited users. Or how can I collect the edited users in a list and return this list?

rxjs javascript
2 Answers

There is .toArray() operator for this:

function modify(v) {
  return Rx.Observable.of('modified: ' + v)
    .delay(Math.random() * 1000);
}

const users = Array.from({ length: 10 })
  .map((_, i) => i);

Rx.Observable.from(users)
  .mergeMap(v => modify(v))
  .toArray()
  .subscribe(e => console.log(e));
<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

user_37723790

I think this does what you want:

function asyncOperation(user) {
  return Rx.Observable
  .timer(3000 * Math.random())
  .map(() => {
    user.isModified = true;
    return user;
  }).do((x) => { console.log('async', x); });
}

const users = [
  { id: 1, name: 'First' },
  { id: 2, name: 'Second' },
  { id: 3, name: 'Third' }
];

Rx.Observable.of(users)
 .switchMap((u) => Rx.Observable.forkJoin(u.map(x => asyncOperation(x))))
 .subscribe((x) => { console.log('final', x); });
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.2/Rx.js"></script>

It assumes that the observable returned from your asyncOperation completes. It uses the forkJoin operator to emit the last value from each observable when they all complete. You could still use mergeMap/flatMap but I like switchMap since it will only keep one subscription to the internal observable. So if the source emits again then you will only get the result of the latest.

This solution does not require the source observable (this.service.find()) to complete like using the toArray operator does.

user_9597450
Related