Дождитесь изменения состояния, вызывающего повторное действие.

Question

Когда мой эпос отвечает в точке 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)).

Answer 1

Во-первых, обратите внимание, что оператор concat устарел в RxJS 6 (это будет оператор, импортированный из rxjs/operators). Однако функция concat, используемая для создания наблюдаемой, является , а не устаревшей (это будет функция, импортированная из rxjs). Я бы порекомендовал использовать операторы, которые не устарели.

Во-вторых, есть несколько проблем с вашим текущим подходом.

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

Вышеуказанные действия фильтруют действия, соответствующие типу "START_CONTINUE_SESSION" , и позволяют им возвращаться в хранилище Redux . Это связано с тем, что оператор concat позволяет исходным событиям проходить и ожидает завершения предыдущей наблюдаемой до того, как начнет следующую наблюдаемую. Тем не менее, поскольку поток действий, наблюдаемый при редуксе, никогда не завершается, concat должен никогда не начинать следующие наблюдаемые! Посмотрите на следующую мраморную диаграмму из старых документов RxJS:

Как видно из диаграммы, исходные события проходят. При использовании избыточного-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 . Но обратите внимание, эта подписка никогда не заканчивается! Опять же, у вас будет бесконечный цикл действий, который будет входить и выходить обратно в магазин Redux, пока foo остается true.

---

Вместо подписки на 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 - Слушайте все действия, но выполняйте только одно за раз. Отменяет любой предыдущий рабочий процесс при получении нового действия.