Redux-Observable Epics返回动作数组报错及动作重复问题求助

先来看你遇到的核心错误：

Error: Actions must be plain objects. Use custom middleware for async actions.

这个错误的根源很清晰——Redux-Observable的Epic要求必须返回一个Observable流，而不是Promise或者动作数组。你的第一个Epic实现直接返回了Promise，Promise内部又返回动作数组，Epic无法识别这种格式，就会把整个Promise当成一个"action"来处理，自然就抛出了这个错误。

先回顾下你的代码问题点：
你的Epic入口是这样的：

const epic = (action$, store) => action$.ofType(String(key)).mergeMap(action => func(action, store))

而func里的具体实现直接返回了Promise：

[String(usersActions.updateUser)]: (action, store) => { 
  return authFetch(`${API_ROOT}/user/${get(action, 'payload.id')}`, { 
    method: 'put', 
    headers: { ... }, 
    body: ..., 
  }) 
  .then(resJson => { 
    if (resJson['status'] === 200) { 
      return [ 
        appActions.pushNotification('success', USER_UPDATE_SUCCESS), 
        appActions.setNextPath(`/users/${get(action, 'payload.id')}`), 
      ] 
    } else return [ 
      appActions.pushNotification( 
        'error', 
        USER_UPDATE_FAILED + ': ' + resJson['message'] 
      ), 
    ] 
  }) 
  .catch(() => { 
    return [appActions.pushNotification('error', USER_UPDATE_FAILED)] 
  }) 
},

解决错误的核心思路 #

必须把你的异步逻辑包装成Observable，让Epic能正确处理。下面给你两个可行的方案：

方案1：简化Observable流的创建（推荐）

用fromPromise把fetch请求转换成Observable，然后用switchMap（比mergeMap更适合请求场景，能自动取消未完成的重复请求）处理响应，最后用Observable.of直接发射多个动作——of支持传入多个参数，会自动把它们转换成连续发射的Observable流，不需要用concat拼接单个流。

另外注意catch操作符必须返回一个Observable，之前的代码里你漏掉了return，这也是导致后续问题的原因之一。

修改后的代码：

[String(usersActions.updateUser)]: (action, store) => { 
  return Rx.Observable.fromPromise(
    fetch(`${API_ROOT}/user/${get(action, 'payload.id')}`, { 
      method: 'put', 
      headers: {}, 
    }).then(res => res.json())
  )
  .switchMap(resJson => { 
    if (resJson['status'] === 200) {
      // 直接用of发射多个动作，自动转为Observable
      return Rx.Observable.of(
        appActions.pushNotification('success', USER_UPDATE_SUCCESS),
        appActions.setNextPath(`/users/${get(action, 'payload.id')}`)
      )
    } else {
      return Rx.Observable.of(
        appActions.pushNotification(
          'error',
          USER_UPDATE_FAILED + ': ' + resJson['message']
        )
      )
    }
  })
  .catch(() => {
    // 必须返回Observable，这里用of返回错误提示动作
    return Rx.Observable.of(
      appActions.pushNotification('error', USER_UPDATE_FAILED)
    )
  })
},

方案2：保留原Promise逻辑，转换数组为Observable

如果你想保留在Promise中返回动作数组的写法，可以在Promise外层用fromPromise包装，然后用mergeMap把数组转换成Observable流：

[String(usersActions.updateUser)]: (action, store) => { 
  return Rx.Observable.fromPromise(
    authFetch(`${API_ROOT}/user/${get(action, 'payload.id')}`, { 
      method: 'put', 
      headers: { ... }, 
      body: ..., 
    })
    .then(resJson => { 
      if (resJson['status'] === 200) { 
        return [ 
          appActions.pushNotification('success', USER_UPDATE_SUCCESS), 
          appActions.setNextPath(`/users/${get(action, 'payload.id')}`), 
        ] 
      } else {
        return [ 
          appActions.pushNotification(
            'error',
            USER_UPDATE_FAILED + ': ' + resJson['message']
          ), 
        ]
      }
    })
    .catch(() => { 
      return [appActions.pushNotification('error', USER_UPDATE_FAILED)] 
    })
  )
  // 将Promise返回的动作数组转换成Observable流
  .mergeMap(actions => Rx.Observable.from(actions))
},

关于动作重复分发的问题 #

你之前尝试的版本出现重复分发，主要有两个原因：

1. catch操作符没有返回Observable，导致流异常后没有正确处理，可能触发额外的流发射；
2. 使用mergeMap时，如果同一个action被多次触发，会同时存在多个未完成的请求，最终导致多个动作被分发——换成switchMap可以解决这个问题，它会取消之前未完成的请求，只保留最新的一个。

另外，确保你的Epic入口代码里，func返回的始终是Observable，而不是Promise或者其他类型。

内容的提问来源于stack exchange，提问作者Yassine S.