RxJS Observable触发传感器首次发射后停止响应问题排查

核心问题分析 #

当前代码的核心问题在于嵌套switchMap的逻辑设计，以及对Observable特性的利用不当，导致传感器后续状态变化无法触发运输任务的启停逻辑。以下是具体排查点和修复方案：

1. 嵌套订阅导致的状态同步问题 #

你在sensorStatus$的switchMap内部嵌套订阅currentTransport$并使用take(1)，如果currentTransport$是冷Observable，或者仅在初始化时发射一次值，后续传感器状态变化时，currentTransport$无法提供最新的运输状态，直接导致后续逻辑不执行。

修复方案：
改用withLatestFrom组合传感器状态和当前运输状态，确保每次传感器变化都能获取到最新的运输状态：

this.sensor.sensorStatus$.pipe(
  tap(() => console.log('BLE Sensor Emission')),
  skip(1),
  // 实时获取当前运输状态的最新值
  withLatestFrom(this.transportService.currentTransport$),
  switchMap(([sensor, currentTransport]) => {
    if (currentTransport && !sensor.isLoaded) {
      return this.transportService.completeTransport$();
    }
    if (sensor.isLoaded && !currentTransport) {
      return this.transportService.startTransport$(this.transportForm.value);
    }
    return of(null);
  }),
  // 过滤无意义的空值发射
  filter(transport => transport !== null)
).subscribe({
  next: (transport) => {
    this.startedTransport.emit('started');
  },
  error: err => {
    this.isTransportAlertOpen = true;
    this.errorMessage = 'Verbindung mit Zigpos überprüfen';
  },
  complete: () => console.warn('Completed')
});

2. switchMap的取消特性干扰后续逻辑 #

外层switchMap会在sensorStatus$发射新值时，直接取消上一次未完成的内部Observable订阅。如果completeTransport$或startTransport$的后端调用耗时较长，新的传感器状态会直接中断未完成的请求，导致后续逻辑无响应。

调整方案：
根据业务需求替换外层switchMap：

• 用concatMap：按顺序处理每个传感器状态变化，前一个后端请求完成后再处理下一个
• 用exhaustMap：忽略传感器的频繁触发，直到当前后端请求完成再响应新的状态

3. 确认currentTransport$的热Observable实现 #

检查transportService中currentTransport$的实现，必须用BehaviorSubject或ReplaySubject确保它是热Observable，这样每次订阅都能获取到最新的运输状态：

// transportService.ts 示例实现
private currentTransportSubject = new BehaviorSubject<boolean>(false);
public currentTransport$ = this.currentTransportSubject.asObservable();

// 运输状态变化时更新Subject
updateTransportStatus(status: boolean) {
  this.currentTransportSubject.next(status);
}

4. 验证后端请求的完成状态 #

在completeTransport$和startTransport$的调用处添加日志，确认请求是否正常完成或报错：

return this.transportService.completeTransport$().pipe(
  tap(() => console.log('completeTransport$ 执行完成')),
  catchError(err => {
    console.error('completeTransport$ 错误:', err);
    throw err;
  })
);

如果请求报错未被正确捕获，可能导致整个Observable流被终止，后续传感器状态变化无法触发逻辑。

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