import { spawn , ChildProcess } from 'child_process' ; import { ReadStream } from 'fs' ; import { CancelablePromise } from "indexer-neon" ; import { EventEmitter } from 'events' ; import { Readable } from 'stream' ; import { IRangeDefinition } from './transform.results' ; import { CommonInterfaces } from '../../../interfaces/interface.common' ; import Logger from '../../../tools/env.logger' ; import Transform from './transform.results' ; import ServicePaths from '../../../services/service.paths' ; import NullWritableStream from '../../../classes/stream.writable.null' ; import * as path from 'path' ; import * as fs from 'fs' ; export { IRangeDefinition } ; export interface IRangeEvent { guid : string ; range : CommonInterfaces . TimeRanges . IRange ; } export class OperationSearch extends EventEmitter { public static Events = { range : 'range' , } ; private _logger : Logger ; private _streamFile : string ; private _streamGuid : string ; private _tasks : Map < string , CancelablePromise < CommonInterfaces . TimeRanges . IRange [ ] , void > > = new Map ( ) ; constructor ( streamGuid , streamFile ) { super ( ) ; this . _streamGuid = streamGuid ; this . _streamFile = streamFile ; this . _logger = new Logger ( `Time ranges: search (${ streamGuid })` ) ; } public destroy ( ) { this . removeAllListeners ( ) ; return this . drop ( ) ; } public perform ( guid , format , definition ) { let cleaner ; const task = new CancelablePromise < CommonInterfaces . TimeRanges . IRange [ ] , void > ( ( resolve , reject , cancel , ref , self ) => { let canceled = false ; self . canceled ( ( ) => { canceled = true ; } ) ; fs . stat ( this . _streamFile , ( err , stats ) => { if ( canceled ) { return ; } if ( err ) { return reject ( err ) ; } const measurer = this . _logger . measure ( `search #${ guid }` ) ; const transform = new Transform ( { } , this . _streamGuid , format , definition ) ; transform . on ( Transform . Events . range , ( range ) => { this . emit ( OperationSearch . Events . range , { guid : guid , range : range , } ) ; } ) ; const writer = new NullWritableStream ( ) ; const process = spawn ( ServicePaths . getRG ( ) , this . _getProcArgs ( definition . points , this . _streamFile ) , { cwd : path . dirname ( this . _streamFile ) , stdio : [ 'pipe' , 'pipe' , 'pipe' ] , } ) ; if ( process . stdin === null || process . stdout === null ) { return reject ( new Error ( `No stdin / stdout` ) ) ; } [ process , process . stdin , process . stdout , writer ] . forEach ( ( smth ) => { smth . once ( 'error' , ( error ) => { if ( cleaner === undefined ) { return ; } this . _logger . error ( `Error during search: ${ error . message }` ) ; reject ( error ) ; } ) ; } ) ; process . once ( 'close' , ( ) => { transform . result ( ( ranges ) => { resolve ( ranges ) ; } ) ; } ) ; process . stdout . pipe ( transform ) . pipe ( writer ) ; cleaner = ( ) => { process . removeAllListeners ( ) ; process . stdin ?. removeAllListeners ( ) ; process . stdin ?. end ( ) ; process . stdin ?. destroy ( ) ; process . stdout ?. removeAllListeners ( ) ; process . stdout ?. unpipe ( ) ; process . stdout ?. destroy ( ) ; process . kill ( ) ; writer . removeAllListeners ( ) ; transform . lock ( ) ; transform . removeAllListeners ( ) ; measurer ( ) ; this . _tasks . delete ( guid ) ; } ; } ) ; } ) . finally ( ( ) => { if ( cleaner === undefined ) { this . _tasks . delete ( guid ) ; return ; } cleaner ( ) ; } ) ; this . _tasks . set ( guid , task ) ; return task ; } public drop ( guid ? ) { return new Promise ( ( resolve ) => { if ( guid !== undefined ) { const task = this . _tasks . get ( guid ) ; if ( task === undefined ) { resolve ( ) ; } else { task . abort ( ) . finally ( ( ) => { resolve ( ) ; } ) ; } return ; } if ( this . _tasks . size > 0 ) { this . _tasks . forEach ( ( task ) => { task . abort ( ) . finally ( ( ) => { if ( this . _tasks . size === 0 ) { resolve ( ) ; } } ) ; } ) ; } else { resolve ( ) ; } } ) ; } 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 _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 _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 ; } }