import { spawn , ChildProcess } from 'child_process' ; import * as path from 'path' ; import * as fs from 'fs' ; import { ReadStream } from 'fs' ; import Logger from '../../../tools/env.logger' ; import guid from '../../../tools/tools.guid' ; import { CancelablePromise } from '../../../tools/promise.cancelable' ; import ServicePaths from '../../../services/service.paths' ; import NullWritableStream from '../../../classes/stream.writable.null' ; import Transform , { IMatch , CGroupDelimiter } from './transform.charting' ; import { EventEmitter } from 'events' ; import { Readable , Writable } from 'stream' ; export { IMatch } ; type THandler = ( ) => void ; export class OperationAppend extends EventEmitter { private _logger : Logger ; private _streamFile : string ; private _streamGuid : string ; private _cleaners : Map < string , THandler > = new Map ( ) ; private _last : { from : number ; to : number ; } | undefined ; private _tasks : Map < string , CancelablePromise < IMatch [ ] , void > > = new Map ( ) ; constructor ( streamGuid , streamFile ) { super ( ) ; this . _streamGuid = streamGuid ; this . _streamFile = streamFile ; this . _logger = new Logger ( `Chart operation: append (${ streamGuid })` ) ; } public destroy ( ) { this . removeAllListeners ( ) ; } public perform ( regExp , range , rowOffset , groups = false ) { const taskId = guid ( ) ; const task = new CancelablePromise < IMatch [ ] , void > ( ( resolve , reject ) => { if ( this . _last !== undefined ) { if ( this . _last . to !== range . from - 1 && this . _last . to !== range . to ) { this . _logger . error ( `Last ranage was finished with ${ this . _last . to } bytes, but new range is started from ${ range . from } byte.` ) ; return reject ( new Error ( `Wrong range for append operation` ) ) ; } } this . _last = Object . assign ( range ) ; const groupsCount = this . _getGroupsCount ( regExp . source ) ; groups = ! groups ? false : ( groupsCount > 0 ) ; const measurer = this . _logger . measure ( `Appending (#${ taskId }): bytes: ${ range . from } - ${ range . to };` ) ; const reader = fs . createReadStream ( this . _streamFile , { encoding : 'utf8' , start : range . from , end : range . to } ) ; const writer = new NullWritableStream ( ) ; const transform = new Transform ( { } , rowOffset , groups ) ; const process = spawn ( ServicePaths . getRG ( ) , this . _getProcArgs ( regExp , '-' , groups ? groupsCount : 0 ) , { cwd : path . dirname ( this . _streamFile ) , stdio : [ 'pipe' , 'pipe' , 'pipe' ] , detached : true , } ) ; reader . pipe ( process . stdin ) ; process . stdout . pipe ( transform ) . pipe ( writer ) ; [ process , process . stdin , process . stdout , writer , reader ] . forEach ( ( smth ) => { smth . once ( 'error' , ( error ) => { if ( ! this . _cleaners . has ( taskId ) ) { return ; } this . _logger . error ( `Error during charting: ${ error . message }` ) ; reject ( error ) ; } ) ; } ) ; process . once ( 'close' , ( ) => { resolve ( transform . getMatches ( ) ) ; } ) ; this . _cleaners . set ( taskId , ( ) => { reader . removeAllListeners ( ) ; reader . close ( ) ; transform . stop ( ) ; transform . removeAllListeners ( ) ; process . removeAllListeners ( ) ; process . stdin . removeAllListeners ( ) ; process . stdin . end ( ) ; process . stdin . destroy ( ) ; process . stdout . removeAllListeners ( ) ; process . stdout . unpipe ( ) ; process . stdout . destroy ( ) ; process . kill ( ) ; writer . removeAllListeners ( ) ; this . _cleaners . delete ( taskId ) ; measurer ( ) ; } ) ; } ) . finally ( this . _clear . bind ( this , taskId ) ) ; this . _tasks . set ( taskId , task ) ; return task ; } public dropCursorPosition ( ) { this . _last = undefined ; } public drop ( ) { this . _tasks . forEach ( ( task , id ) => { this . _logger . warn ( `Dropping charting controller, while appending operation is still in progress. Current task "${ id }" will be dropped` ) ; task . break ( ) ; } ) ; this . _tasks . clear ( ) ; this . _cleaners . clear ( ) ; this . _last = undefined ; } private _clear ( id ) { const cleaner = this . _cleaners . get ( id ) ; this . _cleaners . delete ( id ) ; this . _tasks . delete ( id ) ; if ( cleaner !== undefined ) { cleaner ( ) ; } } private _getGroupsCount ( regAsStr ) { const matches = regAsStr . match ( /\(.*?\)/gi ) ; if ( matches === null ) { return 0 ; } return matches . length ; } private _isCaseInsensitive ( reg ) { return reg . flags . includes ( 'i' ) ? true : false ; } private _getProcArgs ( reg , target , groups = 0 ) { const args = [ '-n' , '--text' , '--pcre2' , this . _isCaseInsensitive ( reg ) ? '-i' : '' , '-e' , reg . source , '-o' , ] . filter ( x => x !== '' ) ; if ( groups > 0 ) { args . push ( ... [ '-r' , ( new Array ( groups ) ) . fill ( 0 ) . map ( ( v , i ) => { return `$${ i + 1 }` ; } ) . join ( CGroupDelimiter ) , ] ) ; } args . push ( target ) ; this . _logger . env ( `Next regular expresition will be used with ripgrep: ${ reg . source }. Full command: rg ${ args . join ( ' ' ) }` ) ; return args ; } }