import { spawn } from 'child_process' ; import { Observable } from 'rxjs' ; import { Disposable } from 'vscode' ; import { StdErrError } from '../errors/stdErr' ; import { createDeferred } from '../helpers' ; import { DEFAULT_ENCODING } from './constants' ; import { ExecutionResult , IBufferDecoder , IProcessService , ObservableExecutionResult , Output , SpawnOptions } from './types' ; export class ProcessService implements IProcessService { constructor ( private readonly decoder , private readonly env ? ) { } public execObservable ( file , args , options = { } ) { const encoding = options . encoding = typeof options . encoding === 'string' && options . encoding . length > 0 ? options . encoding : DEFAULT_ENCODING ; delete options . encoding ; const spawnOptions = { ... options } ; if ( ! spawnOptions . env || Object . keys ( spawnOptions ) . length === 0 ) { const env = this . env ? this . env : process . env ; spawnOptions . env = { ... env } ; } spawnOptions . env . PYTHONUNBUFFERED = '1' ; if ( ! spawnOptions . env . PYTHONIOENCODING ) { spawnOptions . env . PYTHONIOENCODING = 'utf-8' ; } const proc = spawn ( file , args , spawnOptions ) ; let procExited = false ; const output = new Observable < Output > ( subscriber => { const disposables = [ ] ; const on = ( ee , name , fn ) => { ee . on ( name , fn as any ) ; disposables . push ( { dispose : ( ) => ee . removeListener ( name , fn as any ) } ) ; } ; if ( options . token ) { disposables . push ( options . token . onCancellationRequested ( ( ) => { if ( ! procExited && ! proc . killed ) { proc . kill ( ) ; procExited = true ; } } ) ) ; } const sendOutput = ( source , data ) => { const out = this . decoder . decode ( [ data ] , encoding ) ; if ( source === 'stderr' && options . throwOnStdErr ) { subscriber . error ( new StdErrError ( out ) ) ; } else { subscriber . next ( { source , out : out } ) ; } } ; on ( proc . stdout , 'data' , ( data ) => sendOutput ( 'stdout' , data ) ) ; on ( proc . stderr , 'data' , ( data ) => sendOutput ( 'stderr' , data ) ) ; proc . once ( 'close' , ( ) => { procExited = true ; subscriber . complete ( ) ; disposables . forEach ( disposable => disposable . dispose ( ) ) ; } ) ; proc . once ( 'error' , ex => { procExited = true ; subscriber . error ( ex ) ; disposables . forEach ( disposable => disposable . dispose ( ) ) ; } ) ; } ) ; return { proc , out : output } ; } public exec ( file , args , options = { } ) { const encoding = options . encoding = typeof options . encoding === 'string' && options . encoding . length > 0 ? options . encoding : DEFAULT_ENCODING ; delete options . encoding ; const spawnOptions = { ... options } ; if ( ! spawnOptions . env || Object . keys ( spawnOptions ) . length === 0 ) { const env = this . env ? this . env : process . env ; spawnOptions . env = { ... env } ; } spawnOptions . env . PYTHONUNBUFFERED = '1' ; if ( ! spawnOptions . env . PYTHONIOENCODING ) { spawnOptions . env . PYTHONIOENCODING = 'utf-8' ; } const proc = spawn ( file , args , spawnOptions ) ; const deferred = createDeferred < ExecutionResult > ( ) ; const disposables = [ ] ; const on = ( ee , name , fn ) => { ee . on ( name , fn as any ) ; disposables . push ( { dispose : ( ) => ee . removeListener ( name , fn as any ) } ) ; } ; if ( options . token ) { disposables . push ( options . token . onCancellationRequested ( ( ) => { if ( ! proc . killed && ! deferred . completed ) { proc . kill ( ) ; } } ) ) ; } const stdoutBuffers = [ ] ; on ( proc . stdout , 'data' , ( data ) => stdoutBuffers . push ( data ) ) ; const stderrBuffers = [ ] ; on ( proc . stderr , 'data' , ( data ) => { if ( options . mergeStdOutErr ) { stdoutBuffers . push ( data ) ; stderrBuffers . push ( data ) ; } else { stderrBuffers . push ( data ) ; } } ) ; proc . once ( 'close' , ( ) => { if ( deferred . completed ) { return ; } const stderr = stderrBuffers . length === 0 ? undefined : this . decoder . decode ( stderrBuffers , encoding ) ; if ( stderr && stderr . length > 0 && options . throwOnStdErr ) { deferred . reject ( new StdErrError ( stderr ) ) ; } else { const stdout = this . decoder . decode ( stdoutBuffers , encoding ) ; deferred . resolve ( { stdout , stderr } ) ; } disposables . forEach ( disposable => disposable . dispose ( ) ) ; } ) ; proc . once ( 'error' , ex => { deferred . reject ( ex ) ; disposables . forEach ( disposable => disposable . dispose ( ) ) ; } ) ; return deferred . promise ; } }