等待状态变化导致重复动作发出答案

【问题标题】：Wait for change in state causing repeated action emit等待状态变化导致重复动作发出
【发布时间】：2019-04-21 08:57:17
【问题描述】：

当我的史诗在ofType 点响应时，我需要等到state$.value.foo 变为true。一旦它是真的，那么我希望它到达from，它会做一些重要的事情。我是这样做的：

action$.pipe(
    ofType(START_CONTINUE_SESSION),
    concat(
        iif(
            () => state$.value.foo === true,
            EMPTY,
            action$.pipe(
                filter(() => state$.value.foo === true),
            )
        ),
        from(fetch(...)).pipe(
            // ... do important stuff here BUT only after state$.value.foo has become true
        )
    )
)

发生的情况是我发出了超大量的操作，但它永远不会到达from(fetch))。

【问题讨论】：

标签： rxjs redux-observable

【解决方案1】：

首先，请注意 concat 运算符在 RxJS 6 中已弃用（这将是从 rxjs/operators 导入的运算符）。但是，用于创建 observable 的函数 concat 未被弃用（这将是从 rxjs 导入的函数）。我建议您使用未弃用的运算符。

其次，您当前的方法存在几个问题。

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

上述过滤匹配类型“START_CONTINUE_SESSION”的操作并允许它们传回 Redux 存储。这是因为concat 运算符让源事件通过并等待前一个可观察对象完成在开始下一个可观察对象。然而，因为 redux-observable 动作流永远不会完成，concat 应该永远不会开始下一个 observables！查看旧 RxJS 文档中的以下大理石图：

如图所示，源事件通过。使用 redux-oberable，这意味着您的“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 存储区。

您应该订阅state$，而不是订阅action$ 并检查状态。下面的例子有点不同，但我认为它展示了一种实现目标的方法。这将等待初始操作 (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 - 监听所有动作，但一次只运行一个。收到新操作时取消任何以前的工作流程。

【讨论】：

非常感谢您，在这里可以学到很多东西。但问题是在获得START_CONTINUE_SESSION 之后，它可能已经拥有foo = true 的状态。所以它可能不必等待。此代码检查下一个刻度是否正确？（我只是想学习，我为什么要做这个项目，学习 redux-observables 和 rxjs :)）
来自rxjs 的import { concat } 是否已弃用或来自rxjs/operators 的那个已弃用？
@Noitidart 我刚刚做了一个（大）编辑，试图解决你所有的 cmets/问题。你做了一个很好的说明——我提供了一个替代解决方案，但没有描述为什么你的代码没有按照你的预期工作。我希望这个编辑有助于澄清！
@Noitidart 不，我的意思是严格的可观察定义意义上的。完成一个流是一个显式的流通知（参见subscribe 方法的第三个回调）。例如，of(1) 是一个 observable，它将“下一个”值 1，然后“完成”。 action$ 流 never 完成（这就是 redux-observable 的工作方式）。因为它从不发送“完成”通知，所以等待“完成”通知的东西，比如过时的concat 运算符，将无限期地等待。
@Noitidart 鉴于评论格式有限，我很难理解这个问题。也许将其发布到一个新问题中？ 我认为这仍然行不通，因为它没有结束外部，只是内部。我想你会想看看链接first() 和ignoreElements() 的hack。同样，一个新问题可能是回答这个问题的更好地方。