javascript - 如何使用浏览器的 aws-sdk V3 (javascript) 跟踪上传到 S3 的进度

Question

我可以在网上找到很多关于如何使用 aws-sdk V2 跟踪上传到 S3 的进度的资源，例如:

.on('httpUploadProgress', event => {}

但是自从我将 aws-sdk 更新到 V3 后，就不再有监听器了。我相信我现在必须使用中间件功能，但是我尝试了一些东西并没有奏效。我也深入了 API reference docs和 github repository没有成功。
我当前的代码是这样的:

import { S3Client, PutObjectCommand } from '@aws-sdk/client-s3';

export const UploadToS3 = (credentials, fileData) => {

    const s3 = new S3Client({
        region: credentials.region,
        credentials: {
            accessKeyId: credentials.access_key,
            secretAccessKey: credentials.secret_key,
            sessionToken: credentials.session_token,
        }
    });

    return new Promise((resolve) => {
        s3.send(new PutObjectCommand({
            Bucket: credentials.bucket,
            Key: credentials.file,
            Body: fileData,
        }));
    });
};

任何帮助，将不胜感激

Answer 1

我遇到了完全相同的问题(从 aws-sdk v2 切换到 v3)，发现这是因为库使用了 Fetch所有 HTTP 请求的 API 和 Fetch does not (yet) support tracking upload progress
为了解决这个问题，我交换了 Fetch作者:好老 XMLHttpRequest至少对于 PUT请求，您可以通过提供自定义 来完成请求处理程序 初始化 S3Client 时。

import { S3Client } from '@aws-sdk/client-s3';

const myHttpHandler = new MyHttpHandler();
myHttpHandler.onProgress$.subscribe(progress => {
  const percentComplete = progress.progressEvent.loaded / progress.progressEvent.total * 100;
  console.log('upload progress', percentComplete);
});

const myClient = new S3Client({
  endpoint: this.configService.s3Api,
  region: 'eu',
  credentials: { ... },
  requestHandler: myHttpHandler
});

自定义请求处理程序只是扩展了 FetchHttpHandler来自@aws-sdk/fetch-http-handler。如果方法是 PUT并且有一个主体(所以我们要上传一些东西)，它使用自定义 XHR 处理程序 - 否则它只使用 Fetch处理程序来自它的 super类(class)。
在 XHR 处理程序中，您可以将某些内容绑定(bind)到 progress XHR 处理程序的事件 - 在我的情况下，我发出一个 rxjs Subject我可以在自定义处理程序之外使用它。

import { FetchHttpHandler, FetchHttpHandlerOptions } from '@aws-sdk/fetch-http-handler';
import { HeaderBag, HttpHandlerOptions } from '@aws-sdk/types';
import { buildQueryString } from '@aws-sdk/querystring-builder';
import { HttpResponse, HttpRequest } from '@aws-sdk/protocol-http';
import { Subject } from 'rxjs';

class MyHttpHandler extends FetchHttpHandler {
  private myRequestTimeout;

  onProgress$: Subject<{ path: string, progressEvent: ProgressEvent }> = new Subject();

  constructor({ requestTimeout }: FetchHttpHandlerOptions = {}) {
    super({ requestTimeout });
    this.myRequestTimeout = requestTimeout;
  }

  handle(request: HttpRequest, { abortSignal }: HttpHandlerOptions = {}): Promise<{ response: HttpResponse }> {
    // we let XHR only handle PUT requests with body (as we want to have progress events here), the rest by fetch
    if (request.method === 'PUT' && request.body) {
      return this.handleByXhr(request, { abortSignal });
    }
    return super.handle(request, { abortSignal });
  }

  /**
   * handles a request by XHR instead of fetch
   * this is a copy the `handle` method of the `FetchHttpHandler` class of @aws-sdk/fetch-http-handler
   * replacing the `Fetch`part with XHR
   */
  private handleByXhr(request: HttpRequest, { abortSignal }: HttpHandlerOptions = {}): Promise<{ response: HttpResponse}> {
    const requestTimeoutInMs = this.myRequestTimeout;

    // if the request was already aborted, prevent doing extra work
    if (abortSignal?.aborted) {
      const abortError = new Error('Request aborted');
      abortError.name = 'AbortError';
      return Promise.reject(abortError);
    }

    let path = request.path;
    if (request.query) {
      const queryString = buildQueryString(request.query);
      if (queryString) {
        path += `?${queryString}`;
      }
    }

    const { port, method } = request;
    const url = `${request.protocol}//${request.hostname}${port ? `:${port}` : ''}${path}`;
    // Request constructor doesn't allow GET/HEAD request with body
    // ref: https://github.com/whatwg/fetch/issues/551
    const body = method === 'GET' || method === 'HEAD' ? undefined : request.body;
    const requestOptions: RequestInit = {
      body,
      headers: new Headers(request.headers),
      method,
    };


    const myXHR = new XMLHttpRequest();
    const xhrPromise = new Promise<{headers: string[], body: Blob, status: number}>((resolve, reject) => {
      try {
        myXHR.responseType = 'blob';

        // bind the events
        myXHR.onload = progressEvent => {
          resolve({
            body: myXHR.response,
            headers: myXHR.getAllResponseHeaders().split('\n'),
            status: myXHR.status
          });
        };
        myXHR.onerror = progressEvent => reject(new Error(myXHR.responseText));
        myXHR.onabort = progressEvent => {
          const abortError = new Error('Request aborted');
          abortError.name = 'AbortError';
          reject(abortError);
        };

        // progress event musst be bound to the `upload` property
        if (myXHR.upload) {
          myXHR.upload.onprogress = progressEvent => this.onProgress$.next({ path, progressEvent });
        }


        myXHR.open(requestOptions.method, url);
        // append headers
        if (requestOptions.headers) {
          (requestOptions.headers as Headers).forEach((headerVal, headerKey, headers) => {
            if (['host', 'content-length'].indexOf(headerKey.toLowerCase()) >= 0) {
              // avoid "refused to set unsafe header" error message
              return;
            }

            myXHR.setRequestHeader(headerKey, headerVal);
          });
        }
        myXHR.send(requestOptions.body);
      } catch (e) {
        console.error('S3 XHRHandler error', e);
        reject(e);
      }
    });

    const raceOfPromises = [
      xhrPromise.then((response) => {
        const fetchHeaders = response.headers;
        const transformedHeaders: HeaderBag = {};

        fetchHeaders.forEach(header => {
          const name = header.substr(0, header.indexOf(':') + 1);
          const val =  header.substr(header.indexOf(':') + 1);
          if (name && val) {
            transformedHeaders[name] = val;
          }
        });

        const hasReadableStream = response.body !== undefined;

        // Return the response with buffered body
        if (!hasReadableStream) {
          return response.body.text().then(body => ({
            response: new HttpResponse({
              headers: transformedHeaders,
              statusCode: response.status,
              body,
            }),
          }));
        }
        // Return the response with streaming body
        return {
          response: new HttpResponse({
            headers: transformedHeaders,
            statusCode: response.status,
            body: response.body,
          }),
        };
      }),
      this.requestTimeoutFn(requestTimeoutInMs),
    ];
    if (abortSignal) {
      raceOfPromises.push(
        new Promise<never>((resolve, reject) => {
          abortSignal.onabort = () => {
            myXHR.abort();
          };
        })
      );
    }
    return Promise.race(raceOfPromises);
  }

  private requestTimeoutFn(timeoutInMs = 0): Promise<never> {
    return new Promise((resolve, reject) => {
      if (timeoutInMs) {
        setTimeout(() => {
          const timeoutError = new Error(`Request did not complete within ${timeoutInMs} ms`);
          timeoutError.name = 'TimeoutError';
          reject(timeoutError);
        }, timeoutInMs);
      }
    });
  }
}