RxJS订阅Subject报错：Actions必须是纯对象，二次执行触发

从错误信息和你的代码来看，问题的核心是Redux要求dispatch的必须是Plain Object类型的Action，但你的代码在某些分支下尝试dispatch了非Plain Object的内容，具体原因如下：

错误根源 #

在UploadSceneWithFile函数的onError回调中，你向Subject发送了一个字符串：

onError:()=>subject$.next('error'),

当上传出错时（第二次执行可能触发了这个错误分支），这个字符串会顺着Observable流到达你的Epic中的map操作。而你的map函数只处理了带有progress属性的对象，完全没处理这个字符串类型的错误值，最终要么直接尝试dispatch这个字符串，要么map返回undefined，这两种情况都违反了Redux的Action规则，从而触发了Actions must be plain objects错误。

另外还有一个潜在的异步时序问题：FileToScenePreview是异步操作，但你没有等待它完成就启动了S3上传，这可能导致scene.thumbName还未赋值就开始上传，后续逻辑可能会出现异常。

具体修复步骤 #

1. 统一Subject发送的数据格式

修改UploadSceneWithFile中的错误回调，确保发送的始终是Plain Object，比如给错误数据加上标识：

// 替换原来的onError回调
onError:()=>subject$.next({...scene, error: true, progress: null, message: 'Upload failed'}),

同时，记得处理FileToScenePreview本身可能出现的错误：

FileToScenePreview(scene).then(res => {
  scene.thumbName = res.thumbName
  // 这里初始化并启动上传逻辑
}).catch(err => {
  subject$.next({...scene, error: true, progress: null, message: err.message})
  subject$.complete()
})

2. 在Epic中处理所有分支的数据流

更新Epic中的map函数，处理错误分支，确保每个分支都返回合法的Plain Object Action：

.map(res => {
  // 优先处理错误情况
  if (res.error) {
    return { type: UPLOAD_SCENE_FAILURE, payload: res };
  } else if (res.progress > 0){
    return { type: UPLOAD_SCENE_PROGRESS, payload: res };
  } else if(res.progress === -1){
    return { type: UPLOAD_SCENE_SUCCESS, payload: res };
  }
  // 兜底处理意外情况，避免返回undefined
  return { type: UPLOAD_SCENE_UNKNOWN, payload: res };
})

3. 修复异步时序问题

把S3上传的初始化逻辑放到FileToScenePreview的then回调中，确保scene.thumbName赋值完成后再开始上传：

export function UploadSceneWithFile(scene){
  const subject$ = new Subject()
  scene.filename = scene.file.name
  scene.type = scene.file.type.match('image') ? 0 : 1

  FileToScenePreview(scene).then(res => {
    scene.thumbName = res.thumbName
    // 等待thumbName赋值完成后再创建上传器
    const uploader = new S3Upload({
      getSignedUrl: getSignedUrl,
      uploadRequestHeaders: {'x-amz-acl': 'public-read'},
      contentType: scene.file.type,
      contentDisposition: 'auto',
      s3path: 'assets/',
      onError:()=>subject$.next({...scene, error: true, progress: null, message: 'Upload failed'}),
      onProgress: (val)=>subject$.next({...scene,progress:val}),
      onFinishS3Put: ()=> {
        subject$.next({...scene,progress:-1})
        subject$.complete()
      },
    })
    uploader.uploadFile(scene.file)
  }).catch(err => {
    subject$.next({...scene, error: true, progress: null, message: err.message})
    subject$.complete()
  })

  return subject$.asObservable()
}

4. 给Epic添加错误捕获，避免Epic挂掉

使用catchError操作符处理Observable流中的错误，确保单次上传失败不会导致整个Epic停止工作：

import { of } from 'rxjs';
import { catchError } from 'rxjs/operators';

export function uploadSceneFile(action$, store) {
  return action$.ofType(CREATE_SCENE_SUCCESS)
    .mergeMap(({payload}) => 
      UploadSceneWithFile(payload)
        .map(res => {
          if (res.error) {
            return { type: UPLOAD_SCENE_FAILURE, payload: res };
          } else if (res.progress > 0){
            return { type: UPLOAD_SCENE_PROGRESS, payload: res };
          } else if(res.progress === -1){
            return { type: UPLOAD_SCENE_SUCCESS, payload: res };
          }
          return { type: UPLOAD_SCENE_UNKNOWN, payload: res };
        })
        .catchError(err => {
          // 捕获流中的错误，返回错误Action
          return of({ type: UPLOAD_SCENE_EPIC_ERROR, payload: err.message });
        })
    )
}

内容的提问来源于stack exchange，提问作者beek