import * as Stream from 'stream' ; import { IRange , IMapItem } from '../file.map' ; import Logger from '../../../tools/env.logger' ; export interface ITransformResult { output : string ; bytesSize : number ; map : IMapItem ; } export { IMapItem } ; export type TBeforeCallbackHandle = ( results : ITransformResult ) => Promise < void > ; export interface IMapChunkEvent { total : number ; chunk : number ; map : IMapItem ; } export interface IOffset { bytes : number ; rows : number ; } export default class Transform extends Stream . Transform { public static Events = { found : 'found' , } ; private _logger : Logger ; private _rest : string = '' ; private _streamId : string ; private _beforeCallbackHandle : TBeforeCallbackHandle | undefined ; private _offsets : IOffset = { bytes : 0 , rows : 0 } ; private _map : IMapItem [ ] = [ ] ; private _stopped : boolean = false ; private _found : number = 0 ; constructor ( options , streamId , offsets ? ) { super ( options ) ; this . _streamId = streamId ; this . _offsets = offsets === undefined ? { bytes : 0 , rows : 0 } : offsets ; this . _logger = new Logger ( `Transform.Map: ${ this . _streamId }` ) ; } public setBeforeCallbackHandle ( handle ) { this . _beforeCallbackHandle = handle ; } public _transform ( chunk , encoding , callback ) { let output = '' ; if ( typeof chunk === 'string' ) { output = this . _rest + chunk ; } else { output = this . _rest + chunk . toString ( 'utf8' ) ; } const rest = this . _getRest ( output ) ; this . _rest = rest . rest ; output = rest . cleared ; const rowsInChunk = output . split ( /[\n\r]/gi ) . length - 1 ; this . _found += rowsInChunk ; const rows = { from : this . _offsets . rows , to : this . _offsets . rows + rowsInChunk , } ; const bytes = { from : this . _offsets . bytes , to : this . _offsets . bytes , } ; const size = Buffer . byteLength ( output , 'utf8' ) ; rows . to -= 1 ; bytes . to += size - 1 ; const results = { output : output , bytesSize : size , map : { rows : rows , bytes : bytes } , } ; this . _offsets . rows = results . map . rows . to + 1 ; this . _offsets . bytes = results . map . bytes . to + 1 ; this . _map . push ( results . map ) ; if ( this . _stopped ) { return results ; } if ( callback !== undefined ) { if ( typeof this . _beforeCallbackHandle === 'function' ) { this . _beforeCallbackHandle ( results ) . then ( ( ) => { callback ( undefined , output ) ; } ) . catch ( ( error ) => { this . _logger . warn ( `Error from "beforeCallbackHandle": ${ error . message }` ) ; callback ( undefined , output ) ; } ) ; } else { callback ( undefined , output ) ; } } else if ( typeof this . _beforeCallbackHandle === 'function' ) { this . _beforeCallbackHandle ( results ) . catch ( ( error ) => { this . _logger . warn ( `Error from "beforeCallbackHandle": ${ error . message }` ) ; } ) ; } this . emit ( Transform . Events . found , { total : this . _found , chunk : rowsInChunk , map : results . map , } ) ; return results ; } public convert ( chunk ) { return this . _transform ( chunk , 'utf8' , undefined ) ; } public getMap ( ) { return this . _map ; } public getOffsets ( ) { return Object . assign ( { } , this . _offsets ) ; } public lock ( ) { this . _stopped = true ; } private _getRest ( str ) { const last = str . length - 1 ; for ( let i = last ; i >= 0 ; i -= 1 ) { if ( str [ i ] === '\n' && i > 0 ) { return { rest : str . substr ( i + 1 , last ) , cleared : str . substr ( 0 , i + 1 ) , } ; } } return { rest : '' , cleared : str } ; } }