anycms-event

A type-safe, async event bus system for AnyCMS, built on tokio broadcast channels.

类型安全的异步事件总线，基于 tokio broadcast channels 实现，支持本地进程内通信与 Redis 跨进程分布式通信。

Features

• Type-safe — 编译期保证事件类型正确，无需手动类型转换
• Async-first — 基于 tokio 构建的异步 API，handler 以独立 tokio task 运行
• Thread-safe — EventBus 内部使用 Arc<RwLock<>>，可安全跨 task/线程共享
• Derive Macro — #[derive(Event)] 自动实现 Event trait，零样板代码
• event_bus! Macro — 一键定义事件结构体 + topic 分组 + 类型化总线
• Wildcard Topics — 支持 *（单段匹配）和 **（多段匹配）通配符
• Telemetry — 可插拔的遥测中间件，内置 tracing 和自定义实现
• Event Registry — 事件注册表，支持事件发现、查询和元数据管理
• Execution Log — 执行日志，追踪事件发布和 Handler 执行历史
• Trigger Rule Engine — 触发规则引擎，动态配置事件到动作的映射
• Testing Utilities — EventCollector 消除测试中的 sleep() 等待
• SSE Streaming — 通过 anycms-event-sse 将事件实时推送到前端
• Redis Transport — 通过 anycms-event-redis 实现跨进程事件传递
• Framework Integrations — anycms-event-axum / anycms-event-actix 简化框架集成

Workspace Structure

anycms-event/
├── src/                          # 核心事件总线库
├── crates/
│   ├── anycms-event-derive/      # 过程宏（#[derive(Event)] + event_bus!）
│   ├── anycms-event-redis/       # Redis 传输层（分布式支持）
│   ├── anycms-event-axum/        # Axum 集成辅助
│   ├── anycms-event-actix/       # Actix-web 集成辅助
│   └── anycms-event-sse/         # SSE 实时推流
├── examples/                     # 示例代码
└── tests/                        # 集成测试

Quick Start

添加依赖到 Cargo.toml：

[dependencies]
anycms-event = "0.1"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }

手动定义事件

use anycms_event::prelude::*;

#[derive(Clone, Debug, Serialize, Deserialize)]
struct UserCreated {
    user_id: u64,
    name: String,
}

impl Event for UserCreated {
    fn event_name() -> &'static str { "user.created" }
}

#[tokio::main]
async fn main() {
    let bus = EventBus::new();

    // 订阅事件
    bus.subscribe(|event: UserCreated| async move {
        println!("New user: {} (id={})", event.name, event.user_id);
        Ok(())
    }).await.unwrap();

    // 发布事件
    bus.publish(UserCreated { user_id: 1, name: "Alice".into() }).await.unwrap();
}

使用 #[derive(Event)] 宏

use anycms_event::prelude::*;

#[derive(Clone, Debug, Serialize, Deserialize, Event)]
struct UserCreated {
    user_id: u64,
    name: String,
}
// 自动生成: event_name() -> "user.created", topic() -> "user"

通过属性自定义名称：

#[derive(Clone, Debug, Serialize, Deserialize, Event)]
#[event(name = "user.registered", topic = "user")]
struct UserCreated {
    user_id: u64,
    name: String,
}

使用 event_bus! 宏

一键定义事件结构体、topic 分组和类型化事件总线：

use anycms_event::event_bus;

event_bus! {
    bus AppEventBus {
        event UserCreated { user_id: u64, username: String }
        event UserDeleted { user_id: u64, reason: String }
        event OrderPlaced { order_id: u64, product: String, amount: f64 }

        topic user_events => [UserCreated, UserDeleted]
    }
}

#[tokio::main]
async fn main() {
    let bus = AppEventBus::new();

    // 按 topic 分组订阅
    bus.subscribe_topic_user_events(|event: AppEventBusTopicEvent| async move {
        match event {
            AppEventBusTopicEvent::UserCreated(e) => println!("Created: {}", e.username),
            AppEventBusTopicEvent::UserDeleted(e) => println!("Deleted: {}", e.reason),
        }
        Ok(())
    }).await.unwrap();

    // 发布事件
    bus.publish(UserCreated { user_id: 1, username: "Alice".into() }).await.unwrap();
}

