文件上传进度未在可观察对象中更新或在可观察对象准备好之前完成

Question

下面是调用下面函数的代码片段。这是一个允许人们将文件从系统拖放到网站的过程。它显示所有文件的列表，并在加载时显示进度条。大多数时候它工作正常，但是当有大量文件时，我运行会遇到一些问题。我有一个正在加载的测试目录，其中包含 100 多个文件。第一个加载的文件非常小，所以看起来它们是在设置 observable 之前加载的，因为进度条没有显示任何进度并且 forkJoin 没有完成但是如果我在系统上查看文件实际加载。

我是不是没有正确设置主题？有没有更好的方法来跟踪正在上传的文件的进度？任何帮助将不胜感激。

if (this.files.size > 0) {
  this.progress = await this.uploadService.dndUpload(
    this.files, this.currentDir, this.currentProject, timestamp
  );
  let allProgressObservables = [];
  for (let key in this.progress) {
    allProgressObservables.push(this.progress[key].progress);
  }

  this.sfUploadSnackbar.openSnackbar(this.files, this.progress);

  forkJoin(allProgressObservables).subscribe(async end => {
    this.sfUploadSnackbar.closeSnackbar();
    this.uploadService.clearUploadDir(this.currentProject, timestamp)
      .subscribe();
    this.uploadInProgress = false;
    this.getFiles();
  });
}





 async dndUpload(files: Set<any>, dir: string, projectId: number, timestamp: number) {
    const status: { [key: string]: { progress: Observable<number> } } = {};

    for (let it = files.values(), file = null; file = it.next().value;) {

      let params = new HttpParams()
        .set('dir', dir)
        .set('path', file.fullPath.replace(file.name,''))
        .set('projectId', projectId.toString())
        .set('timestamp', timestamp.toString());
      let f: File = await new Promise((resolve, reject) => file.file(resolve, reject))
      const formData: FormData = new FormData();
      formData.append('file', f, f.name);

      const req = new HttpRequest('POST', '/api/dndUpload', formData, {
        reportProgress: true, params
      });

      const progress = new Subject<number>();

      status[file.name] = {
        progress: progress.asObservable()
      };

      this.http.request(req).subscribe(event => {
        if (event.type === HttpEventType.UploadProgress) {

          const percentDone = Math.round(100 * event.loaded / event.total);

          progress.next(percentDone);
        } else if (event instanceof HttpResponse) {

          progress.complete();
        }
      });
    }

    return status;
  }

Answer 1

为了 forkJoin 完成，您必须确保提供的所有可观察对象都已完成。可能发生的情况是 forkJoin 订阅 allProgressObservables 的 Subject 太晚了。

我假设 this.sfUploadSnackbar.openSnackbar(this.files, this.progress); 将订阅 this.progress 以便接收每个文件的百分比。

这是一个想法：

dndUpload (files: Set<...>/* ... */): Observable<any> {
  // Note that there are no subscriptions
  return [...files].map(
    f => from(new Promise((resolve, reject) => file.file(resolve, reject)))
      .pipe(
        map(f => (new FormData()).append('file', f, f.name)),
      )
  )
}

const fileObservables$ = this.dndUpload(files);

const progressObservables$ = fileObservables$.map(
  (file$, fileIdx) => file$.pipe(
    switchMap(formData => {
      const req = /* ... */;

      return this.http.request(req)
        .pipe(
          filter(event.type === HttpEventType.UploadProgress),
          // Getting the percent
          map(e => Math.round(100 * e.loaded / e.total)),
          tap(percent => this.updatePercentVisually(percent, fileIdx))
        )
    })
  )
);

// Finally subscribing only once to the observables
forkJoin(progressObservables$).subscribe(...);

注意有一些变化：

我不再为每个文件使用 Subject
- 因此，this.sfUploadSnackbar.openSnackbar(this.files, this.progress); 必须用另一种方法替换(this.updatePercentVisually)
文件 observables 的订阅只发生在一次地方，在 forkJoin;

this.http.request 将在请求完成时完成，因此 forkJoin 也应该能够完成，允许您执行 'cleanups'（删除加载进度等...） .

Answer 2

我接受了上面@Andrei 的回答，因为如果不是他，我可能永远不会停止用头撞墙。我在下面包含了我最终使用的实际代码，以供将来可能运行遇到类似问题的其他人参考。可能还有很多改进可以做，但至少它有效。

if (this.files.size > 0) {
  const status: { [key: string]: { progress: BehaviorSubject<number> } } = {};
  const allProgressObservables = [];
  for (let f of this.files) {
    const progress = new BehaviorSubject<number>(0);
    status[f.fullPath] = {progress: progress};
    allProgressObservables.push(status[f.fullPath].progress);
  }
  this.sfUploadSnackbar.openSnackbar(this.files, status);
  forkJoin(allProgressObservables)
    .subscribe(() => {
      this.sfUploadSnackbar.closeSnackbar();
    });
  from(this.files)
    .pipe(
      mergeMap(async (file) => {
        let params = new HttpParams()
          .set('dir', this.currentDir)
          .set('path', file.fullPath.replace(file.name,''))
          .set('projectId', this.currentProject.toString())
          .set('timestamp', timestamp.toString());
        let f: File = await new Promise((resolve, reject) => file.file(resolve))
        const formData: FormData = new FormData();
        formData.append('file', f, f.name);
        const req = new HttpRequest('POST', '/api/dndUpload', formData, {
          reportProgress: true, params
        });
        return this.http.request(req)
          .subscribe(event => {
            if (event.type === HttpEventType.UploadProgress) {
              const percentDone = Math.round(100 * event.loaded / event.total);
              status[file.fullPath].progress.next(percentDone);
            } else if (event instanceof HttpResponse) {
              status[file.fullPath].progress.complete();
            }  
          });
      })
    ).subscribe();
}

文件上传进度未在可观察对象中更新或在可观察对象准备好之前完成

File upload progress not being updated in the observable or completed before the observable is ready

angular

rxjs

subject-observer