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之后）块永远不会被调用。 日志：

• 配置过滤器假之前
• 在与LatestFrom 之前
• 在配置过滤器为真之前
• 配置过滤器为真后

感谢您的建议。

Answer 1

我想你想要的是.combineLatest。

withLatestFrom 仅将其前面的内容视为触发器。 combineLatest 将源和提供给它的 observables 都视为触发器。

所以您的问题是源（路由更改）触发并且传递给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>

Answer 2

withLatestFrom 仅在其源 Observable 发出时才会发出。发生的情况是this.store.select 在源之前发出withLatestFrom。从排放的顺序可以看出：

• 配置过滤器假之前

• 在与LatestFrom 之前

• ...

现在这取决于您想要实现的目标。只能在源发出时使用concatMap 触发this.store.select(selectConfigInitialized)：

navigationToActiveRedemptionOffers = this.actions$.ofType(ROUTER_NAVIGATION)
  .filter(...)
  .concatMap(() => this.store.select(selectConfigInitialized)...)
  .switchMap(data => { ... })
  ...

或者你可以传递两个 Observable 的结果：

navigationToActiveRedemptionOffers = this.actions$.ofType(ROUTER_NAVIGATION)
  .filter(...)
  .concatMap(nav => this.store.select(selectConfigInitialized)
    .map(result => [nav, result])
  )
  .switchMap(data => { ... })
  ...

Answer 3

你可以通过反转 observables 来简化它。

public navigationToActiveRedemptionOffers = 

  return this.store.select(selectConfigInitialized)
    .filter(initialized => initialized)      
    .flatMap(x => 
      return this.actions$.ofType(ROUTER_NAVIGATION)
        .filter((action: RouterNavigationAction<RouterStateProjection>) => {
          return action.payload.event.url.includes('/redemption-offers/list/active');
        })
    .map(action =>
      return [
        new MapListingOptionsAction(action.payload.routerState.queryParams),
        new LoadActiveOffersAction(),
      ];
    )

由于这是一种常见的模式，我使用通用的waitFor 方法。

export const waitFor$ = function() {
  return this.filter(data => !!data ).take(1);
};

用法，

return this.config$
  .waitFor$()
  .flatMap(config => {
    return observableWithRequiredValue()
  }

提供的流利语法

Observable.prototype.waitFor$ = waitFor$;

declare module 'rxjs/Observable' {
  interface Observable<T> {
    waitFor$: typeof waitFor$;
  }
}

注意，这种创建运算符的方法在 rxjs v5.5 中已被Lettable Operators 取代（但上述方法仍然有效）。