如何在 RxJS observable 中添加额外的字段并等待它完成？

我有一个返回的可观察值items，我只需要一个特定的项目，并且我想从也可观察到的 http 请求向其添加一些额外的字段。下面的代码是我试图实现这一目标的方法，但它不起作用。我需要完成所有可观察量才能获取完整的项目数据ngOnInit。我缺少什么？

ngOnInit() {
    myItemsObservable$(this.store, items, items.data)
        .map(items => items.find(
            item => {
                return item.id === id
            }
        ))
        .concatMap(item => {
            return this.apiService.get(`/items/${item.id}/extradata`).map(extra => ({
                ...item,
                extra
            }))
        })
        .subscribe(item => {
            // I expect item to have extra fields here.
            this.item = item
        })

        // this.item here should already be complete.
}

让我们看一下代码的简化版本：

1  function ngOnInit() {
2    myItemsObservable$().subscribe(item => this.item = item);
3    console.log(this.item); // undefined
4  }

您本质上是在调用两个立即执行的函数。

第 2 行创建一个订阅对象，该对象启动可观察对象内的数据流。但是...在第 2 行之后执行并没有暂停。因此，在可观察对象内的异步任务完成之前，将执行第 3 行。这就是为什么this.item还是undefined在第 3 行。

希望您能明白为什么您的评论不正确：

// this.item here should already be complete.

您正在传递一个函数（item => this.item = item）到subscribe()当可观察到的排放发生时对其进行处理的方法。这是代码中实际具有发出值的位置。

所以，如果我们移动console.log()在订阅内，this.item将不再是未定义的：

1  function ngOnInit() {
2    myItemsObservable$().subscribe(item => {
3      this.item = item;
4      console.log(this.item); // not undefined :-)
5    });
6  }

要解决您问题的两个部分：

如何在 RxJS observable 中添加额外字段

你已经在这样做了。您已经使用过map and concatMap运算符将源可观察值发出的值修改为您想要的值。

...然后等待它完成？

好吧，你本身并不“等待”它。使用 RxJS，您可以定义数据流动的行为。您可以访问实际数据的唯一位置是订阅内部。

但是...您可以直接在代码的其他部分直接引用可观察量，而不是订阅，然后将数据从可观察量复制到另一个变量。

让我们将代码分成几个不同的部分，以便更容易地了解如何在不订阅的情况下引用不同的可观察源：

id$ = this.route.paramMap.pipe(  // This could come from a form control input
    params => params.get('id')   // or some other observable source.
);

allItems$ = myItemsObservable$(this.store, items, items.data);

getItem$(id) {
  return this.allItems$.pipe(
    map(items => items.find(i.id === id))
  );
}

getExtraData$(id) {
  return this.apiService.get(`/items/${id}/extradata`);
}

item$ = this.id$.pipe(
    switchMap(id => getItem$(id)),
    switchMap(item => this.getExtraData(item.id).pipe(
      map(extra => ({ ...item, ...extra }))
    ))
  );
}

看看如何定义item$开始于id$？这意味着每当id$发出新的值，item$会自动调用getItem$(), then getExtraData()然后发出这个新项目。我们不需要订阅就能实现这一点。

我们可以简单地定义一个可观察量以开始另一个可观察量.pipe()排放量并对其进行改造以满足我们的需求。

我们本质上设计了一个可观察的对象，只要商店中的商品发生变化，或者每当我们选择的商品发生变化时，它都会发出id$发出一个新的值。从某种意义上说，我们已经建立了item$代表我们的item并且它始终是最新的，包括附加“额外数据”。这是非常强大的。现在我们就可以使用它了。

注意以下的定义item$不需要在ngOnInit;它实际上可以直接连接到您的组件。

确实我们可以在我们的组件中订阅......但我们通常可以只使用AsyncPipe在模板中：

<div *ngIf="item$ | async as item">
  <h1>{{ item.name }}</h1>
  <ul>
    <li>{{ item.description  }}</li>
    <li>{{ item.someProperty }}</li>
  </ul>
</div>

如果您发现自己经常在组件中订阅，只是将数据复制到局部变量，只是为了供模板使用；我鼓励你停下来问问自己这真的有必要吗？大多数时候，您可以定义一个可观察的对象，它可以准确地发出您的视图所需的数据，而无需订阅。

RxJS 提供了许多运算符和静态函数，可以轻松创建具有各种常见行为的可观察对象。