RXJS RetryWhen：重试重连时如何向下游通知连接断开状态

我明白你的问题了——retryWhen确实会拦截上游的错误并触发重试，所以这些错误根本流不到下游的catchError里，这就是为什么你的状态通知永远不会触发断开信号。想要同时维持全局重连逻辑，又能让所有订阅者知晓连接状态变化，我们可以用状态Subject + 共享连接流的方式来实现。

问题根源 #

你的retryWhen操作符会捕获WebSocket连接抛出的错误，然后直接触发重试流程，整个错误不会向下传递到connectionStatus里的catchError，所以下游永远收不到false的状态信号。

解决方案 #

我们可以单独维护一个连接状态的Subject，在重连逻辑中更新状态，同时让WebSocket数据流和状态流共享同一个连接上下文，确保所有订阅者拿到的是一致的状态和数据。

完整代码示例

import { Injectable } from '@angular/core';
import { Observable, BehaviorSubject, EMPTY, tap, retryWhen, delay, share } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class WebSocketService {
  // 用BehaviorSubject保存连接状态，初始值为false（未连接）
  private readonly statusSubject$ = new BehaviorSubject<boolean>(false);
  // 共享的WebSocket连接流，确保所有订阅者复用同一个连接
  private readonly connection$: Observable<any>;

  constructor() {
    // 初始化连接流，包含重连逻辑
    this.connection$ = this.websocket().pipe(
      // 当连接成功/收到消息时，更新状态为已连接
      tap(() => {
        if (!this.statusSubject$.value) {
          this.statusSubject$.next(true);
        }
      }),
      // 处理重连逻辑：断开时更新状态，延迟2秒重试
      retryWhen(errors => 
        errors.pipe(
          tap(() => this.statusSubject$.next(false)),
          delay(2000)
        )
      ),
      // 共享流，避免重复创建WebSocket连接
      share()
    );

    // 可选：提前订阅启动重连逻辑（如果需要应用启动就尝试连接）
    this.connection$.subscribe({
      error: () => this.statusSubject$.next(false)
    });
  }

  // 对外暴露WebSocket数据流，供业务逻辑订阅消息
  connect(): Observable<any> {
    return this.connection$;
  }

  // 对外暴露连接状态流，供UI等订阅状态变化
  connectionStatus(): Observable<boolean> {
    return this.statusSubject$.asObservable();
  }

  // 你的原始WebSocket连接方法（断开时抛出错误）
  private websocket(): Observable<any> {
    // 这里替换成你实际的WebSocket创建逻辑，比如使用webSocket()操作符
    // 示例：return webSocket('ws://your-url');
    return EMPTY; // 仅作占位
  }
}

工作原理 #

1. 状态管理：BehaviorSubject会持续保存当前连接状态，新订阅者会立即拿到最新状态（初始为false）。
2. 重连时的状态更新：在retryWhen的错误处理流程中，我们先通过tap更新状态为false，再延迟重试。
3. 连接恢复的状态更新：当WebSocket连接成功并开始推送数据时，tap会检查当前状态，如果是false就更新为true，确保状态只在连接恢复时触发一次更新。
4. 共享连接：share()操作符让所有订阅connect()的地方复用同一个WebSocket连接，避免重复创建连接导致的资源浪费。

使用示例 #

// 订阅连接状态
this.webSocketService.connectionStatus().subscribe(status => {
  console.log('连接状态:', status ? '已连接' : '断开/重连中');
});

// 订阅WebSocket消息
this.webSocketService.connect().subscribe(message => {
  console.log('收到消息:', message);
});

注意事项 #

• 记得在组件销毁时取消对connectionStatus()的订阅（或使用Angular的async管道），避免内存泄漏。
• 如果需要限制重连次数，可以在retryWhen中添加take(n)操作符，比如errors.pipe(tap(...), delay(2000), take(5))，超过次数后会触发最终的error回调，状态会更新为false。

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