import { BaseTask , CancelledError , TaskState } from "@xmcl/task" ; import { openEntryReadStream } from "@xmcl/unzip" ; import { createWriteStream } from "fs" ; import { join } from "path" ; import { Readable , Writable } from "stream" ; import { Entry , ZipFile } from "yauzl" ; import { ensureFile , pipeline } from "./utils" ; export interface EntryResolver { ( entry ) : Promise < string > | string ; } export function getDefaultEntryResolver ( ) { return ( e ) => e . fileName ; } export class UnzipTask extends BaseTask < void > { private streams : [ Readable , Writable ] [ ] = [ ] ; private _onCancelled = ( ) => { } ; constructor ( readonly zipFile , readonly entries , destination , readonly resolver = getDefaultEntryResolver ( ) ) { super ( ) ; this . _to = destination ; } protected async handleEntry ( entry , relativePath ) { const file = join ( this . to ! , relativePath ) ; if ( this . _state === TaskState . Cancelled ) { throw new CancelledError ( ) ; } const readStream = await openEntryReadStream ( this . zipFile , entry ) ; if ( this . isCancelled ) { throw new CancelledError ( ) ; } if ( this . _state === TaskState . Paused ) { readStream . pause ( ) ; } await ensureFile ( file ) ; const writeStream = createWriteStream ( file ) ; readStream . on ( "data" , ( buf ) => { this . _progress += buf . length ; this . update ( buf . length ) ; } ) ; await pipeline ( readStream , writeStream ) ; } protected async runTask ( ) { const promises = [ ] ; for ( const e of this . entries ) { const path = await this . resolver ( e ) ; if ( this . isCancelled ) { throw new CancelledError ( ) ; } this . _total += e . uncompressedSize ; promises . push ( this . handleEntry ( e , path ) ) ; } this . update ( 0 ) ; try { await Promise . all ( promises ) ; } catch ( e ) { if ( e instanceof CancelledError ) { this . _onCancelled ( ) ; } throw e ; } } protected cancelTask ( ) { for ( const [ read , write ] of this . streams ) { read . unpipe ( ) ; read . destroy ( new CancelledError ( ) ) ; this . zipFile . close ( ) ; write . destroy ( new CancelledError ( ) ) ; } return new Promise ( ( resolve ) => { this . _onCancelled = resolve ; } ) ; } protected async pauseTask ( ) { const promise = Promise . all ( this . streams . map ( ( [ read ] ) => new Promise ( ( resolve ) => read . once ( "pause" , resolve ) ) ) ) ; for ( const [ read ] of this . streams ) { read . pause ( ) ; } await promise ; } protected async resumeTask ( ) { const promise = Promise . all ( this . streams . map ( ( [ read ] ) => new Promise ( ( resolve ) => read . once ( "readable" , resolve ) ) ) ) ; for ( const [ read ] of this . streams ) { read . resume ( ) ; } await promise ; } }