问等待状态发生变化，导致重复操作发出

EN

首先，请注意，concat运算符在RxJS 6中不推荐使用(这将是从rxjs/operators导入的操作符)。但是，用于创建可观察的函数concat不受欢迎(这将是从rxjs导入的函数)。我建议使用不建议使用的运算符。

第二，目前的方法有几个问题。

action$.pipe(
  ofType(START_CONTINUE_SESSION),
  concat(
    ...

上述筛选操作匹配类型"START_CONTINUE_SESSION“，并允许它们返回到Redux存储。这是因为concat操作符允许源事件通过，并在开始下一个可观测值之前等待可观察到的前一个事件完成。但是，因为可还原的可观察的动作流永远不会完成，所以concat永远不应该开始下一个可观察的！查看旧RxJS文档中的以下大理石图：

​

​

如图所示，源事件通过。使用redux-obserable，这意味着您的"START_CONTINUE_SESSION“操作将被困在一个永无止境的重复循环中。

即使动作流结束并且concat开始下一个可观测值，还有其他问题：

  ...
  iif(
    () => state$.value.foo === true,
    EMPTY, // I've assumed that this is equivalent to `empty()`
    action$.pipe(
      filter(() => state$.value.foo === true),
    ),
  ),
  ...

您首先要检查商店中foo的当前值。如果其值为true，则不会发出更多的(仅在此特定步骤中)，from将在下一步开始。如果其值为false，则将创建一个新的动作流订阅。对于以后发出的每一个操作，这都会检查存储中foo的当前值。当它的值最终变成true时，流入的动作(可以是任何动作！)被允许传回给Redux商店。但是请注意，这个订阅永远不会结束！同样，只要foo仍然是true，就会有一个无休止的动作循环进入并返回到Redux商店。

与其订阅action$和检查状态，不如订阅state$。下面的例子有点不同，但我认为它展示了一种实现目标的方法。这将等待初始操作(START_CONTINUE_SESSION)，然后等待状态具有foo === true，然后将操作和状态发送到mergeMap，在那里您可以正常地处理它(获取、分派其他操作等)。如果你不需要状态的副本，那就可以不去理会了。

export const epic = (action$, state$) =>
  action$.pipe(
    ofType(START_CONTINUE_SESSION),
    withLatestFrom(state$),
    exhaustMap(([action, state]) =>
      state.foo === true
        ? of([action, state])
        : state$.pipe(
            mergeMap(state =>
              state.foo === true
                ? of([action, state])
                : empty()
            ),
            first(),
          )
    ),
    mergeMap(([action, state]) =>
      // here we have triggering action and state with foo === true
      from(fetch(...)).pipe(
        // ... do important stuff here
      )
    ),
  )

至于对初始操作(START_CONTINUE_SESSION)的响应，我在上面的示例中选择了exhaustMap。可能的替代方案包括concatMap、mergeMap和switchMap。您应该选择最适合您的用例的操作符：

• concatMap -侦听所有操作，然后依次运行多个工作流。
• exhaustMap -侦听第一个操作，直到您完成工作流程后才接受另一个操作。
• mergeMap -侦听所有操作并并行运行多个工作流。
• switchMap -监听所有操作，但一次只运行一次。在接收新操作时取消任何以前的工作流。