javascript - 使用可管道化的 RxJs 操作符实现复杂的文件上传场景？

Question

我有一个场景，我需要将一系列文件按顺序上传到 Web 服务器，每个文件都涉及分 block 。以程序方式实现时，上传过程非常简单，大致如下:

foreach file in files:
  const result = await createMultipartUpload();

  do {
    const chunk = takeChunk();
    const result = await uploadChunk(chunk);

    chunksRemain = doChunksRemain();
  } while (chunksRemain);

  await completeUpload();

也就是说，对于每个顺序上传的过程，会发生一个创建请求，顺序上传多个 block ——但我不知道会上传多少 block ——然后发生一个完成请求。

我需要使用 RxJs 将其转换为响应式代码。我实际上适用于单 block 上传，只需要一个 block 。但我还没有弄清楚如何动态生成上传 block 可观察对象并将它们推送到某种数组中，然后可以从中检索它们。

这是我到目前为止所拥有的:

  // Contains single request. 
  public createMultipartUpload(file: File): Observable<File> {
    return this.filesHttp.store(file);
  }

  // Contains two HTTP requests. Generates a presigned URL, then uploads the chunk with the presaged URL.
  public uploadChunk(file: File): Observable<File> {
    let chunk: Blob;

    // Determine the bytes of the file to take and assign to chunk.
    if (file.originalFile.size < this.chunkSize) {
      chunk = file.originalFile;
    } else {
      chunk = file.originalFile.slice(file.currentByteOffset, file.currentByteOffset + this.chunkSize);
    }

    return this.filesHttp.getPresignedUrlForUpload(file)
      .pipe(
        concatMap(res => {
          return this.filesHttp.upload(res.uri, chunk);
        }),
        map(f => {
          // Set next chunk to take
          f.currentByteOffset += this.chunkSize;
          return f;
        })
      );
  }

  public completeUpload(file: File): Observable<MultipartCompletionResponse> {
    return this.filesHttp.complete(file);
  }

  from(this.files)
    .pipe(
      concatMap(f => this.createMultipartUpload(f)),
      switchMap(f => this.uploadChunk(f)),
      concatMap(f => this.completeUpload(f))
    ).subscribe(output => {
      // done?
    });

在这种情况下，如何按顺序上传所有 block ，而不仅仅是第一个 block ？

Answer 1

使用expand处理动态数量的 Observable。 expand递归地映射到 Observable 并接收其输出作为下一个输入。通过返回 EMPTY 结束此递归.

// Upload mutiple files one after another with concat
concat(...this.files.map(file => uploadFile(file))).subscribe(console.log);

// Upload single file in multiple chunks with expand
uploadFile(file): Observable<any> {
  return createMultipartUpload(file).pipe(   
    // add this for a do...while logic where doChunksRemain() shouldn't be called for 
    // the first uploadChunk()
    //switchMap(f => uploadChunk(f)),

    // Upload next chunks until doChunksRemain() returns false
    expand(f => doChunksRemain() ? uploadChunk(f) : EMPTY),
    // only take the result of the last uploaded chunk
    last(),
    // map to completeUpload when we receive the last result
    switchMap(f => completeUpload())
  );
}

https://stackblitz.com/edit/rxjs-d5vpyf