RxJS：takeUntil 具有多个操作和不同的过滤器？答案

【问题标题】：RxJS: takeUntil with multiple actions and different filters?RxJS：takeUntil 具有多个操作和不同的过滤器？
【发布时间】：2018-02-09 15:54:58
【问题描述】：

我有一个 Observable 我想继续执行直到：

1) uploadActions.MARK_UPLOAD_AS_COMPLETE 动作是用特定的有效载荷调用的

或

2) 使用任何有效负载调用 uploadActions.UPLOAD_FAILURE 操作

这是我所能得到的（但不起作用）：

return Observable.interval(5000)
  .takeUntil(
    action$
      .ofType(
        uploadActions.UPLOAD_FAILURE,
        uploadActions.MARK_UPLOAD_AS_COMPLETE
      )
      .filter(a => { // <---- this filter only applies to uploadActions.MARK_UPLOAD_AS_COMPLETE
        const completedFileHandle = a.payload;
        return handle === completedFileHandle;
      })
  )
  .mergeMap(action =>
    ...
  );

有没有一种干净的方法可以实现这一点？

【问题讨论】：

什么具体不起作用？我的直觉反应是在两个单独的action$.ofType(...) 上做一个combineLatest，这样你就可以过滤那个了。
@bygrace 因为它将filter 应用到uploadActions.UPLOAD_FAILURE，它没有有效负载，因此返回false

标签： rxjs redux-observable

【解决方案1】：

我会将这两个条件分成不同的流，然后像这样合并它们：

const action$ = new Rx.Subject();
const uploadActions = {
  UPLOAD_FAILURE: "UPLOAD_FAILURE",
  MARK_UPLOAD_AS_COMPLETE: "MARK_UPLOAD_AS_COMPLETE"
};
const handle = 42;

window.setTimeout(() => action$.next({
  type: uploadActions.MARK_UPLOAD_AS_COMPLETE,
  payload: handle
}), 1200);

Rx.Observable.interval(500)
  .takeUntil(
    Rx.Observable.merge(
      action$.filter(x => x.type === uploadActions.UPLOAD_FAILURE),
      action$.filter(x => x.type === uploadActions.MARK_UPLOAD_AS_COMPLETE)
      	.filter(x => x.payload === handle)      
    )
  ).subscribe(
    x => { console.log('Next: ', x); },
    e => { console.log('Error: ', e); },
    () => { console.log('Completed'); }
  );

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.min.js"&gt;&lt;/script&gt;

例如，我必须使用 filter 运算符而不是 ofType，因为 ofType 是一个 redux 的东西。

【讨论】：

完美！我使用了 Observable.merge 并最终得到：Observable.merge(action$.ofType(...).filter(...), action$.ofType(...)) 谢谢！