RxJS 将数组映射到可观察对象并返回到数组中的普通对象

RxJS Map array to observable and back to plain object in array

我有一个对象数组,我需要将每个对象分别传递给异步方法(后面的过程用 Promise 处理,然后通过 Observable.fromPromise(...) 转换回 Observable - 这种方式是必需的,因为相同的方法用于任何时候只传递单个对象的情况;该过程将对象保存到数据库中)。例如,这是一个对象数组:

[
  {
    "name": "John",
    ...
  },
  {
    "name": "Anna",
    ...
  },
  {
    "name": "Joe",,
    ...
  },
  {
    "name": "Alexandra",
    ...
  },
  ...
]

现在我有了名为 insert which 的方法,它将对象插入到数据库中。 store 方法从数据库实例returns 新创建的id。最后,初始对象被复制并映射到它的新 id:

insert(user: User): Observable<User> {
  return Observable.fromPromise(this.database.store(user)).map(
    id => {
      let storedUser = Object.assign({}, user);
      storedUser.id = id;
      return storedUser;
    }
  );
}

这在我插入单个对象时效果很好。但是,我想添加对插入多个对象的支持,这些对象只调用单次插入的方法。目前这是我所拥有的,但它不起作用:

insertAll(users: User[]): Observable<User[]> {
  return Observable.forkJoin(
    users.map(user => this.insert(user))
  );
}

insertAll 方法正在按预期插入用户(或者用其他方法用该用户填满数据库),但我没有从它那里得到任何响应。我正在调试正在发生的事情,似乎 forkJoin 只是从第一个映射用户那里得到响应,但其他人被忽略了。订阅 insertAll 不会做任何事情,通过 insertAll 上的 catch 或通过订阅 insertAll.

中的第二个参数也没有任何错误

所以我正在寻找一种解决方案,其中 Observable(在 insertAll 中)会以这种形式发回一组新的用户对象:

[
  {
    "id": 1,
    "name": "John",
    ...
  },
  {
    "id": 2,
    "name": "Anna",
    ...
  },
  {
    "id": 3,
    "name": "Joe",,
    ...
  },
  {
    "id": 4,
    "name": "Alexandra",
    ...
  },
  ...
]

如果有任何指向正确方向的建议,我将非常高兴。提前致谢!

要从数组转换为可观察对象,您可以使用 Rx.Observable.from(array).

要从 observable 转换为数组,请使用 obs.toArray()。请注意,这 return 是一个数组的可观察对象,因此您仍然需要 .subscribe(arr => ...) 才能将其取出。

就是说,您使用 forkJoin 的代码看起来是正确的。但如果你确实想尝试 from,请编写如下代码:

insertAll(users: User[]): Observable<User[]> {
  return Observable.from(users)
    .mergeMap(user => this.insert(user))
    .toArray();
}

另一种更像 rx 的方法是在值完成时发出值,而不是像上一个示例中的 forkJointoArray does. We can just omit the toArray 那样等待所有值,我们得到了它:

insertAll(users: User[]): Observable<User> {
  return Observable.from(users)
    .mergeMap(user => this.insert(user));
}

如@cartant 所述,问题可能不在 Rx 中,可能是您的数据库不支持多连接。在这种情况下,您可以替换 mergeMap with concatMap 以使 Rx 仅发送 1 个并发请求:

insertAll(users: User[]): Observable<User[]> {
  return Observable.from(users)
    .concatMap(user => this.insert(user))
    .toArray(); // still optional
}