RxJS将数组映射到可观察并返回数组中的普通对象 [英] RxJS Map array to observable and back to plain object in array

问题描述

我有一个对象数组，我需要从中将每个对象分别传递到异步方法中(后面的过程由Promise处理，然后通过Observable.fromPromise(...)转换回Observable －因为在相同的方法中使用了相同的方法如果只是随时传递单个对象；该过程会将对象保存到数据库中).例如，这是一个对象数组:

I have an array of objects from which I need to pass each object separately into async method (process behind is handled with Promise and then converted back to Observable via Observable.fromPromise(...) - this way is needed because the same method is used in case just single object is passed anytime; the process is saving objects into database). For example, this is an array of objects:

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

现在，我有了一种称为insert which的方法，该方法将对象插入数据库.数据库实例中的store方法返回新创建的ID.最后，初始对象被复制并映射为其新的ID:

Now I have the method called insert which which inserts object into database. The store method from database instance returns newly created id. At the end the initial object is copied and mapped with its new 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;
    }
  );
}

这在插入单个对象的情况下效果很好.但是，我想增加对插入多个对象的支持，这些对象只是调用一次插入的方法.目前，这是我所拥有的，但是它不起作用:

This works well in case I insert single object. However, I would like to add support for inserting multiple objects which just call the method for single insert. Currently this is what I have, but it doesn't work:

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

insertAll方法正在按预期插入用户(或其他填充该用户的数据库)，但是我没有从中得到任何响应.我正在调试正在发生的事情，似乎forkJoin仅从第一个映射的用户那里得到响应，但是其他人被忽略了.订阅insertAll不会执行任何操作，也不会通过捕获insertAll或通过订阅insertAll中的第二个参数而发生任何错误.

The insertAll method is inserting users as expected (or something else filled up the database with that users), but I don't get any response back from it. I was debugging what is happening and seems that forkJoin is getting response just from first mapped user, but others are ignored. Subscription to insertAll does not do anything, also there is no any error either via catch on insertAll or via second parameter in subscribe to insertAll.

所以我正在寻找一个解决方案，其中Observable(在insertAll中)将以这种形式向用户发出一系列新对象:

So I'm looking for a solution where the Observable (in insertAll) would emit back an array of new objects with users in that form:

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

对于任何指向正确方向的建议，我将感到非常高兴.预先感谢！

I would be very happy for any suggestion pointing in the right direction. Thanks in advance!

推荐答案

要从数组转换为可观察值，可以使用

To convert from array to observable you can use Rx.Observable.from(array).

要从可观察对象转换为数组，请使用 obs.toArray() .请注意，这确实返回了一个可观察的数组，因此您仍然需要.subscribe(arr => ...)才能将其取出.

To convert from observable to array, use obs.toArray(). Notice this does return an observable of an array, so you still need to .subscribe(arr => ...) to get it out.

也就是说，您的代码与 forkJoin 确实看起来正确.但是，如果您确实想尝试from，请编写如下代码:

That said, your code with forkJoin does look correct. But if you do want to try from, write the code like this:

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

另一种类似的接收方法是在完成值时发出值，而不是像forkJoin或 toArray 来自上一个示例，我们得到了它:

Another more rx like way to do this would be to emit values as they complete, and not wait for all of them like forkJoin or toArray does. We can just omit the toArray from the previous example and we got it:

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

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

As @cartant mentioned, the problem might not be in Rx, it might be your database does not support multiple connections. In that case, you can replace the mergeMap with concatMap to make Rx send only 1 concurrent request:

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

这篇关于RxJS将数组映射到可观察并返回数组中的普通对象的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！