import { Agent } from 'egg' ; import { getClassName , IPC_EVENT_NAME , IpcEventType } from './Helper' ; import { EventHandlerMap } from './EventHandlerMap' ; import { AgentPluginBase } from '../../basic/AgentPluginBase' ; export class AgentEventManager extends AgentPluginBase < null > { private handlerMap : EventHandlerMap ; constructor ( config , agent ) { super ( config , agent ) ; this . handlerMap = new EventHandlerMap ( ) ; this . agent . messenger . on ( IPC_EVENT_NAME , ( e ) => { this . consume ( e . className , e . type , e . payload , e . from ) ; } ) ; } public async onReady ( ) { } public async onStart ( ) { } public async onClose ( ) { } public subscribe < T > ( constructor , func ) { const className = getClassName ( constructor ) ; this . handlerMap . add ( className , func ) ; } public publish < T > ( type , constructor , param ) { const className = getClassName ( constructor ) ; const p = { type , className , payload : param , from : 'agent' , } ; switch ( type ) { case 'worker' : this . agent . messenger . sendRandom ( IPC_EVENT_NAME , p ) ; break ; case 'workers' : this . agent . messenger . sendToApp ( IPC_EVENT_NAME , p ) ; break ; case 'all' : this . agent . messenger . sendToApp ( IPC_EVENT_NAME , p ) ; this . consume ( className , type , param , 'agent' ) ; this . agent . messenger . sendRandom ( IPC_EVENT_NAME , { ... p , type : 'worker' , } ) ; break ; case 'agent' : this . consume ( className , type , param , 'agent' ) ; break ; default : break ; } } private async consume < T > ( className , type , param , from ) { switch ( type ) { case 'agent' : case 'all' : try { await this . handlerMap . exec ( className , param ) ; } catch ( e ) { this . logger . error ( `Error processing handlers, className=${ className }, e=` , e ) ; } break ; case 'worker' : if ( from === 'worker' ) { this . agent . messenger . sendRandom ( IPC_EVENT_NAME , { className , type , payload : param , from : 'agent' , } ) ; } break ; default : break ; } } } type EventHandler < T > = ( event : T ) => Promise < void > ;