redux - 如何在 redux-observable 中重新初始化 Action ？

Question

例如，这段代码this jsbin example :

const pingEpic = action$ =>
  action$.ofType(PING)
    .delay(1000) // Asynchronously wait 1000ms then continue
    .mapTo({ type: PONG })
    .takeUntil(action$.ofType(CANCEL));

当我如上所述使用 takeUntil 时，在分派(dispatch) CANCEL 操作并延迟 1 秒后，该操作再也不会触发。为什么？

Answer 1

问题是对 RxJS 工作原理的微妙但严重的误解——但不要害怕，这是很常见的。

举个例子:

const pingEpic = action$ =>
  action$.ofType(PING)
    .delay(1000)
    .mapTo({ type: PONG })
    .takeUntil(action$.ofType(CANCEL));

这个史诗的行为可以被描述为过滤掉所有不匹配类型 PING 的 Action 。当一个 Action 匹配时，等待 1000 毫秒，然后将该 Action 映射到另一个 Action { type: PONG } ，它将被发出，然后由 redux-observable 调度。如果在应用程序运行时的任何时间有人发出CANCEL类型的操作，然后从源取消订阅，这意味着整个链将取消订阅，终止史诗。

如果您命令式地执行它，看看它看起来如何可能会有所帮助:

const pingEpic = action$ => {
  return new Rx.Observable(observer => {
    console.log('[pingEpic] subscribe');
    let timer;

    const subscription = action$.subscribe(action => {
      console.log('[pingEpic] received action: ' + action.type);

      // When anyone dispatches CANCEL, we stop listening entirely!
      if (action.type === CANCEL) {
        observer.complete();
        return;
      }

      if (action.type === PING) {
        timer = setTimeout(() => {
          const output = { type: PONG };
          observer.next(output);
        }, 1000);
      }
    });

    return {
      unsubscribe() {
        console.log('[pingEpic] unsubscribe');
        clearTimeout(timer);
        subscription.unsubscribe();
      }
    };
  });
};

您可以在此处使用假商店运行此代码:http://jsbin.com/zeqasih/edit?js,console

---

相反，您通常想要做的是将您希望取消的订阅者链与假定无限期监听的顶级链隔离开来。尽管您的示例(根据文档进行了修改)是人为设计的，但让我们先运行一下。

这里我们使用 mergeMap运算符让我们采取匹配的操作并映射到另一个单独的可观察链。

演示:http://jsbin.com/nofato/edit?js,output

const pingEpic = action$ =>
  action$.ofType(PING)
    .mergeMap(() =>
      Observable.timer(1000)
        .takeUntil(action$.ofType(CANCEL))
        .mapTo({ type: PONG })
    );

我们使用 Observable.timer等待 1000 毫秒，然后将它发出的值(碰巧是数字零，但这在这里并不重要)映射到我们的 PONG 操作。我们还说我们想从计时器源“获取”，直到它正常完成或我们收到类型为 CANCEL 的操作。

这隔离了链，因为 mergeMap 将继续订阅您返回的可观察对象，直到它出错或完成。但是当这种情况发生时，它自身并不会停止订阅您应用它的来源；本例中的 action$.ofType(PING)。

一个更真实的例子在 Cancellation section 的 redux-observable 文档中。

Here we placed the .takeUntil() after inside our .mergeMap(), but after our AJAX call; this is important because we want to cancel only the AJAX request, not stop the Epic from listening for any future actions.

const fetchUserEpic = action$ =>
  action$.ofType(FETCH_USER)
    .mergeMap(action =>
      ajax.getJSON(`/api/users/${action.payload}`)
        .map(fetchUserFulfilled)
        .takeUntil(action$.ofType(FETCH_USER_CANCELLED))
    );

这一切听起来可能令人困惑，但就像最强大的东西一样，一旦你掌握了它，它就会变得直观。 Ben Lesh 出色地解释了 Observables 如何在 his recent talk 中工作，包括讨论运营商如何成为 Observables 链，甚至关于隔离订阅者链。尽管谈话是在 AngularConnect 上进行的，但它并不是特定于 Angular 的。

---

顺便说一句，重要的是要注意你的史诗不会吞噬或以其他方式阻止 Action 到达 reducer ，例如当您将传入操作映射到另一个不同的操作时。事实上，当您的史诗收到一个 Action 时，它已经通过了您的 reducer 。将您的史诗想象成边车进程，它会监听您的应用程序操作流，但无法阻止正常的 redux 事件发生，它只能发出新操作。