中途派遣行动

Question

RxJS新手。和redux-observable。

我遇到麻烦我试图在下面的代码中调用of(takeAction())中途的动作，然后听取响应的动作，无论是RESOLVE_TAKE类型还是REJECT_TAKE类型。但是我的of(takeAction)没有触发，有人知道如何修复下面的代码吗？

除了这个问题，还有一个好处：关于如何重构我的代码的任何风格建议，我不确定它是否是最干净最易读的方式。我正在进行取指，然后对不同的状态代码进行switch，然后获取res（fetch的响应）和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是一种流出某种东西的方式，同时创建另一个对动作流的订阅以暂时监听另一个事件。