import * as Net from 'net' ; import * as IPCPluginMessages from '../../../../common/ipc/plugins.ipc.messages/index' ; import * as Tools from '../../tools/index' ; import { ChildProcess , fork } from 'child_process' ; import { Emitter , Subscription } from '../../tools/index' ; import { CStdoutSocketAliases } from '../../consts/controller.plugin.process' ; import Logger from '../../tools/env.logger' ; import ControllerIPCPlugin from './plugin.process.ipc' ; import ServiceProduction from '../../services/service.production' ; import ServicePaths from '../../services/service.paths' ; export const CFirstDebugPort = 9240 ; export const CDebugPortSeqName = 'plugin_debug_ports' ; const CSettings = { resolverTimeout : 3000 , } ; interface IConnection { socket : Net . Socket ; file : string ; } interface IOpt { id : number ; name : string ; entrypoint : string ; token : string ; } const CPluginStartTimeout = 10000 ; class description export default class ControllerPluginProcessSingle extends Emitter { public static Events = { close : Symbol ( ) , disconnect : Symbol ( ) , error : Symbol ( ) , output : Symbol ( ) , } ; private _logger : Logger ; private _opt : IOpt ; private _process : ChildProcess | undefined ; private _ipc : ControllerIPCPlugin | undefined ; private _connections : Map < string , IConnection > = new Map ( ) ; private _subscriptions : { [ key ] : Subscription ; } = { } ; private _resolvers : { bind : Map < string , ( ) => void > ; unbind : Map < string , ( ) => void > ; } = { bind : new Map ( ) , unbind : new Map ( ) , } ; private _stderr : string = '' ; private _rejector ? : ( error : Error ) => void ; private _resolver ? : ( ) => void ; constructor ( opt ) { super ( ) ; this . _opt = opt ; this . _logger = new Logger ( `Plugin: ${ this . _opt . name }` ) ; this . _onSTDErr = this . _onSTDErr . bind ( this ) ; this . _onSTDOut = this . _onSTDOut . bind ( this ) ; this . _onClose = this . _onClose . bind ( this ) ; this . _onError = this . _onError . bind ( this ) ; this . _onDisconnect = this . _onDisconnect . bind ( this ) ; } returns { Promise < void > } public attach ( ) { return new Promise ( ( resolve , reject ) => { this . _rejector = reject ; this . _resolver = resolve ; const args = [ `--chipmunk-settingspath=${ ServicePaths . getPluginsCfgFolder ( ) }` , `--chipmunk-plugin-alias=${ this . _opt . name . replace ( /[^\d\w-_]/gi , '_' ) }` , ] ; if ( ! ServiceProduction . isProduction ( ) ) { args . push ( `--inspect=127.0.0.1:${ Tools . getSequence ( CDebugPortSeqName , CFirstDebugPort ) }` ) ; } this . _process = fork ( this . _opt . entrypoint , args , { stdio : [ 'pipe' , 'pipe' , 'pipe' , 'ipc' , ] } ) ; this . _process . stderr ?. on ( 'data' , this . _onSTDErr ) ; this . _process . stdout ?. on ( 'data' , this . _onSTDOut ) ; this . _process . on ( 'exit' , this . _onClose ) ; this . _process . on ( 'close' , this . _onClose ) ; this . _process . on ( 'error' , this . _onError ) ; this . _process . on ( 'disconnect' , this . _onDisconnect ) ; const ipc = new ControllerIPCPlugin ( this . _opt . name , this . _process , this . _opt . token ) ; let subscription ; ipc . subscribe ( IPCPluginMessages . PluginState , ( state ) => { if ( subscription !== undefined ) { subscription . destroy ( ) ; } if ( state . state !== IPCPluginMessages . EPluginState . ready ) { return this . _reject ( new Error ( this . _logger . error ( `Fail to attach plugin, because plugin state is: ${ state . state }` ) ) ) ; } ipc . send ( new IPCPluginMessages . PluginToken ( { token : this . _opt . token , id : this . _opt . id , } ) ) . then ( ( ) => { Promise . all ( [ ipc . subscribe ( IPCPluginMessages . SessionStreamBound , this . _onSessionStreamBound . bind ( this ) ) . then ( ( _subscription ) => { this . _subscriptions . SessionStreamBound = _subscription ; } ) , ipc . subscribe ( IPCPluginMessages . SessionStreamUnbound , this . _onSessionStreamUnbound . bind ( this ) ) . then ( ( _subscription ) => { this . _subscriptions . SessionStreamUnbound = _subscription ; } ) , ] ) . then ( ( ) => { this . _ipc = ipc ; this . _resolve ( ) ; } ) . catch ( ( error ) => { this . _reject ( new Error ( this . _logger . warn ( `Fail to subscribe to plugin's IPC events due error: ${ error . message }` ) ) ) ; } ) ; } ) . catch ( ( sendingError ) => { this . _reject ( new Error ( this . _logger . error ( `Fail delivery plugin token due error: ${ sendingError . message }` ) ) ) ; } ) ; } ) . then ( ( _subscription ) => { subscription = _subscription ; setTimeout ( ( ) => { if ( this . _rejector !== undefined ) { this . _reject ( new Error ( this . _logger . error ( `Fail to start plugin because timeout.` ) ) ) ; } } , CPluginStartTimeout ) ; } ) . catch ( ( error ) => { this . _reject ( new Error ( this . _logger . warn ( `Fail to subscribe to plugin's state event due error: ${ error . message }` ) ) ) ; } ) ; } ) ; } returns { Promise < void > } public getIPC ( ) { if ( this . _ipc === undefined ) { return new Error ( `IPC controller isn't initialized.` ) ; } return this . _ipc ; } returns public kill ( signal = 'SIGTERM' ) { Object . keys ( this . _subscriptions ) . forEach ( ( key ) => { this . _subscriptions [ key ] . destroy ( ) ; } ) ; if ( this . _ipc !== undefined ) { this . _ipc . destroy ( ) ; } if ( this . _process !== undefined ) { if ( ! this . _process . killed ) { this . _process . kill ( signal ) ; } this . _process = undefined ; this . _connections . clear ( ) ; } return true ; } public bindStream ( guid , connection ) { return new Promise ( ( resolve , reject ) => { if ( this . _process === undefined ) { return reject ( new Error ( this . _logger . warn ( `Attempt to bind stream to plugin, which doesn't attached. Stream GUID: ${ guid }.` ) ) ) ; } if ( this . _connections . has ( guid ) ) { return reject ( new Error ( this . _logger . warn ( `Plugin process is already bound with stream "${ guid }"` ) ) ) ; } this . _logger . debug ( `Sent information about stream GUID: ${ guid }.` ) ; this . _bindRefWithId ( connection . socket ) . then ( ( ) => { if ( this . _process === undefined ) { return reject ( new Error ( this . _logger . warn ( `Fail to bind stream to plugin, which doesn't attached. Stream GUID: ${ guid }.` ) ) ) ; } this . _connections . set ( guid , connection ) ; if ( process . platform === 'win32' ) { this . _process . send ( `${ CStdoutSocketAliases . bind }${ guid };${ connection . file }` ) ; } else { this . _process . send ( `${ CStdoutSocketAliases . bind }${ guid };${ connection . file }` , connection . socket ) ; } this . _setBindResolver ( guid , resolve ) ; } ) . catch ( reject ) ; } ) ; } public unbindStream ( guid ) { return new Promise ( ( resolve ) => { if ( this . _process === undefined ) { this . _logger . warn ( `Attempt to unbind stream to plugin, which doesn't attached. Stream GUID: ${ guid }.` ) ; return resolve ( ) ; } if ( ! this . _connections . has ( guid ) ) { this . _logger . warn ( `Plugin process is already unbound from stream "${ guid }" or it wasn't bound at all, because plugin doesn't listen "openStream" event` ) ; return resolve ( ) ; } this . _connections . delete ( guid ) ; this . _process . send ( `${ CStdoutSocketAliases . unbind }${ guid }` ) ; this . _setUnbindResolver ( guid , resolve ) ; } ) ; } public isAttached ( ) { return this . _process !== undefined ; } private _reject ( error ) { if ( this . _rejector === undefined || this . _resolver === undefined ) { return ; } this . kill ( ) ; this . _rejector ( error ) ; this . _rejector = undefined ; this . _resolver = undefined ; } private _resolve ( ) { if ( this . _rejector === undefined || this . _resolver === undefined ) { return ; } this . _resolver ( ) ; this . _rejector = undefined ; this . _resolver = undefined ; } private _bindRefWithId ( socket ) { return new Promise ( ( resolve , reject ) => { socket . write ( `[plugin:${ this . _opt . id }]` , ( error ) => { if ( error ) { return reject ( new Error ( this . _logger . error ( `Cannot send binding message into socket due error: ${ error . message }` ) ) ) ; } resolve ( ) ; } ) ; } ) ; } private _setBindResolver ( guid , resolver ) { this . _resolvers . bind . set ( guid , resolver ) ; setTimeout ( this . _applyBindResolver . bind ( this , guid ) , CSettings . resolverTimeout ) ; } private _setUnbindResolver ( guid , resolver ) { this . _resolvers . unbind . set ( guid , resolver ) ; setTimeout ( this . _applyUnbindResolver . bind ( this , guid ) , CSettings . resolverTimeout ) ; } private _applyBindResolver ( guid ) { const resolver = this . _resolvers . bind . get ( guid ) ; if ( resolver === undefined ) { return ; } resolver ( ) ; this . _resolvers . bind . delete ( guid ) ; } private _applyUnbindResolver ( guid ) { const resolver = this . _resolvers . unbind . get ( guid ) ; if ( resolver === undefined ) { return ; } resolver ( ) ; this . _resolvers . bind . delete ( guid ) ; } returns private _onSTDOut ( chunk ) { const str = chunk . toString ( ) ; this . _logger . save ( this . _opt . token , str ) ; this . emit ( ControllerPluginProcessSingle . Events . output ) ; } returns private _onSTDErr ( chunk ) { const str = chunk . toString ( ) ; this . _logger . save ( this . _opt . token , str ) ; this . _stderr += str ; } returns private _onClose ( ... args ) { this . kill ( ) ; this . emit ( ControllerPluginProcessSingle . Events . close , ... args ) ; } returns private _onError ( ... args ) { this . kill ( ) ; this . emit ( ControllerPluginProcessSingle . Events . error , ... args ) ; } returns private _onDisconnect ( ... args ) { this . kill ( ) ; this . emit ( ControllerPluginProcessSingle . Events . disconnect , ... args ) ; this . _reject ( new Error ( this . _stderr ) ) ; } private _onSessionStreamBound ( message ) { if ( message . error ) { this . _logger . warn ( `Fail to bind stream due error: ${ message . error }` ) ; } this . _applyBindResolver ( message . streamId ) ; } private _onSessionStreamUnbound ( message ) { if ( message . error ) { this . _logger . warn ( `Fail to unbind stream due error: ${ message . error }` ) ; } this . _applyUnbindResolver ( message . streamId ) ; } }