Диспетчерское действие в среднем течении

Question

Впервые на RxJS. И наблюдаемый редукс.

У меня возникли проблемы. Я пытаюсь отправить действие в середине потока of(takeAction()) в приведенном ниже коде, а затем прослушать ответные действия этого типа RESOLVE_TAKE или REJECT_TAKE. Однако мой of(takeAction) не запускается, кто-нибудь знает, как исправить этот код ниже?

В дополнение к этому вопросу, бонус: Любой совет по стилю о том, как реструктурировать мой код, я не уверен, что это самый чистый и читаемый способ. Я делаю выборку, затем switch для различных кодов состояния, затем получаю res (ответ выборки) и reply, и, если возможно, включаю ответ в json, а затем передаю res и reply оба аргумента takeAction. Затем дождитесь, пока конвейер takeAction отправит RESOLVE_TAKE или REJECT_TAKE.

action$.pipe(
    ofType(START_FOO),
    switchMap({ url } =>
        from(fetch(url)).pipe(
            mergeMap(res => from(res.text()).pipe(
                mergeMap(reply => {
                    try { reply = JSON.parse(reply) } catch(ignore) {}
                    switch (res.status) {
                        case 200: {
                            return of(takeAction(res, reply)).pipe( // not dispatching
                                action$.pipe(
                                    ofType(RESOLVE_TAKE, REJECT_TAKE),
                                    mergeMap(({ type }) => {
                                        if (type === RESOLVE_TAKE) {
                                            return of(resolveFooAction())
                                        } else {
                                            return of(rejectFooAction())
                                        }
                                    })
                                )
                            )
                        }
                        // other res.status cases go here
                    }
                })
            )
        )
    )
)

Answer 1

Похоже, проблема связана с вашим of(takeAction(res, reply)). В частности, это .pipe(...). Это отправляет вашу команду «takeAction» на материал внутри .pipe, а не позволяет ей «перетекать» в ваш магазин Redux. Может быть, что-то вроде следующего будет работать лучше:

action$.pipe(
  ofType(START_FOO),
  switchMap({ url } =>
    from(fetch(url)).pipe(
      mergeMap(res => from(res.text()).pipe(
        mergeMap(reply => {
          try {
            reply = JSON.parse(reply)
          } catch (ignore) {
          }

          switch (res.status) {
            case 200: {
              return merge(
                of(takeAction(res, reply)),
                action$.pipe(
                  ofType(RESOLVE_TAKE, REJECT_TAKE),
                  map(({ type }) => {
                    if (type === RESOLVE_TAKE) {
                      return resolveFooAction()
                    } else {
                      return rejectFooAction()
                    }
                  }),
                ),
              )
            }
            // other res.status cases go here
          }
        })
      )
    )
  )
)

В приведенном выше тексте of(takeAction(res, reply)) - это , а не . Вместо этого он «возвращается» в магазин Redux. merge - это один из способов что-то выдать, одновременно создавая другую подписку на поток действий для временного прослушивания другого события.