Core API

Event Trait

所有事件必须实现此 trait：

pub trait Event: Clone + Send + Sync + 'static {
    fn event_name() -> &'static str;  // 唯一事件名称，用于路由
    fn topic() -> &'static str;       // 所属 topic，默认等于 event_name
}

EventBus

Builder 模式

use anycms_event::telemetry::TracingTelemetry;
use anycms_event::execution_log::{ExecutionLog, ExecutionLogTelemetry};
use std::sync::Arc;

let bus = EventBus::builder()
    .capacity(2048)
    .telemetry(TracingTelemetry)
    .execution_log(Arc::new(ExecutionLog::in_memory()))
    .build();

Topic 通配符

System Management 系统管理

anycms-event 提供三大系统管理模块，支持事件发现、执行追踪和动态触发规则配置。

P1: Event Registry 事件注册表

自动跟踪已注册事件，支持事件发现和元数据查询：

// 事件在 publish/subscribe 时自动注册
bus.publish(UserCreated { user_id: 1, name: "Alice".into() }).await.unwrap();

let registry = bus.registry();

// 列出所有已注册事件
for desc in registry.list_all() {
    println!("{} (topic: {}, 发布次数: {})", desc.event_name, desc.topic, desc.publish_count);
}

// 按条件查询
let user_events = registry.query(EventQuery {
    name: Some("user.*".to_string()),  // 支持前缀通配
    topic: Some("user".to_string()),
    tags: vec!["auth".to_string()],
    search: Some("created".to_string()),  // 文本搜索
    limit: Some(10),
    offset: Some(0),
    ..Default::default()
});

// 手动注册带完整元数据的事件
registry.register(EventDescriptor {
    event_name: "system.maintenance".to_string(),
    topic: "system".to_string(),
    description: "系统维护事件".to_string(),
    schema: Some(serde_json::json!({"type": "object", "properties": {...}})),
    source_module: Some("anycms-system".to_string()),
    tags: vec!["system".to_string()],
    ..Default::default()
});

P2: Execution Log 执行日志

记录和查询事件发布和 Handler 执行历史：

use anycms_event::execution_log::{ExecutionLog, ExecutionLogTelemetry, ExecutionLogQuery};

// 创建共享存储，Telemetry 和查询接口使用同一个后端
let log = Arc::new(ExecutionLog::in_memory());
let bus = EventBus::builder()
    .telemetry(ExecutionLogTelemetry::new(ExecutionLog::in_memory()))
    .execution_log(log.clone())
    .build();

// 发布事件后查询执行日志
bus.publish(UserCreated { user_id: 1, name: "Alice".into() }).await.unwrap();

let log = bus.execution_log().unwrap();

// 查询所有发布记录
let publishes = log.query(ExecutionLogQuery {
    execution_type: Some(ExecutionType::Publish),
    ..Default::default()
});

// 查询失败的 Handler 执行
let failures = log.query(ExecutionLogQuery {
    status: Some(ExecutionStatus::Failed),
    event_name: Some("user.created".to_string()),
    since: Some(std::time::SystemTime::now() - Duration::from_secs(3600)),
    limit: Some(50),
    ..Default::default()
});

P3: Trigger Rule Engine 触发规则引擎

动态配置事件到动作的映射规则，可与 WorkflowEngine 集成：

use anycms_event::trigger::{TriggerRuleEngine, TriggerRule, TriggerContext};

let trigger_engine = TriggerRuleEngine::new(bus.clone());

// 注册动作处理器
trigger_engine.register_action("workflow", |ctx: TriggerContext| {
    // 在这里调用 WorkflowEngine.emit()
    // workflow_engine.emit(&ctx.event_name, ctx.event_data, &entity_id).await
    async move { Ok(()) }
});

trigger_engine.register_action("notify", |ctx: TriggerContext| {
    // 发送通知
    async move { Ok(()) }
});

