Блог

Я внедряю набор API-интерфейсов для групповой загрузки из пакета SDK AWS S3 для обработки больших загрузок файлов. Согласно документации SDK, загрузка нескольких частей должна быть явно закрыта после загрузки всех частей. Этот вызов API закрытия — CompleteMultipartUpload, требует отображения фрагментов, которое представляет собой отображение поля «Etag», возвращаемого API uploadPart, и номера детали. Поскольку заказ деталей должен быть отображен кодом — я попытался использовать concat, чтобы убедиться, что следующая деталь загружается только после завершения текущей. Однако даже при использовании concat все отдельные части загружаются одновременно и возвращаются в разное время. Это не помогает в идентификации номеров деталей.

Код выглядит так, как показано ниже — я создаю массив наблюдаемых объектов для загружаемых частей и пытаюсь их объединить. Использование RxJS v6.6.3 и AWS SDK версии v2 (в lib-хранилище SDK v3 есть проблема с исполнителем браузера, иначе это сократило бы код).

Почему конкат здесь не работает, это проблема с обработкой массива наблюдаемых объектов?

   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,
              Key: filePath   file.name,
              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,
        Key: filePath   file.name,
        UploadId: multipartUploadId,
        MultipartUpload: {
          Parts: ETagMap,
        },
      };
      return from(
        s3Obj
          .completeMultipartUpload(s3CompleteMultipartUploadParams)
          .promise()
      );
    })
  )