/**
 * Abstract class for process discovery and communication
 */ export type IPCRequestHandler < T = any > = ( data : any ) => Promise < T > ; export type ProcessListener < T > = ( data : T ) => void ; export interface IProcessRequest { requestId : string ; method : string ; data : any ; } export interface IProcessResponse { error ? : any ; data ? : any ; } export interface IProcessState { [ key ] : any ; } export interface IPCParams < T extends IProcessState > { timeout ? : number ; state ? : T ; processId : string ; instanceName ? : string ; healthCheckInterval ? : number ; } export interface IProcessInstance < T extends IProcessState > { ttl : number ; state : T ; } export interface IProcessInfo < T extends IProcessState > { id : string ; state : T ; } export abstract class IPCManager < S extends IProcessState = any > { public processId : string ; public timeout : number ; public healthCheckInterval : number ; public requestHandlers : { [ method ] : IPCRequestHandler ; } ; public state : S ; public instances : Map < string , IProcessInstance < S > > ; private healthCheckTimeout ? : NodeJS . Timeout ; private instanceName : string ; constructor ( params ) { this . processId = params . processId ; this . timeout = params . timeout || 1000 ; this . instanceName = params . instanceName = "instance" ; this . requestHandlers = { } ; this . instances = new Map < string , IProcessInstance < S > > ( ) ; this . healthCheckInterval = params ?. healthCheckInterval || 30000 ; this . state = params ?. state || { } as S ; } public init ( ) { this . subscribe ( `${ this . instanceName }:add` , ( id ) => { this . addInstance ( id ) ; console . log ( `Process ${ this . processId }: added instance ${ id } [${ this . instances . size }]` ) ; this . publish ( `${ id }:discovery` , { id : this . processId , state : this . state } ) ; } ) ; this . subscribe ( `${ this . instanceName }:delete` , ( id ) => { this . deleteInstance ( id ) ; } ) ; this . subscribe ( `${ this . processId }:discovery` , ( { id , state } ) => { if ( id === this . processId ) { return ; } this . addInstance ( id ) ; console . log ( `Process ${ this . processId }}: added instance ${ id } [${ this . instances . size }]` ) ; this . updateInstanceState ( id , state ) ; } ) ; this . subscribe ( `ipc-request:${ this . processId }` , ( request ) => { const { requestId , method , data } = request ; const listener = this . requestHandlers [ method ] ; if ( ! listener ) { return ; } listener ( data ) . then ( ( result ) => { this . publish ( `ipc-response:${ requestId }` , { data : result } ) ; } ) . catch ( ( error ) => { this . publish ( `ipc-response:${ requestId }` , { error } ) ; } ) ; } ) ; this . instances . set ( this . processId , { state : this . state , ttl : - 1 } ) ; this . publish ( `${ this . instanceName }:add` , this . processId ) ; this . startHealthCheck ( ) ; } public startHealthCheck ( ) { if ( this . healthCheckTimeout ) { this . stopHealthCheck ( ) ; } this . healthCheckTimeout = setInterval ( ( ) => { this . publishState ( ) ; for ( const [ id , { ttl } ] of this . instances . entries ( ) ) { if ( ttl > 0 && ttl < Date . now ( ) ) { this . deleteInstance ( id ) ; } } } , this . healthCheckInterval ) ; } public terminate ( ) { this . publish ( `${ this . instanceName }:delete` , this . processId ) ; close ( ) ; } public stopHealthCheck ( ) { if ( ! this . healthCheckTimeout ) { return ; } clearInterval ( this . healthCheckTimeout ) ; this . healthCheckTimeout = undefined ; } public publishState ( ) { this . publish ( `${ this . processId }:info` , this . state ) ; } private addInstance ( id ) { this . subscribe ( `${ id }:info` , ( state ) => { if ( id === this . processId ) { return ; } this . updateInstanceState ( id , state ) ; } ) ; } private updateInstanceState ( id , state ) { const ttl = Date . now ( ) + this . healthCheckInterval * 2 ; this . instances . set ( id , { state , ttl } ) ; if ( this . instances . size > 1 && ! this . healthCheckTimeout ) { this . startHealthCheck ( ) ; } } private deleteInstance ( id ) { this . unsubscribe ( `${ id }:info` ) ; this . instances . delete ( id ) ; if ( this . instances . size < 2 ) { this . stopHealthCheck ( ) ; } } public async requestProcess < L , R > ( id , method , data ) { return new Promise ( ( resolve , reject ) => { const requestId = Date . now ( ) . toString ( 36 ) + Math . random ( ) . toString ( 36 ) . slice ( 3 ) ; const timer = setTimeout ( ( ) => { this . unsubscribe ( `ipc-response:${ requestId }` ) ; console . debug ( "REJECT: ipcm requestProcess timeout error" ) ; reject ( `IPC request timeout. Process: ${ this . processId }, method: ${ method }` ) ; } , this . timeout ) ; this . subscribe ( `ipc-response:${ requestId }` , ( response ) => { clearTimeout ( timer ) ; this . unsubscribe ( `ipc-response:${ requestId }` ) ; if ( response . error ) { console . log ( "REJECT: ipcm requestProcess error" ) ; reject ( response . error ) ; } else { resolve ( response . data ) ; } } ) ; this . publish ( `ipc-request:${ id }` , { requestId , method , data } ) ; } ) ; } public onRequest ( method , handler ) { this . requestHandlers [ method ] = handler ; } public pids ( ) { return [ ... this . instances . keys ( ) ] ; } public abstract subscribe < T > ( channel , listener ) ; public abstract unsubscribe ( channel ) ; public abstract publish < T > ( channel , data ) ; public abstract close ( ) ; }