import { Worker } from "cluster" ; import { IProcessRequest , IProcessResponse } from ".." ; export interface IWorkerMessage { event : string ; processId : string ; channel : string ; data ? : any ; } export type WorkerListner = ( processId : string , channel : string , data : any ) => void ; export type IMasterIPCMiddleware = ( ipcm : IPCMaster , message : IWorkerMessage , next : ( ) => void ) => void ; /**
 * IPC interface for master node
 */ export class IPCMaster { public workers : { [ pid ] : Worker ; } = { } ; public subscribers : { [ channel ] : Set < string > ; } = { } ; public listners : { [ event ] : WorkerListner ; } = { } ; public middlewares : IMasterIPCMiddleware [ ] = [ ] ; public add ( pid , worker ) { this . workers [ pid ] = worker ; worker . on ( "message" , ( message ) => this . nextMiddleware ( 0 , message ) ) ; } public use ( middleware ) { this . middlewares . push ( middleware ) ; } public nextMiddleware ( index , message ) { if ( index < this . middlewares . length ) { this . middlewares [ index ] ( this , message , ( ) => this . nextMiddleware ( index + 1 , message ) ) ; } else { this . onMessage ( message ) ; } } protected onMessage ( message ) { const { event , processId , channel , data } = message ; switch ( event ) { case "subscribe" : return this . onSubscribe ( processId , channel ) ; case "unsubscribe" : return this . onUnsubscribe ( processId , channel ) ; case "publish" : return this . onPublish ( processId , channel , data ) ; } } public onSubscribe ( processId , channel ) { if ( ! this . subscribers [ channel ] ) { this . subscribers [ channel ] = new Set ( ) ; } this . subscribers [ channel ] . add ( processId ) ; } public onUnsubscribe ( processId , channel ) { if ( ! this . subscribers [ channel ] ) { return ; } this . subscribers [ channel ] . delete ( processId ) ; if ( ! this . subscribers [ channel ] . size ) { delete this . subscribers [ channel ] ; } } public onPublish ( processId , channel , data ) { if ( ! this . subscribers [ channel ] ) { return ; } for ( const pid of this . subscribers [ channel ] ) { this . workers [ pid ] . send ( { channel , data } ) ; } } }