// 动态配置触发规则
trigger_engine.add_rule(TriggerRule {
    id: "content-sitemap".into(),
    name: "内容发布→生成 Sitemap".into(),
    event_pattern: "content.published".into(),
    condition: None,
    action_type: "workflow".into(),
    action_config: serde_json::json!({"workflow_id": "generate-sitemap"}),
    enabled: true,
    priority: 0,
});

// 带条件过滤的规则
trigger_engine.add_rule(TriggerRule {
    id: "vip-order".into(),
    name: "VIP 订单处理".into(),
    event_pattern: "order.placed".into(),
    condition: Some(serde_json::json!({
        "amount": {"$gt": 1000},
        "customer_level": {"$gte": 3}
    })),
    action_type: "workflow".into(),
    action_config: serde_json::json!({"workflow_id": "vip-handler"}),
    enabled: true,
    priority: 0,
});

// 运行时管理规则
trigger_engine.disable_rule("content-sitemap");   // 禁用
trigger_engine.enable_rule("content-sitemap");    // 启用
trigger_engine.remove_rule("vip-order");          // 删除
trigger_engine.list_rules();                      // 列出所有

// 处理事件
let results = trigger_engine.process_event("content.published", &data).await;

条件匹配操作符：

Telemetry 遥测

可插拔的遥测中间件，监控事件总线的发布/订阅生命周期：

use anycms_event::telemetry::Telemetry;
use std::time::Duration;

// 自定义遥测实现
struct ConsoleTelemetry;

impl Telemetry for ConsoleTelemetry {
    fn on_publish(&self, event_name: &str, receivers: usize) {
        println!("[Telemetry] Publishing '{}' to {} receivers", event_name, receivers);
    }

    fn on_publish_complete(&self, event_name: &str, elapsed: Duration) {
        println!("[Telemetry] '{}' published in {:.2}ms", event_name, elapsed.as_secs_f64() * 1000.0);
    }

    fn on_subscribe(&self, event_name: &str, sub_id: usize) {
        println!("[Telemetry] Subscriber #{} for '{}'", sub_id, event_name);
    }

    fn on_handler_start(&self, event_name: &str, sub_id: usize) {}
    fn on_handler_complete(&self, event_name: &str, sub_id: usize, elapsed: Duration, error: Option<&str>) {}
    fn on_handler_lagged(&self, event_name: &str, sub_id: usize, count: usize) {}
}

// 使用自定义遥测
let bus = EventBus::builder()
    .telemetry(ConsoleTelemetry)
    .build();

内置实现：

• TracingTelemetry — 基于 tracing 的结构化日志
• NoopTelemetry — 空操作，用于禁用遥测
• ExecutionLogTelemetry — 将事件生命周期记录到执行日志

Testing 测试工具

EventCollector 消除测试中的 sleep() 等待，提供类型安全的事件断言。

启用方式：

[dev-dependencies]
anycms-event = { version = "0.1", features = ["testing"] }

use anycms_event::testing::EventCollector;
use std::time::Duration;

#[tokio::test]
async fn test_user_created_event() {
    let bus = EventBus::new();
    let collector = EventCollector::<UserCreated>::new(&bus).await;
    // 无需 sleep — 订阅在 new() 返回时即已生效

    // 执行业务操作
    bus.publish(UserCreated { user_id: 1, username: "alice".into() }).await.unwrap();

    // 断言
    collector.assert_count(1);
    collector.assert_contains(|e| e.username == "alice");
}

可用的断言方法：

SSE 实时推流

通过 anycms-event-sse 将 EventBus 事件通过 Server-Sent Events 推送到前端：

[dependencies]
anycms-event-sse = "0.1"

use anycms_event_sse::{SseBridge, filter::PatternFilter};

// 创建 SSE 桥接器，注册需要推送的事件类型
let (stream, _subscriptions) = SseBridge::new(bus)
    .subscribe_type::<UserCreated>()
    .subscribe_type::<OrderPlaced>()
    .with_filter(PatternFilter::new("user.*"))  // 可选：只推送匹配的事件
    .into_stream()
    .await;

