RxJs withLatestFrom在Observable延迟发射时不触发问题排查

我在NgRx Effect中需要从Store获取若干值用于后续服务调用，这些值可能延迟到达，所以用filter操作符等待它们变为可用状态。我写了如下代码：

public loadData$ = createEffect(() =>
 this.actions$
  .pipe(
    ofType(appActions.viewOpened),
    tap(() => this.logger.info('loadData$')),
    
    withLatestFrom(
      combineLatest([
        this.store$.select(fromApp.getToken).pipe(filter(x => !!x)),
        this.store$.select(fromApp.getAuthorizationInfo).pipe(filter(x => !!x))
      ])),
        
    switchMap(([_, [token, authInfo]]) => {
      tap(() => this.logger.info(`loadData$ Got  ${token}`));
      const obs = from(this.getData(token, authInfo.userName));
      return obs;
    })
  ), { dispatch: false });

  
  private async getData(): Promise<void> {
     ....
     // 分发 action(s)
  }

但发现当Effect首次触发时，如果其中一个store$.select没有值，就算后续两个Store值都变为可用，switchMap也永远不会执行。只有当两个值都存在，或者移除filter时，switchMap才会正常运行。

我的问题是：为什么会出现这种情况？是不是withLatestFrom导致switchMap无法被调用？另外，如果上述用法不正确，应该如何正确获取这些Store值？

更新1 #

我发现即使是只获取单个Store值的简单场景也有这个问题，比如以下代码也永远无法进入switchMap：

public loadData$ = createEffect(() =>
 this.actions$
  .pipe(
    ofType(appActions.viewOpened),
    tap(() => this.logger.info('loadData$')),
    
    withLatestFrom(this.store$.select(fromApp.getToken).pipe(filter(x => !!x))),
        
    switchMap(([_, token]) => {
      tap(() => this.logger.info(`loadData$ Got  ${token}`));
      const obs = from(this.getData(token));
      return obs;
    })
  ), { dispatch: false });

核心问题出在withLatestFrom的工作机制上：

• withLatestFrom会在源Observable（此处为actions$.pipe(ofType(appActions.viewOpened))）发出值的瞬间，去获取传入Observable的最新值。
• 如果你给withLatestFrom传入的Observable（比如store$.select(...).pipe(filter(x => !!x))）此时还没有发出过任何值（因为filter把初始空值过滤掉了），那么withLatestFrom就不会产生任何输出，整个流会卡在这一步，switchMap自然不会执行。
• 后续Store值更新时，源Observable（action流）已经发出过viewOpened动作，不会再触发withLatestFrom重新取值，所以永远不会进入switchMap。

方案1：用concatMap + combineLatest等待值就绪 #

把withLatestFrom换成concatMap，在concatMap内部等待所需Store值都可用后再执行服务调用：

public loadData$ = createEffect(() =>
 this.actions$
  .pipe(
    ofType(appActions.viewOpened),
    tap(() => this.logger.info('loadData$')),
    concatMap(() => 
      combineLatest([
        this.store$.select(fromApp.getToken).pipe(filter(x => !!x)),
        this.store$.select(fromApp.getAuthorizationInfo).pipe(filter(x => !!x))
      ]).pipe(
        take(1), // 只取一次就绪后的值，避免重复触发
        switchMap(([token, authInfo]) => {
          this.logger.info(`loadData$ Got  ${token}`);
          return from(this.getData(token, authInfo.userName));
        })
      )
    )
  ), { dispatch: false });

• concatMap会在action触发后，订阅内部的combineLatest流，直到两个Store值都通过filter校验（即就绪），才会继续执行后续逻辑。
• take(1)确保只获取一次就绪的值，防止Store后续更新再次触发服务调用。

方案2：用withLatestFrom结合startWith和统一过滤（适合允许初始默认值的场景） #

如果你的Store值可以设置合理默认值，也可以给每个store$.select加上startWith，再在外层统一过滤：

public loadData$ = createEffect(() =>
 this.actions$
  .pipe(
    ofType(appActions.viewOpened),
    tap(() => this.logger.info('loadData$')),
    withLatestFrom(
      combineLatest([
        this.store$.select(fromApp.getToken).pipe(startWith(null)),
        this.store$.select(fromApp.getAuthorizationInfo).pipe(startWith(null))
      ])
    ),
    filter(([_, [token, authInfo]]) => !!token && !!authInfo), // 统一过滤确保值就绪
    switchMap(([_, [token, authInfo]]) => {
      this.logger.info(`loadData$ Got  ${token}`);
      return from(this.getData(token, authInfo.userName));
    })
  ), { dispatch: false });

• startWith(null)确保withLatestFrom在action触发时能拿到值（哪怕是null），然后通过外层的filter判断两个值是否都就绪。
• 后续Store值更新时，只要action已经触发过，withLatestFrom会在Store更新时重新检查值是否就绪，一旦满足条件就会进入switchMap。

注意：你之前代码里的tap在switchMap内部没有被订阅，所以不会执行，要直接调用this.logger.info或者把tap放到pipe链中。

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