import { parse } from 'date-fns' ; import * as fs from 'fs' ; import { type , pick } from 'ramda' ; import * as util from 'util' ; import { NosBaseClient } from '../client' ; import { MAX_PART_LENGTH } from '../lib/constant' ; import { Callbackable , compactObject , normalizeArray } from '../lib/util' ; import { Callback } from '../type/callback' ; import { AbortMultipartUploadParams , CompleteMultipartParams , InitMultipartUploadParams , ListMultipartParams , ListMultipartResult , ListPartsOptions , ListPartsResult , MultipartUpload , MultipartUploadObject , Part , PutBigObjectParams , UploadMultipartParams , } from '../type/multipart-upload' ; export class NosClientMultipartUploadExt extends NosBaseClient { return initMultipartUpload ( params ) ; initMultipartUpload ( params , cb ) ; @ Callbackable async initMultipartUpload ( params ) { const { bucket , headers , resource } = this . validateParams ( params ) ; Object . assign ( resource , { uploads : true , } ) ; const result = await this . requestBody ( 'post' , headers , resource ) ; return result . initiateMultipartUploadResult . uploadId ; } return abortMultipartUpload ( params ) ; abortMultipartUpload ( params , cb ) ; @ Callbackable async abortMultipartUpload ( params ) { const { bucket , headers , resource } = this . validateParams ( params ) ; Object . assign ( resource , { uploadId : params . uploadId , } ) ; const result = await this . requestBody ( 'delete' , headers , resource ) ; return true ; } return uploadMultipart ( params ) ; uploadMultipart ( params , cb ) ; @ Callbackable async uploadMultipart ( params ) { const { bucket , headers , resource } = this . validateParams ( params ) ; Object . assign ( resource , { partNumber : params . partNumber , uploadId : params . uploadId , } ) ; const resp = await this . request ( 'put' , headers , resource , params . body ) ; return { partNumber : params . partNumber , eTag : ( resp . headers . get ( 'etag' ) as string ) . slice ( 1 , - 1 ) , size : params . body . length , lastModified : parse ( resp . headers . get ( 'date' ) as string ) , } ; } /**
     * 获取一个对象的所有分片，必须保证这个对象并未完成分片的全部上传
     */ listParts ( params ) ; listParts ( params , cb ) ; @ Callbackable async listParts ( params ) { const { bucket , headers , resource } = this . validateParams ( params ) ; Object . assign ( resource , compactObject ( { uploadId : params . uploadId , 'max-parts' : params . limit , 'part-number-marker' : params . marker , } ) ) ; const { listPartsResult : result } = await this . requestBody ( 'get' , headers , resource ) ; let parts = result . part ; if ( type ( parts ) !== 'Array' ) { parts = [ parts ] ; } for ( const part of parts ) { part . lastModified = parse ( part . lastModified ) ; } return { ... ( pick ( [ 'isTruncated' , 'owner' , 'storageClass' , 'bucket' ] , result ) as any ) , limit : result . maxParts , nextMarker : result . nextPartNumberMarker , items : parts , } ; } /**
     * 获取所有的未完成的分片对象
     */ listMultipartUpload ( params ? ) ; listMultipartUpload ( params , cb ) ; @ Callbackable async listMultipartUpload ( params = { } ) { const { bucket , headers , resource } = this . validateParams ( params ) ; Object . assign ( resource , compactObject ( { uploads : true , prefix : params . prefix , 'key-marker' : params . marker , 'max-uploads' : params . limit , } ) ) ; if ( params . limit ) { headers [ 'max-uploads' ] = params . limit ; } const { listMultipartUploadsResult : result } = await this . requestBody ( 'get' , headers , resource ) ; let uploads = normalizeArray ( result . upload ) ; if ( type ( uploads ) !== 'Array' ) { uploads = [ uploads ] ; } for ( const upload of uploads ) { upload . initiated = parse ( upload . initiated ) ; } return { ... ( pick ( [ 'isTruncated' ] , result ) as any ) , items : uploads , nextMarker : result . nextKeyMarker , bucket : result . bucket , limit : params . limit || 1000 , } ; } /**
     * 完成分片上传。这是分片上传的结束，如果你上传完成之后但是没有执行完成上传，那么服务器并不会对分片进行拼接。
     */ completeMultipartUpload ( params ) ; completeMultipartUpload ( params , cb ) ; @ Callbackable async completeMultipartUpload ( params ) { const { bucket , headers , resource } = this . validateParams ( params ) ; Object . assign ( resource , { uploadId : params . uploadId , } ) ; const result = await this . requestBody ( 'post' , headers , resource , { completeMultipartUpload : { part : params . parts . map ( part => ( { partNumber : part . partNumber , eTag : part . eTag } ) ) , } , } ) ; return result . completeMultipartUploadResult ; } /**
     * 使用分片上传大文件。如果你不想自己手写分片文件的上传，你可以使用这个接口。
     * 自动创建、上传、完成分片，省时省力，方便快捷。
     */ putBigObject ( params ) ; putBigObject ( params , cb ) ; @ Callbackable async putBigObject ( params ) { const uploadId = await this . initMultipartUpload ( params ) ; const stream = 'body' in params ? params . body : fs . createReadStream ( params . file ) ; const lengthComputable = 'file' in params ; const { parallel = Infinity , maxPart = MAX_PART_LENGTH } = params ; let totalLength = 0 ; if ( 'file' in params ) { const stat = await util . promisify ( fs . stat ) ( params . file ) ; totalLength = stat . size ; } const rootParams = { uploadId , objectKey : params . objectKey , bucket : params . bucket , } ; const bufs = [ ] ; const partPromises = [ ] ; let length = 0 ; let partNumber = 1 ; let aborted = false ; let uploadedLength = 0 ; let onProgress = params . onProgress ; let workers = 0 ; const uploadPart = async ( ) : Promise < Part > => { const tBuf = Buffer . concat ( bufs ) ; const tPartNumber = partNumber ++ ; bufs . length = 0 ; length = 0 ; if ( ++ workers >= parallel ) { stream . pause ( ) ; } const part = await this . uploadMultipart ( { ... rootParams , partNumber : tPartNumber , body : tBuf , } ) ; workers -- ; if ( stream . isPaused ( ) ) { stream . resume ( ) ; } uploadedLength += tBuf . length ; if ( onProgress ) { onProgress ( { lengthComputable , uploaded : uploadedLength , total : totalLength , } ) ; } return part ; } ; return await new Promise < MultipartUploadObject > ( ( resolve , reject ) => { const completeUpload = async ( ) => { const parts = await Promise . all ( partPromises ) ; const resp = await this . completeMultipartUpload ( { ... rootParams , parts , } ) ; resolve ( resp ) ; } ; const abortUpload = ( e ) => { if ( aborted ) { return ; } stream . removeListener ( 'data' , onData ) ; aborted = true ; reject ( e ) ; } ; const onData = ( data ) => { if ( data . length + length > maxPart ) { const partPromise = uploadPart ( ) ; partPromise . catch ( abortUpload ) ; partPromises . push ( partPromise ) ; } bufs . push ( data ) ; length += data . length ; } ; stream . on ( 'data' , onData ) ; stream . once ( 'end' , ( ) => { if ( length ) { const partPromise = uploadPart ( ) ; partPromise . catch ( abortUpload ) ; partPromises . push ( partPromise ) ; } completeUpload ( ) . catch ( abortUpload ) ; } ) ; stream . once ( 'error' , abortUpload ) ; } ) ; } }