import { v4 as uuidv4 } from 'uuid' ; import * as assert from 'assert' ; import * as pm2 from 'pm2' ; import { Handler , Options , RequestPacket , ResponsePacket } from './types' ; export { Handler , Options } from './types' ; const handlers = { } ; const myName = process . env . name ; const myPmId = Number ( process . env . pm_id ) ; let myBus ; const myBusListeners = { } ; process . on ( 'message' , async ( { topic , data : { targetInstanceId , requestId , data } } ) : Promise < void > => { if ( typeof handlers [ topic ] === 'function' && process . send ) { const response = { type : `process:${ targetInstanceId }` , data : { instanceId : myPmId , requestId , message : await handlers [ topic ] ( data ) } , } ; process . send ( response ) ; } } ) ; /**
 * Connects to pm2.
 */ export const connect = function connect ( ) : Promise < void > { return new Promise ( ( approve , reject ) : void => pm2 . connect ( false , ( err ) : void => { if ( err ) return reject ( err ) ; pm2 . launchBus ( ( err , bus ) : void => { if ( err ) return reject ( err ) ; myBus = bus ; myBus . on ( `process:${ myPmId }` , ( { data : { instanceId , requestId , message } } ) : void => { myBusListeners [ requestId ] ?. ( instanceId , message ) ; } ) ; approve ( ) ; } ) ; } ) ) ; } ; /**
 * Determines whether there is a connection with pm2.
 */ export const isConnected = function isConnected ( ) : boolean { return ! ! myBus ; } ; /**
 * Disconnects from pm2.
 */ export const disconnect = function disconnect ( ) : Promise < void > { return new Promise ( ( approve , reject ) : void => pm2 . disconnect ( ( err ) : void => { if ( err ) return reject ( err ) ; myBus = null ; approve ( ) ; } ) ) ; } ; /**
 * Requests messages from processes managed by pm2.
 */ export const getMessages = function getMessages < T = any > ( topic , data ? , { filter = ( process ) : boolean => process . name === myName , includeSelfIfUnmanaged = false , timeout = 1000 , } = { } ) : Promise < T [ ] > { return new Promise < T [ ] > ( ( resolve , reject ) : void => { assert . ok ( isConnected ( ) , 'not connected' ) ; const requestId = uuidv4 ( ) ; const timer = setTimeout ( ( ) : void => reject ( new Error ( `${ topic } timed out` ) ) , timeout ) ; const done = function done ( err , messages ) : void { clearTimeout ( timer ) ; delete myBusListeners [ requestId ] ; if ( err ) reject ( err ) ; else resolve ( messages ) ; } ; new Promise < T [ ] > ( ( resolve , reject ) : void => { pm2 . list ( ( err , processes ) : void => { if ( err ) return reject ( err ) ; const messages = [ ] ; const resolvers = [ ] ; const resolverBusTargets = [ ] ; const resolverSelf = async ( ) : Promise < void > => { messages . push ( await handlers [ topic ] ( data ) ) ; } ; if ( includeSelfIfUnmanaged && Number . isNaN ( myPmId ) ) resolvers . push ( resolverSelf ( ) ) ; for ( const process of processes ) { if ( filter ( process ) ) { if ( process . pm_id === myPmId ) { resolvers . push ( resolverSelf ( ) ) ; } else if ( typeof process . pm_id === 'number' ) { resolverBusTargets . push ( process . pm_id ) ; } } } if ( resolverBusTargets . length ) { resolvers . push ( new Promise ( ( resolve , reject ) : void => { const pendingBusTargets = new Set ( resolverBusTargets ) ; myBusListeners [ requestId ] = ( instanceId , message ) => { if ( pendingBusTargets . delete ( instanceId ) ) { messages . push ( message ) ; if ( ! pendingBusTargets . size ) resolve ( ) ; } } ; const request = { topic , data : { targetInstanceId : myPmId , requestId , data } , } ; resolverBusTargets . forEach ( ( pmId ) : void => { pm2 . sendDataToProcessId ( pmId , request , ( err ) : void => err && reject ( err ) ) ; } ) ; } ) ) ; } Promise . all ( resolvers ) . then ( ( ) : void => resolve ( messages ) ) . catch ( reject ) ; } ) ; } ) . then ( ( messages ) : void => done ( null , messages ) ) . catch ( ( err ) : void => done ( err , [ ] ) ) ; } ) ; } ; /**
 * Attaches a handler for the given topic.
 */ export const onMessage = function onMessage ( topic , handler ) : void { handlers [ topic ] = handler ; } ; export default { connect , isConnected , disconnect , getMessages , onMessage } ;