import * as fs from 'fs' ; import { EventEmitter } from 'events' ; import Logger from '../../tools/env.logger' ; import Transform , { ITransformResult } from './file.map.transform' ; import BytesRowsMap from './file.map' ; export { ITransformResult } ; interface IPendingTask { chunk : string ; plugingId : number ; resolver : ( notify : boolean ) => void ; rejector : ( error : Error ) => void ; } export class FileWriter extends EventEmitter { public static Events = { ChunkWritten : 'ChunkWritten' , } ; private _file : string ; private _stream : fs . WriteStream | undefined ; private _logger : Logger ; private _guid : string ; private _blocked : boolean = false ; private _locked : boolean = false ; private _tasks : IPendingTask [ ] = [ ] ; private _map : BytesRowsMap ; constructor ( guid , file , map ) { super ( ) ; this . _guid = guid ; this . _file = file ; this . _map = map ; this . _logger = new Logger ( `StreamFileWriter: ${ this . _guid }` ) ; this . resume ( ) ; } public destroy ( ) { return new Promise ( ( resolve , reject ) => { this . _tasks . forEach ( task => task . resolver ( false ) ) ; this . _tasks = [ ] ; if ( this . _stream !== undefined ) { this . _stream . close ( ) ; this . _stream . destroy ( ) ; this . _stream = undefined ; } resolve ( ) ; } ) ; } public write ( chunk , pluginId ) { return new Promise ( ( resolve , reject ) => { if ( this . _isLocked ( ) && this . _tasks . length > 0 ) { const task = this . _tasks [ this . _tasks . length - 1 ] ; if ( task . plugingId === pluginId ) { task . chunk += chunk ; resolve ( false ) ; return this . _next ( ) ; } } this . _tasks . push ( { chunk : chunk , plugingId : pluginId , resolver : resolve , rejector : reject , } ) ; this . _next ( ) ; } ) ; } public stop ( ) { if ( this . _stream !== undefined ) { this . _stream . close ( ) ; this . _stream . destroy ( ) ; this . _stream = undefined ; } } public resume ( ) { if ( this . _stream !== undefined ) { return ; } this . _stream = fs . createWriteStream ( this . _file , { encoding : 'utf8' , flags : 'a' } ) ; this . _next ( ) ; } private _write ( chunk , pluginId ) { return new Promise ( ( resolve , reject ) => { const transform = new Transform ( { } , this . _guid , pluginId , { bytes : this . _map . getByteLength ( ) , rows : this . _map . getRowsCount ( ) } ) ; transform . convert ( chunk , ( converted ) => { if ( converted . output === '' ) { this . _logger . env ( `Chunk is empty. Nothing to write.` ) ; return resolve ( ) ; } if ( this . _stream === undefined ) { this . _logger . env ( `Cannot write to stream, while it's blocked.` ) ; return resolve ( ) ; } this . _map . add ( converted . map ) ; this . _stream . write ( converted . output , ( writeError ) => { if ( writeError ) { this . _logger . error ( `Fail to write data (pluginId "${ pluginId }") into stream file due error: ${ writeError . message }` ) ; return reject ( writeError ) ; } this . emit ( FileWriter . Events . ChunkWritten , converted . map ) ; return resolve ( ) ; } ) ; } ) ; } ) ; } private _next ( ) { if ( this . _tasks . length === 0 ) { return ; } if ( this . _stream === undefined ) { return ; } if ( this . _isLocked ( ) ) { return ; } const task = this . _tasks [ 0 ] ; this . _tasks . splice ( 0 , 1 ) ; this . _lock ( ) . _write ( task . chunk , task . plugingId ) . then ( ( ) => { task . resolver ( true ) ; this . _unlock ( ) . _next ( ) ; } ) . catch ( ( err ) => { task . rejector ( err ) ; this . _unlock ( ) . _next ( ) ; } ) ; } private _lock ( ) { this . _locked = true ; return this ; } private _unlock ( ) { this . _locked = false ; return this ; } private _isLocked ( ) { return this . _locked ; } }