Блог

Новое в RxJS. И redux-наблюдаемый.

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

Помимо этого вопроса, бонус: Любые советы по стилю о том, как реструктурировать мой код, я не уверен, что это самый чистый и читаемый способ. Я выполняю выборку, затем a 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
                    }
                })
            )
        )
    )
)
  

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