Files
makemd/server/src/core/runtime/EventBusOptimizationService.ts

76 lines
2.3 KiB
TypeScript
Raw Normal View History

import { RedisService } from '../../utils/RedisService';
import { logger } from '../../utils/logger';
export type EventCallback = (payload: any) => Promise<void>;
/**
* [BIZ_KER_115] Domain (EventBusOptimizationService)
* @description /
* Redis Pub/Sub
* Order, Product, Logistics
*/
export class EventBusOptimizationService {
private static handlers: Map<string, EventCallback[]> = new Map();
private static readonly REDIS_EVENT_CHANNEL = 'core:event_bus:broadcast';
/**
*
*/
static async init() {
const redis = RedisService.getClient();
// 订阅 Redis 广播通道
await RedisService.subscribeToStateSync('EVENT_BUS', async (msg) => {
if (msg.type === 'BROADCAST_EVENT') {
await this.emitLocal(msg.event, msg.payload);
}
});
logger.info(`[EventBus] Distributed event bus initialized`);
}
/**
*
*/
static subscribe(event: string, callback: EventCallback) {
const callbacks = this.handlers.get(event) || [];
callbacks.push(callback);
this.handlers.set(event, callbacks);
logger.info(`[EventBus] Subscribed to event: ${event}`);
}
/**
* ()
*/
static async emitLocal(event: string, payload: any) {
const callbacks = this.handlers.get(event);
if (callbacks) {
for (const cb of callbacks) {
try {
await cb(payload);
} catch (err: any) {
logger.error(`[EventBus] Handler for ${event} failed: ${err.message}`);
}
}
}
}
/**
* (/)
*/
static async emitGlobal(event: string, payload: any) {
// 1. 先触发本地监听
await this.emitLocal(event, payload);
// 2. 通过 Redis 广播到其它节点
try {
await RedisService.broadcastStateChange('EVENT_BUS', {
type: 'BROADCAST_EVENT',
event,
payload,
timestamp: Date.now()
});
} catch (err: any) {
logger.error(`[EventBus] Global emit failed: ${err.message}`);
}
}
}