订阅嵌套的 Observable

Question

我有一个应用程序发出一个 http 请求以获取项目列表，然后对列表中的每个项目发出一个 http 请求以获取有关每个项目的更多详细信息。有效：

class ItemsService {
  fetchItems() {
    return this.http.get(url)
    .map(res => res.json())
    .map(items => items.map(this.fetchItem(item)));
  }

  fetchItem(item: Item) {
    this.http.get(`${url}/${item.id}`)
    .map(res => res.json());
  }
}

然后我会做类似itemsService.fetchItems().subscribe(items => console.log(items)) 的事情，但最终发生的是我得到了一系列可观察的数据（来自fetchItem 的每个响应）。我还需要订阅每个内部可观察对象，以便真正触发 fetchItem 请求。

我也尝试过使用flatMap 代替地图，但在这种情况下似乎有相同的结果。有没有办法订阅嵌套的 observable？

Answer 1

我会这样做：

function mockRequest() {
    return Observable.of('[{"id": 1}, {"id": 2}, {"id": 3}]');
}
function otherMockRequest(id) {
    return Observable.of(`{"id":${id}, "desc": "description ${id}"}`);
}

class ItemsService {
    fetchItems() {
        return mockRequest()
            .map(res => JSON.parse(res))
            .concatAll()
            .mergeMap(item => this.fetchItem(item));
    }

    fetchItem(item: Item) {
        return otherMockRequest(item.id)
            .map(res => JSON.parse(res));
    }
}

let service = new ItemsService();
service.fetchItems().subscribe(val => console.log(val));

观看现场演示：http://plnkr.co/edit/LPXfqxVsI6Ja2J7RpDYl?p=preview

我正在使用.concatAll() 的技巧将对象数组（例如[{"id": 1}, {"id": 2}, {"id": 3}]）转换为一个一个发出的单独值{"id": 1}、{"id": 2} 和{"id": 3}（到目前为止，它是一个未记录的功能）。然后我使用mergeMap() 在单独的请求中获取它们的内容并将其结果合并到运算符链中。

这个 plnkr 示例打印到控制台：

{ id: 1, desc: 'description 1' }
{ id: 2, desc: 'description 2' }
{ id: 3, desc: 'description 3' }

Answer 2

你可能遇到的问题是你没有压扁。

flatMap 或mergeMap 将变平Observables、Promises、Arrays，甚至generators（不要在最后一个引用我的话），几乎任何你想扔给它的东西。

所以当你做.flatMap(items => items.map(item => this.fetchItem(item))时，你真的只是在做Observable<Array<Item>> => Observable<Observable<Item>>

当你只是做map时，你就是在做Observable<Array<Item>> => Observable<Array<Observable<Item>>>。

你需要做的是首先展平数组，然后展平每个请求：

class ItemsService {
  fetchItems() {
    return this.http.get(url)
    .map(res => res.json())
    // Implicitly map Array into Observable and flatten it
    .flatMap(items => items)
    // Flatten the response from each item
    .flatMap((item: Item) => this.fetchItem(item));
  }
}

现在，如果您不介意单独获取每个项目响应，则上述方法有效。如果您需要获取所有项目，那么您应该在所有内部值上使用forkJoin，但您仍然需要flatMap 以展平生成的内部值：

fetchItems(): Observable<Response[]> {
  return this.http.get(url)
    .map(res => res.json())
    .flatMap(items => {
       const requests = items.map(item => this.fetchItem(item));
       return Rx.Observable.forkJoin(requests);
    });
}

Answer 3

您可以在调用this.fetchItem 的行之前分解 items 数组。 你可以在一个 Observable 上使用 mergeMap，它的值是一个数组，每个项目将被单独发射。

fetchItems() {
    return this.http.get(url)
       .map(res => res.json())
       .mergeMap(arrItem => this.fetchItem(arrItem));
}

编辑：我想我应该提供更多解释。 mergeMap 与 rxjs 中的 flatMap 同义。通常，当您的投影函数返回一个 Observable 时，您使用 flatMap，但它也会展平数组，因此，调用 mergeMap 将分别发出每个项目，我认为这就是 OP 想要实现的。我还意识到，您可以将mergeMap 调用和最后一个map 调用结合起来，因为mergeMap 的投影将为数组中的每个项目调用，我在上面的代码中进行了更改。