import { spawn , ChildProcess } from 'child_process' ; import * as path from 'path' ; import * as fs from 'fs' ; import { WriteStream , ReadStream } from 'fs' ; import Logger from '../../../tools/env.logger' ; import { CancelablePromise } from '../../../tools/promise.cancelable' ; import ServicePaths from '../../../services/service.paths' ; import Transform , { IMapItem , IMapChunkEvent , IOffset } from './transform.map' ; import { EventEmitter } from 'events' ; import { Readable , Writable } from 'stream' ; export { IMapChunkEvent } ; type THandler = ( ) => void ; export class OperationAppend extends EventEmitter { public static Events = { onMapUpdated : 'onMapUpdated' , } ; private _logger : Logger ; private _streamFile : string ; private _searchFile : string ; private _streamGuid : string ; private _cleaner : THandler | undefined ; private _readFrom : number = 0 ; private _offset : IOffset = { bytes : 0 , rows : 0 } ; private _task : CancelablePromise < IMapItem [ ] , void > | undefined ; constructor ( streamGuid , streamFile , searchFile ) { super ( ) ; this . _streamGuid = streamGuid ; this . _streamFile = streamFile ; this . _searchFile = searchFile ; this . _logger = new Logger ( `Search operation: append (${ streamGuid })` ) ; this . _clear = this . _clear . bind ( this ) ; } public destroy ( ) { this . removeAllListeners ( ) ; } public perform ( regExp , readTo , guid ) { if ( this . _readFrom >= readTo || isNaN ( this . _readFrom ) || ! isFinite ( this . _readFrom ) || isNaN ( readTo ) || ! isFinite ( readTo ) ) { return new Error ( `(append) Cannot perform search because a range isn't correct: from = ${ this . _readFrom }; to = ${ readTo }` ) ; } if ( this . _cleaner !== undefined ) { this . _logger . warn ( `Attempt to start search, while previous isn't finished` ) ; return new Error ( `(append) Fail to start search, because previous process isn't finished.` ) ; } this . _task = new CancelablePromise < IMapItem [ ] , void > ( ( resolve , reject , cancel , self ) => { self . cancel ( ( ) => { this . _clear ( ) ; } ) ; const measurer = this . _logger . measure ( `appending search #${ guid }` ) ; const transform = new Transform ( { } , this . _streamGuid , this . _offset ) ; transform . on ( Transform . Events . found , ( event ) => { this . emit ( OperationAppend . Events . onMapUpdated , event ) ; } ) ; const reader = fs . createReadStream ( this . _streamFile , { encoding : 'utf8' , start : this . _readFrom , end : readTo } ) ; const writer = fs . createWriteStream ( this . _searchFile , { flags : 'a' } ) ; const process = spawn ( ServicePaths . getRG ( ) , this . _getProcArgs ( regExp , '-' ) , { cwd : path . dirname ( this . _streamFile ) , stdio : [ 'pipe' , 'pipe' , 'pipe' ] , } ) ; if ( process . stdin === null || process . stdout === null ) { return reject ( new Error ( `No stdin / stdout` ) ) ; } reader . pipe ( process . stdin ) ; process . stdout . pipe ( transform ) . pipe ( writer ) ; [ process , process . stdin , process . stdout , writer , reader , transform ] . forEach ( ( smth , i ) => { smth . once ( 'error' , ( error ) => { if ( this . _cleaner === undefined ) { return ; } this . _logger . error ( `Error during append (task: ${ guid }): ${ error . message }` ) ; reject ( error ) ; } ) ; } ) ; process . once ( 'close' , ( ) => { if ( ! writer . writableFinished ) { writer . once ( 'finish' , ( ) => { this . setReadFrom ( readTo + 1 ) ; resolve ( transform . getMap ( ) ) ; } ) ; } else { this . setReadFrom ( readTo + 1 ) ; resolve ( transform . getMap ( ) ) ; } } ) ; this . _cleaner = ( ) => { process . removeAllListeners ( ) ; process . stdin ?. removeAllListeners ( ) ; process . stdin ?. end ( ) ; process . stdin ?. destroy ( ) ; process . stdout ?. removeAllListeners ( ) ; process . stdout ?. unpipe ( ) ; process . stdout ?. destroy ( ) ; process . kill ( ) ; reader . removeAllListeners ( ) ; reader . close ( ) ; reader . destroy ( ) ; writer . removeAllListeners ( ) ; writer . close ( ) ; transform . lock ( ) ; transform . removeAllListeners ( ) ; measurer ( ) ; this . _task = undefined ; this . _cleaner = undefined ; this . _logger . debug ( `RG process is finished/killed (task ID: ${ guid })` ) ; } ; } ) . finally ( this . _clear . bind ( this ) ) ; return this . _task ; } public setOffset ( offset ) { this . _offset = offset ; } public setReadFrom ( read ) { this . _readFrom = read ; } public drop ( ) { return new Promise ( ( resolve ) => { this . _readFrom = 0 ; this . _offset = { bytes : 0 , rows : 0 } ; if ( this . _task !== undefined ) { this . _logger . warn ( `Dropping search controller, while search operation is still in progress. Current task will be dropped` ) ; this . _task . after ( ( ) => { resolve ( ) ; } ) . break ( ) ; } else { resolve ( ) ; } } ) ; } private _clear ( ) { if ( this . _cleaner !== undefined ) { this . _cleaner ( ) ; } } private _isCaseInsensitive ( regulars ) { if ( regulars instanceof Array ) { return regulars . length > 0 ? ( regulars [ 0 ] . flags . includes ( 'i' ) ? true : false ) : true ; } else { return regulars . flags . includes ( 'i' ) ? true : false ; } } private _constractRegExpStr ( regulars ) { if ( ! ( regulars instanceof Array ) ) { regulars = [ regulars ] ; } const i = this . _isCaseInsensitive ( regulars ) ; const regs = regulars . map ( ( regular ) => { return i ? `((?i)${ regular . source })` : regular . source ; } ) ; return `(${ regs . join ( '|' ) }).*\\x{0003}\\d*\\x{0003}` ; } private _getProcArgs ( regulars , target ) { const expression = this . _constractRegExpStr ( regulars ) ; const args = [ '-N' , '--text' , '--pcre2' , '-e' , expression , target , ] . filter ( x => x !== '' ) ; this . _logger . env ( `Next regular expresition will be used with ripgrep: ${ expression }. Full command: rg ${ args . join ( ' ' ) }` ) ; return args ; } }