redux-observable 使用 RxJS 为 ajax 调用发出进度 Action

Question

我一直在努力解决这个问题，觉得我有一个根本的误解。我正在使用 redux-observable图书馆在 React胶水redux连同RxJS用于处理异步。我的问题是我必须处理大量上传，我想在文件加载时显示进度。

函数uploadFileEpic需要返回 Observable<Action>与 redux-observable 一起工作. uploadObservable代表我想要完成的工作流程。如果我只是返回 uploadObservable上传有效，但我没有收到任何 handleUploadFileProgress来自 progressSubscriber 的行动在 ajax称呼。理想情况下是 progressSubscriber将向另一个 observable 添加元素，我可以将其与 uploadObservable 合并.你看到我试图使用 merge在这里，但 TypeScript 编译器提示说返回值不能分配给 ObservableInput .

我一直在兜圈子，所以我觉得我的理解必须从根本上偏离。我觉得我缺少一些简单的 RxJS魔法在这里。谢谢您的帮助!

import { Observable, Observer, Subscriber, Subject, of } from 'rxjs';
import { ajax } from 'rxjs/ajax';
import { ofType } from 'redux-observable';
import { catchError, delay, map, mergeMap, tap, merge } from 'rxjs/operators';
import { apis } from '../../config';

export const enum ActionType {
  InitialFileUpload
  FileProgress
  UploadFileSuccess
  UploadFileFail
}

const handleInitialFileUpload = (file: File, timeLimit: number) => ({
  type: ActionType.InitialFileUpload,
  file,
  timeLimit
})

const handleFileProgress = (file: File, percentComplete: number) => ({
  type: ActionType.FileProgress,
  file,
  percentComplete
})

const handleUploadFileSuccess = (file: File, timeLimit: number) => ({
  type: ActionType.UploadFileSuccess,
  file,
  timeLimit
})

const handleUploadFileFail = (file: File, timeLimit: number) => ({
  type: ActionType.UploadFileFail,
  file,
  timeLimit
})


export const uploadFileEpic= action$ =>
  action$.pipe(
    ofType(ActionType.InitialFileUpload),
    mergeMap((action: any) => {
      const { file, timeLimit } = action;
      const data = new FormData()
      data.append('importFile', file, file.name)
      data.append('timeLimit', timeLimit)
      const progressSubject = new Subject();

      const ajaxRequest = {
        url: apis.gateway.run,
        method: 'POST',
        body: data,
        headers: {},
        progressSubscriber: Subscriber.create(
          (e: ProgressEvent) => {
            const percentComplete = Math.round((e.loaded / e.total) * 100)
            console.log("Progress event")
            progressSubject.next(handleUploadFileProgress(file, percentComplete))
          }
        )
      }

      const uploadObservable = ajax(ajaxRequest)
        .pipe(
          map(res => handleUploadFileSuccess(file)),
          delay(SUCCESSFUL_UPLOAD_NOTIFICATION_LENGTH),
          map(() => handleUploadFileRemove(file)),
          catchError(error => of(handleUploadFileFail(file, error.response)))
        )

      return merge(uploadObservable, progressSubject)
      }
    )
  )

Answer 1

您似乎正在导入 merge来自 rxjs/operators .那里，merge被视为运算符，因此返回 OperatorFunction .通过简单地从 rxjs 导入你会得到正确返回 Observable 的静态合并这将被您的 mergeMap 弄平.