rxjs - withLatestFrom 不发出值

Question

我试图阻止主流直到使用 withLatestFrom 初始化配置运算符和辅助流(包含有关配置状态的信息)。

这是我的代码:

@Effect()
public navigationToActiveRedemptionOffers = this.actions$.ofType(ROUTER_NAVIGATION)
  .filter((action: RouterNavigationAction<RouterStateProjection>) => {
    return action.payload.event.url.includes('/redemption-offers/list/active');
  })
  .do(() => console.log('before withLatestFrom'))
  .withLatestFrom(
    this.store.select(selectConfigInitialized)
      .do(initialized => console.log('before config filter', initialized))
      .filter(initialized => initialized)
      .do(initialized => console.log('after config filter', initialized)),
  )
  .do(() => console.log('after withLatestFrom'))
  .switchMap(data => {
    const action = data[0] as RouterNavigationAction<RouterStateProjection>;

    return [
      new MapListingOptionsAction(action.payload.routerState.queryParams),
      new LoadActiveOffersAction(),
    ];
  });

问题是第二个 do (在 withLatestFrom 之后)块永远不会被调用。
日志:

配置过滤器之前假

之前与最新来自

在配置过滤器为真之前

配置过滤器为真后

谢谢你的建议。

Answer 1

我想你想要的是.combineLatest .
withLatestFrom只将它之前的内容视为触发器。combineLatest将源和提供给它的 observables 都视为触发器。

所以你的问题是源(路由更改)触发并且 observable 传递给 withLatest (状态已初始化)由于其过滤器尚未发出。它只在源触发后发出一个值。所以它被忽略了。

这是您正在执行的操作的运行示例:

const first = Rx.Observable.create((o) => {
  setTimeout(() => {
    o.next();
  }, 2000);
});
const second = Rx.Observable.create((o) => {
  setTimeout(() => {
    o.next(false);
  }, 1000);
  setTimeout(() => {
    o.next(true);
  }, 3000);
});

first
  .do(() => console.log('before withLatestFrom'))
  .withLatestFrom(
    second
      .do(initialized => console.log('before config filter', initialized))
      .filter(initialized => initialized)
      .do(initialized => console.log('after config filter', initialized)),
  )
  .subscribe(() => console.log('after withLatestFrom'));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.2/Rx.js"></script>

这是与 combineLatest :

const first = Rx.Observable.create((o) => {
  setTimeout(() => {
    o.next();
  }, 2000);
});
const second = Rx.Observable.create((o) => {
  setTimeout(() => {
    o.next(false);
  }, 1000);
  setTimeout(() => {
    o.next(true);
  }, 3000);
});

first
  .do(() => console.log('before withLatestFrom'))
  .combineLatest(
    second
      .do(initialized => console.log('before config filter', initialized))
      .filter(initialized => initialized)
      .do(initialized => console.log('after config filter', initialized)),
  )
  .subscribe(() => console.log('after withLatestFrom'));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.2/Rx.js"></script>