如何捕获RxJS switchMap丢弃内部Observable事件以维护任务计数器？

我用了个“进行中”指示器，逻辑很简单：任务启动计数器+1，结束时-1，只要计数器>0就显示指示器。

switchMap的特性：当源Observable发射新的内部Observable时，会停掉之前那个内部Observable的发射，转而用新的。

现在的问题是：被switchMap丢弃的旧内部Observable，不会触发计数器减1的操作this.uiService.indicator.task(-1)，导致计数器只增不减，指示器关不掉。

我试过在switchMap后面加tap操作符，但switchMap一收到新搜索词就中断了旧Observable的传递，tap根本触发不了。

相关代码 #

组件初始化逻辑 #

init(): void {
    this.searchResults$: Observable<SearchResult[]> = this.searchTerms.pipe(
        // 按键后等700ms再处理搜索词
        debounceTime(700),
        // 跳过和上一次一样的搜索词
        distinctUntilChanged(),
        // 搜索词变了就切到新的搜索请求，丢掉之前还在跑的请求
        switchMap((searchTerm, i) => this.apiService.searchDB(searchTerm, i))
    );

    this.searchSubscription = this.searchResults$.subscribe({
        next: (response) => {
            console.log('received', response)
            this.uiService.indicator.task(-1);
        }
    });
}

触发搜索的方法 #

triggerSearch(term: string): void {
    this.searchTerms.next(term);
}

API服务的搜索方法 #

apiService.searchDB(searchTerm, i): Observable<SearchResult[]> {
    this.uiService.indicator.task(+1);
    ...
}

核心要解决的就是：旧的Observable被switchMap取消订阅时，也要触发计数器减1。这里有两种靠谱的实现方式：

方式一：给内部Observable加finalize操作符 #

finalize的特性是不管Observable是正常完成、报错还是被取消订阅，都会执行回调，刚好能覆盖switchMap丢旧请求的场景。直接修改API服务的searchDB方法：

apiService.searchDB(searchTerm, i): Observable<SearchResult[]> {
    this.uiService.indicator.task(+1);
    // 假设这里原本是http请求，给它加个finalize
    return this.http.get<SearchResult[]>(/* 你的请求地址 */).pipe(
        finalize(() => {
            this.uiService.indicator.task(-1);
        })
    );
}

同时记得把订阅回调里的减1代码删掉，不然会重复减：

this.searchSubscription = this.searchResults$.subscribe({
    next: (response) => {
        console.log('received', response);
        // 删掉这行：this.uiService.indicator.task(-1);
    }
});

方式二：在switchMap内部处理取消逻辑 #

如果不想改API服务的代码，也可以在switchMap里给内部Observable套一层finalize：

switchMap((searchTerm, i) => 
    this.apiService.searchDB(searchTerm, i).pipe(
        finalize(() => {
            this.uiService.indicator.task(-1);
        })
    )
)

注意这种方式要确保searchDB里的+1和这里的-1一一对应，别搞混计数。

为啥之前的tap没用？ #

tap只有在Observable正常发值、完成或者报错的时候才会触发，但switchMap是直接取消旧Observable的订阅，不会让它走到完成/报错的流程，所以tap的回调根本跑不起来。而finalize不管Observable是啥结局，都会执行，完美适配这个场景。

内容的提问来源于stack exchange，提问作者Wilhelm Mauch