Angular 11 - 用于 AWS S3 分段文件上传的 RxJs concat 不等待请求完成

如何解决Angular 11 - 用于 AWS S3 分段文件上传的 RxJs concat 不等待请求完成

我正在从 AWS S3 开发工具包实施分段上传 API 集来处理大文件上传。根据SDK文档，分段上传需要在所有部分上传后明确关闭。这个闭包 API 调用 - completeMultipartUpload，需要块的映射，它是uploadPart API 返回的“Etag”字段和部件号的映射。因为部分顺序需要由代码映射 - 我尝试使用 concat 来确保仅在当前部分完成后才上传下一部分。但是，即使使用 concat，所有单独的部分也是同时上传的，并且它们在不同的时间返回。这无助于识别部件号。

代码如下 - 我正在为 uploadParts 构建一组 Observables 并尝试连接它们。使用 RxJs v6.6.3 和 AWS SDK 版本 v2（SDK v3 lib-storage 存在浏览器 exec 的问题，否则会缩短代码）。

为什么 concat 在这里不起作用，是否是处理一组 observables 的问题？

  const FILE_CHUNK_SIZE = 5 * 1024 * 1024;
  s3SessionToken$ // API call that returns token for S3 upload
  .pipe(
    // CREATE MULTIPART UPLOAD
    concatMap((tokenVal) => {
      // Create S3 Bucket Object
      s3Obj = new S3({
        accessKeyId: tokenVal.accessKeyId,secretAccessKey: tokenVal.secretAccessKeyId,sessionToken: tokenVal.sessionToken,});
      const s3createMultipartUploadParams: S3.CreateMultipartUploadRequest = {
        Bucket: bucketName,Key: filePath + file.name,ContentType: file.type,};

      // Begin Multipart Upload
      return from(
        s3Obj.createMultipartUpload(s3createMultipartUploadParams).promise()
      );
    }),// START UPLOADING INDIVIDUAL PARTS
    concatMap(
      (createMultipartUploadOutput: S3.CreateMultipartUploadOutput) => {
        multipartUploadId = createMultipartUploadOutput.UploadId;

        numParts = Math.ceil(file.size / FILE_CHUNK_SIZE);
        console.log('numParts',numParts);

        const uploadPartsObservables: Observable<
          S3.UploadPartOutput
        >[] = [];
        for (let i = 0; i < numParts; i++) {
          const start = i * FILE_CHUNK_SIZE;
          const end = (i + 1) * FILE_CHUNK_SIZE;
          const blob =
            i < numParts ? file.slice(start,end) : file.slice(start);

          const uploadPartParams: S3.UploadPartRequest = {
            Body: blob,Bucket: environment.AWS_BUCKET,Key: createMultipartUploadOutput.Key,PartNumber: i + 1,UploadId: createMultipartUploadOutput.UploadId,};

          uploadPartsObservables.push(
            from(
              s3Obj
                .uploadPart(uploadPartParams)
                .promise()
            )
          );
        }

        return concat(...uploadPartsObservables). // All the uploadPart calls start at the same time
        pipe(
          map((uploadPartResult: S3.UploadPartOutput,index: number) => { // the Index here seems to reflect the order of the response received not the request order
            const PartTagMapObj: S3.Part = {
              ETag: uploadPartResult.ETag,PartNumber: index + 1,};

            return PartTagMapObj;
          }),catchError((err) => {
            console.log('error in upload of file : ',err);
            const abortParams: S3.AbortMultipartUploadRequest = {
              Bucket: bucketName,UploadId: multipartUploadId,};
            return this.abortMultipartUpload(s3Obj,abortParams);
          }),toArray()
        );
      }
    ),retry(3),// Needed for createMultipartUpload,which fails on first try sometimes
    concatMap((ETagMap: S3.Parts) => {
      console.log('err',ETagMap);

      const s3CompleteMultipartUploadParams: S3.CompleteMultipartUploadRequest = {
        Bucket: bucketName,MultipartUpload: {
          Parts: ETagMap,},};
      return from(
        s3Obj
          .completeMultipartUpload(s3CompleteMultipartUploadParams)
          .promise()
      );
    })
  )