// 在 Axum handler 中使用
use axum::response::sse::{Event, Sse, KeepAlive};

async fn event_stream(State(bus): State<EventBus>) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let (stream, _) = SseBridge::new(bus)
        .subscribe_type::<UserCreated>()
        .into_stream()
        .await;

    let mapped = stream.map(|result| {
        let e = result.unwrap();
        Ok(Event::default().event(e.event_type).data(e.data))
    });

    Sse::new(mapped).keep_alive(KeepAlive::default())
}

内置过滤器：

• AllowFilter — 白名单过滤
• DenyFilter — 黑名单过滤
• PatternFilter — 通配符过滤（复用 topic 匹配规则）

Framework Integration

Actix-Web

use actix_web::{web, App, HttpServer};

#[actix_web::main]
async fn main() {
    let bus = AppEventBus::new();

    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(bus.clone()))
            .route("/users", web::post().to(create_user))
    })
    .bind("127.0.0.1:8080")
    .unwrap()
    .run()
    .await
    .unwrap();
}

async fn create_user(bus: web::Data<AppEventBus>) -> impl Responder {
    bus.publish(UserCreated { ... }).await.unwrap();
    HttpResponse::Created().finish()
}

Axum

use axum::{Extension, Router};

#[tokio::main]
async fn main() {
    let bus = AppEventBus::new();

    let app = Router::new()
        .route("/users", post(create_user))
        .layer(Extension(bus.clone()));

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

async fn create_user(Extension(bus): Extension<AppEventBus>) -> impl IntoResponse {
    bus.publish(UserCreated { ... }).await.unwrap();
    (StatusCode::CREATED, Json(json!({ "status": "ok" })))
}

Framework Extractor Crates

anycms-event-axum 和 anycms-event-actix 提供 HasEventBus trait，统一框架集成模式：

use anycms_event_axum::HasEventBus;

// 为你的类型化总线实现 trait
impl HasEventBus for AppEventBus {
    fn event_bus(&self) -> &EventBus { self.inner() }
}

Redis 分布式通信

通过 anycms-event-redis 实现跨进程事件传递：

[dependencies]
anycms-event-redis = "0.1"

use anycms_event_redis::RedisTransport;

#[tokio::main]
async fn main() {
    let transport = RedisTransport::new("redis://127.0.0.1:6379").await.unwrap();
    let bus = EventBus::new();
    let bridged = transport.bridge(bus).await.unwrap();

    bridged.forward_from_redis::<UserCreated>().await.unwrap();

    bridged.subscribe(|e: UserCreated| async move {
        println!("Received: {}", e.name);
        Ok(())
    }).await.unwrap();

    bridged.publish(UserCreated { user_id: 1, username: "Alice".into() }).await.unwrap();
}

完整示例见 examples/redis_distributed.rs。

Examples

Error Handling

Handler 返回错误时会通过 tracing 记录日志，不会影响其他订阅者。如果配置了 Telemetry，错误信息会通过 on_handler_complete 回调上报。

use anycms_event::EventBusError;

match bus.publish(event).await {
    Ok(()) => {},
    Err(EventBusError::PublishFailed { event_name, reason }) => {
        // 发布失败
    }
    Err(_) => { /* 其他错误 */ }
}

Architecture

┌──────────────────────────────────────────────────────────┐
│              Trigger Rule Engine (P3)                     │
│  规则 CRUD / 事件模式匹配 / JSON 条件过滤 / 动作执行       │
│  register_action("workflow", |ctx| WorkflowEngine.emit()) │
├──────────────────────────────────────────────────────────┤
│        Event Registry (P1)      Execution Log (P2)       │
│  事件发现/查询/搜索/元数据       执行记录/状态/耗时追踪     │
│  bus.registry().query(...)      bus.execution_log()      │
├──────────────────────────────────────────────────────────┤
│                   Core EventBus                           │
│  publish / subscribe / pattern matching / telemetry       │
│  derive macros / SSE / Redis transport                    │
└──────────────────────────────────────────────────────────┘

License

MIT