RXJS 重复没有机会重复？

Question

我在我的应用程序中使用了以下史诗来处理 api 请求：

action$ => {
  return action$.ofType(actions.requestType)
    .do(() => console.log('handled epic ' + actions.requestType))
    .switchMap((action) => (
      Observable.create((obs) => {
        obs.next({ type: type, value: action.value, form: action.form });
      })
      .debounceTime(250)
      .switchMap((iea) => (
        Observable.ajax(ajaxPost(url(iea.value), body ? body(iea.value) : action.form))
          .mergeMap(payload => {
            return Observable.merge(
              Observable.of(actions.success(payload)),
              /* some other stuff */
            );
          })
          .catch(payload => {
            return [actions.failure(payload)];
          })
      ))
  ))
  .takeUntil(action$.filter((a) => (a.type === masterCancelAction))
  .repeat();
};

基本上，每当我执行一个 api 请求时，我都会发送一个请求动作。如果我快速发送另一个请求，则使用 debounceTime 忽略前一个请求。此外，可以使用 masterCancelAction 取消请求，并在取消时 repeat() 重新启动史诗。这部史诗在所有情况下都能按预期工作。

失败案例发生在用户在请求期间使用浏览器返回。在这种情况下，我将 masterCancelAction 触发到请求。但是，在作为 masterCancelAction 的结果的相同执行上下文中，另一个请求操作分派以对同一史诗执行新请求，但是没有发生 api 请求（尽管确实发生了 console.log），就好像没有重复一样()。在发生取消的其他情况下，不会从相同的执行上下文调用下一个请求并且它工作正常，所以在这种情况下，我的代码似乎没有给重复机会重新启动史诗？

我发现一个肮脏的解决方法是在取消后调度的请求上使用 setTimeout(dispatch(action), 0)。这似乎允许 repeat() 执行。我尝试将不同的调度程序传递给重复，但这似乎没有帮助。此外，将 takeUntil 和 repeat 附加到我的内部 switchMap 可以解决问题，但是其他情况下我的下一个请求未在同一个调用堆栈中执行会失败。

有没有一种方法可以在不使用 setTimeout 的情况下解决这个问题？也许这不是重复相关的问题，但似乎是这样。

使用 rxjs 5.0.3 和 redux-observable 0.14.1。

Answer 1

如果没有 jsbin 之类的东西来了解您的意思，这个问题并不是 100% 清楚的，但我确实看到了一些可能有帮助的一般性问题：

匿名 Observable 永远不会完成

在创建自定义匿名 Observable 时，如果您确实希望它完成，请务必调用 observer.complete()。在大多数情况下，不这样做会导致订阅发生内存泄漏，并且还可能出现其他奇怪的行为

Observable.create((observer) => {
  observer.next({ type: type, value: action.value, form: action.form });
  observer.complete();
})

Observable.of 是等价的：

Observable.of({ type: type, value: action.value, form: action.form })

但是，不清楚为什么要这样做，因为它发出的值是在范围内捕获的。

debounceTime 在这种情况下不会去抖动，它会延迟

由于它所应用的匿名 observable 只发出一个项目，debounceTime 将充当常规 .delay(250)。我打赌你打算改为去抖actions.requestType 动作，在这种情况下，你需要在action$.ofType(actions.requestType) 之后在switchMap 之外应用去抖。

Observable.of 接受任意数量的参数来发出

这更像是“你知道吗？”而不是问题，但我注意到您正在合并您的 of 和 /* some other actions */ 我假设其他 of observables 会合并。相反，您可以只返回一个 of 并将操作作为参数传递。

Observable.of(
  actions.success(payload),
  /* some other actions */
  actions.someOtherOne(),
  actions.etc()
);

此外，当您发现自己像这样同步发出多个操作时，请考虑您的减速器是否应该监听相同的单个操作，而不是两个或更多。有时这没有意义，因为您希望它们具有完全不相关的动作，只是要记住人们经常忘记的一点——所有减速器都接收所有动作，因此多个减速器可以从同一个动作改变它们的状态。

.takeUntil 将阻止史诗监听未来的行动

将takeUntil 放在顶级可观察链上会导致史诗停止侦听action$.ofType(actions.requestType)，这就是您在后面添加.repeat() 的原因。这在某些情况下可能有效，但效率低下并且可能导致其他难以实现的错误。 Epics 应该被认为是类似于通常随应用程序“启动”的 sidecar 进程，然后继续侦听特定操作，直到应用程序“关闭”，即用户离开应用程序。它们实际上不是过程，只是从概念上将它们视为抽象是有帮助的。

因此，每次它匹配其特定操作时，它通常会switchMap、mergeMap、concatMap 或exhaustMap 产生一些副作用，例如 ajax 调用。 inner observable chain 是你想要取消的。因此，您可以将 .takeUntil 放在链中的适当位置。

---

总结

如前所述，如果没有像 jsbin 这样更完整的示例，还不清楚您打算做什么以及问题是什么。但严格根据提供的代码，这是我的猜测：

const someRequestEpic = action$ => {
  return action$.ofType(actions.requestType)
    .debounceTime(250)
    .do(() => console.log('handled epic ' + actions.requestType))
    .switchMap((action) =>
      Observable.ajax(ajaxPost(url(action.value), body ? body(action.value) : action.form))
        .takeUntil(action$.ofType(masterCancelAction))
        .mergeMap(payload => {
          return Observable.of(
            actions.success(payload),
            /* some other actions */
            ...etc
          );
        })
        .catch(payload => Observable.of(
          actions.failure(payload)
        ))
    );
};

查看 redux-observable 文档中的 Cancellation 页面。

---

如果这有点令人困惑，我建议更深入地研究什么是 Observables 以及什么是“操作符”以及它的作用，这样它就不会让人觉得很神奇，并且应该将操作符放在哪里更有意义。

Ben 在Learning Observable by Building Observable 上的帖子是一个好的开始。