Angular RxJS：Reduce 永远不会完成答案

【问题标题】：Angular RxJS: Reduce never completesAngular RxJS：Reduce 永远不会完成
【发布时间】：2019-05-07 20:04:48
【问题描述】：

我用两个按钮（searchButton、lazyButton）编写了一个组件。 ngOnDestroy 是：

public ngOnDestroy() {
    this.unsubscribe$.next();
    this.unsubscribe$.complete();
}

我从两个按钮点击事件中创建了两个可观察对象：

this.eager$ = Rx.Observable.fromEvent(this.searchButton, 'click')
    .takeUntil(this.$unsubscribe);

this.lazy$ = Rx.Observable.fromEvent(this.lazyButton, 'click')
    .takeUntil(this.$unsubscribe);

所以，当组件被销毁时，两个 observable 都会自动取消订阅。

另一方面，每次单击按钮时，我都需要发出一个 http 请求：

Observable
    .merge(this.eager$, this.lazy$)
    .switchMap(() => this.service.getItemsFromStorage())

所以，从现在开始，我会收到Observable<Response>。所以我需要处理响应：

Observable
    .merge(this.eager$, this.lazy$)
    .switchMap(() => this.service.getItemsFromStorage())
    .map(response => response.json())                        (****)
    .map(page => <Array<AdministrationUser>>page.included)   (****)

所以，在那之后，我收到了Observable<Array<AdministrationUser>>。

首先，我需要对每个AdministrationUser 应用一些转换。所以我需要创建一个Observable<AdministrationUser>:

Observable
    .merge(this.eager$, this.lazy$)
    .switchMap(() => this.service.getItemsFromStorage())
    .map(response => response.json())
    .map(page => <Array<AdministrationUser>>page.included)
    .switchMap(page => Observable.from(page.users))  (****)

所以，现在我可以修改每个用户了：

Observable
    .merge(this.eager$, this.lazy$)
    .switchMap(() => this.service.getItemsFromStorage())
    .map(response => response.json())
    .map(page => <Array<AdministrationUser>>page.included)
    .switchMap(page => Observable.from(page.users))
    .do(user => /*modify user*/)  (****)

所以在那之后，我只需要再次收集所有接收到的用户：

Observable
    .merge(this.eager$, this.lazy$)
    .switchMap(() => this.service.getItemsFromStorage())
    .map(response => response.json())
    .map(page => <Array<AdministrationUser>>page.included)
    .switchMap(page => Observable.from(page.users))
    .do(user => /*modify user*/)
    .reduce((acc, value) => [...acc, value], [])   (****)
    .do(users => this.rowsToShow = [...users])    (&&$$&&)
    .takeUntil(this.unsubscribe$)
    .subscribe();

一旦再次收集所有用户，他们必须分配到rowsToShow字段。

问题是(&&$$&&)上的代码从未到达过。

我猜是由于reduce 的行为是等待dor observable 结束。但是，我不知道为什么当我编码后它永远不会结束，一旦达到 http 响应，observable 就会变成.switchMap(page => Observable.from(page.users))。因此 observable 切换到数组元素的 observable ，当数组全部交叉时自动结束。

有什么想法吗？

【问题讨论】：

不会放弃旧版本的rxjs 是吗？

标签： angular typescript rxjs

【解决方案1】：

您的源 observable 是 .merge(this.eager$, this.lazy$)，这是一个 hot observable 并且永远不会完成，因此您的流不会完成尝试在合并后添加first()，这只会从源中获取一个发射并完成它

或者你可以把 reduce 放在内部 observable 中

.switchMap(page => Observable.from(page.users).reduce(....))

【讨论】：

这里是下次我点击哪个按钮，订阅完成就不会被抓到了……
您可以尝试第二个选项，它应该可以工作。它将 reduce 的源 observable 更改为 Observable.from

【解决方案2】：

原因是 reduce 等到父 observable 的最终发射后才弹出最终的归约值。由于您要合并的父可观察对象是点击事件，因此它们会继续发出值，直到达到阈值。

.switchMap => 除了解析它返回的 Observable 并提供值之外，它什么也不做。使用 switchMap 并不意味着发射的结束。每次返回一个新值时，switchMap 都会减少 Observable.from(page.users) 并返回该值。

而不是使用下面的，简单地取消对组件销毁的 Observables 的描述。你也可以改变 observable 如下：

.reduce((acc, value) => {this.rowsToShow = [...acc, value];
    return this.rowsToShow }, [])   (****)
    //.do(users => this.rowsToShow = [...users])    (&&$$&&)
    .takeUntil(this.unsubscribe$)
    .subscribe();

【讨论】：