当组件被销毁时，如何让我的链式可观察计时器停止执行？

Question

以下是模仿我的代码的基本可观察对象：

class DataService {
    #data = this.http.get<Data[]>('https://example.com/getData').pipe(
        timeout(15000),
        catchError(err => of([]))
    )
    
    #adaptedData = this.#data.pipe(
        map(data => data.map(obj => new AdaptedObj(obj)))
    )

    #pollAdapted = timer(0, 60000).pipe(
        switchMap(() => this.#adaptedData),
        shareReplay(1)
    )

    get get() { return this.#adaptedData }
    get poll() { return this.#pollAdapted }
}

然后在我的代码的其他地方，我有一个额外的可观察层，可以进一步操纵数据以用于特定的组件集。示例：

class DataAnalyzerService {
    constructor(private dataService: DataService){}
    
    #analyzedDataStore = new BehaviorSubject<AnalyzedData[]>([])
    get analyzedData() { return this.#analyzedDataStore.value }

    #analyzedData = this.dataService.poll.pipe(
        filter(objs => objs.length > 0),
        map(objs => objs.map(obj => new AnalyzedObj(obj))),
        tap(analyzedData => this.#analyzedDataStore.next(analyzedData))
    )

    #metricCalculator = () => {
        const data = this.analyzedData
        // Do calculations and filtration
        return dashboardMetricsArr
    }

    #dashboardStats = this.#analyzedData.pipe(
        switchMap(() => of(this.#metricCalculator))
    )

    get dashboardStats() { return this.#dashboardStats }
}

然后在我的组件中，我将仪表板指标与异步管道一起使用，如下所示：

<ng-container *ngIf="(dashboardStats | async) as stats">
    // Use the Data
<ng-container>

我之所以将可观察对象分解成这样，是因为有不同的组件需要在不同分析层的数据，此外，一些组件只需要数据一次，而其他组件则需要不断更新1 分钟间隔。

目前，应用程序工作正常，但是当我注销或退出使用数据的组件时，我注意到 http 请求每分钟都在进行，因为 DataService poll 可观察值永远不会取消订阅.现在，我只有一个使用模板中的异步管道初始化的订阅，所以当我离开该组件或一起注销时，我知道我正在关闭该特定订阅。此外，我已经验证 #analyzedData 和 #dashboardStats observables 在导航离开后都停止执行。只有 poll observable 继续。

关于如何停止轮询直到我回到需要它的组件有什么想法吗？

Answer 1

有几个选项

使用下面的library
在组件中实例化 Subject，如果使用 takeUntil operator
取消订阅，则应该用于检查
private destroy$ = new Subject<void>();

this.data$.pipe(takeUntil(this.destroy$))

ngOnDestroy(): void { this.destroy$.next(); this.destroy$.complete(); }

Answer 2

shareReplay 具有 refCount 配置选项，如 manual:

中所述

As of RXJS version 6.4.0 a new overload signature was added to allow for manual control over what happens when the operators internal reference counter drops to zero. If refCount is true, the source will be unsubscribed from once the reference count drops to zero, i.e. the inner ReplaySubject will be unsubscribed. All new subscribers will receive value emissions from a new ReplaySubject which in turn will cause a new subscription to the source observable. If refCount is false on the other hand, the source will not be unsubscribed meaning that the inner ReplaySubject will still be subscribed to the source (and potentially run for ever).

您使用shareReplay(1)，它使用默认值refCount: false，因此不会退订。为自动退订开启引用计数：

shareReplay({bufferSize: 1, refCount: true})

当组件被销毁时，如何让我的链式可观察计时器停止执行？

How can I make my chained observable timer stop executing when component is destroyed?

angular

typescript

rxjs

observable

async-pipe