RxJS永不终止流每次请求成功/报错时执行回调的正确实现方式

问题背景 #

需要在RxJS长生命周期流的每一次请求成功、错误回调节点执行指定逻辑，找到this.showLoader.next(false)的正确调用位置，初始实现代码如下：

this.uiDataObservable.pipe(
    tap(() => this.showLoader.next(true)),
    map((uiData) => this.createHttpRequest(uiData)),
    switchMap((httpRequest) => this.apiService.get(httpRequest)),
    map((response) => this.createUISucessData(response))
).subscribe(
    (success) => /* show data on UI */, 
    (error) => /* show error on UI */
);

实现存在两个硬性限制：

• 无法在外层流使用finalize操作符，因为外层数据流为长生命周期流，永远不会终止
• 无法依赖subscribe方法的complete回调，因为请求发生错误时该回调不会被触发

实现方案 #

核心逻辑：将loading关闭的逻辑绑定到switchMap内部的单次HTTP请求短生命周期流上，而非外层长生命周期流。单次HTTP请求必然会在成功/失败后进入终态，在这里执行收尾逻辑完全不受外层流永不终止的影响，同时需要避免内部请求错误终止外层长流。

具体实现要点 #

• 外层流仅保留「发起新请求时打开loading」的逻辑，不处理任何收尾操作
• 在switchMap返回的HTTP请求流管道中处理单请求的结果转换、收尾逻辑：
  • 可以使用内部流的finalize操作符执行loading关闭，该操作符会在单请求流进入终态（成功/报错/被新请求取消）时触发，和外层长流无关
  • 必须在内部流中捕获HTTP错误，处理完成后重新抛出，避免错误冒泡到外层流导致长流终止，后续UI变更无法触发新请求

完整实现代码 #

import { catchError, finalize, map, switchMap, tap, throwError } from 'rxjs';

this.uiDataObservable.pipe(
  // 外层流：新请求触发时打开loading
  tap(() => this.showLoader.next(true)),
  map((uiData) => this.createHttpRequest(uiData)),
  switchMap((httpRequest) => 
    // 内部单次HTTP请求流，短生命周期，请求结束即终止
    this.apiService.get(httpRequest).pipe(
      map((response) => this.createUISucessData(response)),
      // 单请求结束（成功/失败/被新请求取消）时统一关闭loading
      finalize(() => this.showLoader.next(false)),
      // 捕获内部请求错误，重抛给外层订阅处理，同时保证外层长流不被终止
      catchError((err) => throwError(() => err))
    )
  )
).subscribe(
  (success) => /* 渲染UI成功数据 */,
  (error) => /* 渲染UI错误提示 */
);

可选替代写法 #

如果不想在内部流使用finalize，也可以通过tap的观察者对象分别在成功、错误节点触发关闭逻辑，效果完全一致：

this.apiService.get(httpRequest).pipe(
  map((response) => this.createUISucessData(response)),
  tap({
    next: () => this.showLoader.next(false),
    error: () => this.showLoader.next(false)
  }),
  catchError((err) => throwError(() => err))
)

方案合规性说明 #

• 没有在外层长生命周期流使用finalize，符合限制要求
• 没有依赖subscribe的complete回调，请求成功、错误场景下都会触发loading关闭
• 额外修复了原代码的隐藏问题：原代码中HTTP请求报错会直接终止外层长流，后续UI参数变更再也无法触发新请求，内部catchError重抛的逻辑可以保证外层流持续存活。

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