RxJS zip结合filter操作符无法自动取消订阅问题求助

方案1：使用takeUntil管理订阅（推荐） #

这是Angular组件中管理RxJS订阅的标准做法，既能确保符合条件时自动完成订阅，又能在组件销毁时强制取消订阅，避免内存泄漏。

1. 定义销毁信号Subject：

private destroy$ = new Subject<void>();

2. 修改订阅逻辑，添加销毁信号控制：

constructor(
  private apollo: Apollo,
  private sessionService: SessionService,
  private userService: UserService,
  private notificationService: NotificationService,
) {
  zip(this.userService.user$, this.sessionService.session$)
    .pipe(
      tap((data) => console.log(data)),
      filter(([user, session]) => !!user && !!session),
      take(1),
      takeUntil(this.destroy$) // 绑定组件销毁信号
    )
    .subscribe({
      next: ([user, session]) => {
        this.sessionService.connectUserToSession(
          user!,
          session!,
          UserType.USER
        );
      },
      error: (err) => console.log(err),
      complete: () => console.log('complete')
    });
}

3. 在组件销毁时触发信号：

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

这种方式下，只要组件销毁，不管take(1)是否触发，订阅都会被取消；当符合filter条件的值出现时，take(1)会自动完成订阅，无需手动干预。

方案2：用first替代filter + take(1) #

first操作符支持直接传入判断条件，功能等同于filter(predicate).take(1)，写法更简洁：

constructor(
  // ...注入服务
) {
  zip(this.userService.user$, this.sessionService.session$)
    .pipe(
      tap((data) => console.log(data)),
      first(([user, session]) => !!user && !!session), // 直接用first带条件
      takeUntil(this.destroy$) // 仍需结合销毁信号处理组件销毁
    )
    .subscribe({
      next: ([user, session]) => {
        this.sessionService.connectUserToSession(
          user!,
          session!,
          UserType.USER
        );
      },
      error: (err) => console.log(err),
      complete: () => console.log('complete')
    });
}

注意：如果user$或session$是会主动完成的Observable，且完成前始终没有符合条件的值，first会抛出EmptyError，可添加catchError处理该情况：

first(([user, session]) => !!user && !!session),
catchError(err => {
  if (err instanceof EmptyError) {
    // 处理始终无符合条件值的场景，比如直接返回空Observable
    return EMPTY;
  }
  throw err;
})

方案3：主动完成源Observable（可选） #

如果user$和session$是BehaviorSubject，且在组件生命周期内数据确定不会再更新，可以主动调用complete()，这样zip会在所有源完成后自动结束订阅。但这种方式仅适用于明确知道源会终止的场景，通用性不如前两种方案。

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