修改可观察列表中的每个项目，并将修改后的列表作为可观察列表再次返回答案

【问题标题】：Modify each item in an observable list and return the modified list again as an observable修改可观察列表中的每个项目，并将修改后的列表作为可观察列表再次返回
【发布时间】：2017-11-21 23:27:53
【问题描述】：

查找方法具有以下签名

find(): Observable<User[]>

它返回一个发出用户列表的可观察对象，可能如下所示：

[{ user_1 }, { user_2 }]

我的目标是对列表中的每个用户运行异步操作。该异步操作应该通过添加一些字段来修改元素。最后应该发出更新的用户列表。这是我尝试过的：

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?

使用第一个 mergeMap 为列表中的每个用户创建一个 observable。然后为每个可观察的用户进行另一个异步操作。异步操作返回编辑后的用户。

最后，我的问题是，如何返回已编辑用户的用户列表。或者如何将编辑后的用户收集到一个列表中并返回这个列表？

【问题讨论】：

标签： javascript rxjs

【解决方案1】：

我认为这是你想要的：

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); });

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.2/Rx.js"&gt;&lt;/script&gt;

它假定从您的 asyncOperation 返回的 observable 已完成。它使用forkJoin 运算符在每个可观察对象都完成时发出最后一个值。你仍然可以使用mergeMap/flatMap，但我喜欢switchMap，因为它只会保留一个对内部可观察对象的订阅。因此，如果源再次发出，那么您只会得到最新的结果。

此解决方案不需要像使用 toArray 运算符那样完成源 observable (this.service.find())。

【讨论】：

非常优雅的解决方案。谢谢@bygrace！

【解决方案2】：

这里有.toArray() 运算符：

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));

&lt;script src="https://unpkg.com/rxjs/bundles/Rx.min.js"&gt;&lt;/script&gt;

【讨论】：