OpenClaw Event Bus

基于 Claw Code 事件优先架构的 OpenClaw 事件总线实现。

特性

• ✅ 事件优先 - 不是日志优先，而是事件原生
• ✅ 发布-订阅模式 - 松耦合的组件通信
• ✅ 类型安全 - TypeScript 严格类型
• ✅ RxJS集成 - 支持响应式编程
• ✅ 背压控制 - 防止内存溢出
• ✅ 指标收集 - 内置性能监控
• ✅ 向后兼容 - 可逐步迁移现有代码

安装

npm install @openclaw/event-bus

快速开始

import { OpenClawEventBus, EventType } from '@openclaw/event-bus';

// 创建事件总线
const eventBus = new OpenClawEventBus({
  persistence: { enabled: false },
  debug: { logAllEvents: true },
});

// 订阅事件
const subscription = eventBus.subscribe(
  EventType.TOOL_COMPLETED,
  (event) => {
    console.log(`✅ Tool ${event.payload.toolName} completed in ${event.payload.duration}ms`);
  }
);

// 发布事件
await eventBus.publish({
  type: EventType.TOOL_COMPLETED,
  timestamp: new Date(),
  source: 'MyService',
  payload: {
    toolId: '123',
    toolName: 'file_read',
    result: { content: 'Hello World' },
    duration: 100,
    callerSessionId: 'session-456',
  },
});

// 取消订阅
eventBus.unsubscribe(subscription);

使用RxJS流

import { debounceTime, filter, map } from 'rxjs/operators';

// 创建事件流
const toolErrors$ = eventBus.stream(EventType.TOOL_FAILED).pipe(
  filter(event => event.payload.duration > 5000),
  map(event => ({
    tool: event.payload.toolName,
    error: event.payload.error.message,
  }))
);

// 订阅流
toolErrors$.subscribe(({ tool, error }) => {
  console.error(`Tool ${tool} failed: ${error}`);
});

事件追踪

import { v4 as uuidv4 } from 'uuid';

const correlationId = uuidv4();

// 创建事件链
await eventBus.publish({
  type: EventType.AGENT_INVOKED,
  timestamp: new Date(),
  source: 'Gateway',
  payload: { /* ... */ },
  metadata: { correlationId },
});

await eventBus.publish({
  type: EventType.TOOL_INVOKED,
  timestamp: new Date(),
  source: 'Agent',
  payload: { /* ... */ },
  metadata: { 
    correlationId,
    causationId: previousEventId,
  },
});

配置选项

const eventBus = new OpenClawEventBus({
  enabled: true,
  
  // 背压控制
  backpressure: {
    enabled: true,
    maxPendingEvents: 1000,
    strategy: 'drop_old', // 'drop_old' | 'block' | 'error'
  },
  
  // 持久化（待实现）
  persistence: {
    enabled: false,
  },
  
  // 指标收集
  metrics: {
    enabled: true,
    interval: 60000,
  },
  
  // 调试
  debug: {
    logAllEvents: false,
    logEventPayload: false,
  },
});

架构

┌─────────────────────────────────────────────────────────────┐
│                    OpenClaw Event Bus                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   Publisher ──►  EventBus  ──►  Subscribers                 │
│                      │                                      │
│                      ▼                                      │
│               RxJS Streams                                  │
│                      │                                      │
│                      ▼                                      │
│               Metrics & Debug                               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

事件类型

支持的事件类型：

• 系统事件: SYSTEM_START, SYSTEM_READY, SYSTEM_SHUTDOWN, SYSTEM_ERROR
• Gateway事件: GATEWAY_STARTING, GATEWAY_READY, GATEWAY_STOPPING
• 渠道事件: CHANNEL_CONNECTED, CHANNEL_MESSAGE_RECEIVED, CHANNEL_MESSAGE_SENT
• 会话事件: SESSION_CREATED, SESSION_STARTED, SESSION_ENDED
• Agent事件: AGENT_INVOKED, AGENT_THINKING, AGENT_RESPONSE, AGENT_COMPLETED
• 工具事件: TOOL_INVOKED, TOOL_EXECUTING, TOOL_COMPLETED, TOOL_FAILED
• 恢复事件: RECOVERY_STARTED, RECOVERY_COMPLETED, RECOVERY_FAILED
• 指标事件: METRIC_COLLECTED, METRIC_THRESHOLD

开发

# 安装依赖
npm install

# 开发模式
npm run dev

# 构建
npm run build

# 测试
npm test

# 类型检查
npm run typecheck

许可证

MIT

相关

• Claw Code - 参考架构
• OpenClaw - 主项目