事件

快速开始

#[derive(Entity)]
#[entity(table = "orders", events)]
pub struct Order {
 #[id]
 pub id: Uuid,
 #[field(create, response)]
 pub customer_id: Uuid,
 #[field(create, update, response)]
 pub status: String,
 #[field(create, response)]
 pub total_cents: i64,
 #[field(response)]
 #[auto]
 pub created_at: DateTime<Utc>,
}

生成的代码

/// 由entity-derive生成
#[derive(Debug, Clone)]
pub enum OrderEvent {
 /// 实体已创建。
 Created(Order),
 /// 实体已更新。
 Updated {
 id: Uuid,
 changes: UpdateOrderRequest,
 },
 /// 实体已删除。
 Deleted(Uuid),
}

使用示例

基本事件发布

use async_trait::async_trait;
#[async_trait]
pub trait EventBus: Send + Sync {
 async fn publish<E: Send + Sync>(&self, event: E);
}
async fn create_order(
 repo: &impl OrderRepository,
 bus: &impl EventBus,
 dto: CreateOrderRequest,
) -> Result<Order, sqlx::Error> {
 let order = repo.create(dto).await?;
 // 创建成功后发布事件
 bus.publish(OrderEvent::Created(order.clone())).await;
 Ok(order)
}
async fn update_order(
 repo: &impl OrderRepository,
 bus: &impl EventBus,
 id: Uuid,
 dto: UpdateOrderRequest,
) -> Result<Order, sqlx::Error> {
 let order = repo.update(id, dto.clone()).await?;
 bus.publish(OrderEvent::Updated { id, changes: dto }).await;
 Ok(order)
}
async fn delete_order(
 repo: &impl OrderRepository,
 bus: &impl EventBus,
 id: Uuid,
) -> Result<bool, sqlx::Error> {
 let deleted = repo.delete(id).await?;
 if deleted {
 bus.publish(OrderEvent::Deleted(id)).await;
 }
 Ok(deleted)
}

审计日志

struct AuditLogger {
 pool: PgPool,
}
#[async_trait]
impl EventHandler<OrderEvent> for AuditLogger {
 async fn handle(&self, event: OrderEvent) {
 let (action, entity_id, details) = match &event {
 OrderEvent::Created(order) => (
 "created",
 order.id,
 serde_json::to_string(order).unwrap(),
 ),
 OrderEvent::Updated { id, changes } => (
 "updated",
 *id,
 serde_json::to_string(changes).unwrap(),
 ),
 OrderEvent::Deleted(id) => (
 "deleted",
 *id,
 String::new(),
 ),
 };
 sqlx::query(
 "INSERT INTO audit_log (entity_type, entity_id, action, details, created_at)
 VALUES ('order', 1,ドル 2,ドル 3,ドル NOW())"
 )
 .bind(entity_id)
 .bind(action)
 .bind(details)
 .execute(&self.pool)
 .await
 .ok();
 }
}

消息队列集成

use rdkafka::producer::FutureProducer;
struct KafkaEventBus {
 producer: FutureProducer,
 topic: String,
}
#[async_trait]
impl EventBus for KafkaEventBus {
 async fn publish<E: Serialize + Send + Sync>(&self, event: E) {
 let payload = serde_json::to_vec(&event).unwrap();
 self.producer
 .send(
 FutureRecord::to(&self.topic)
 .payload(&payload)
 .key(&Uuid::new_v4().to_string()),
 Duration::from_secs(5),
 )
 .await
 .ok();
 }
}

事件溯源模式

struct OrderAggregate {
 events: Vec<OrderEvent>,
 current_state: Option<Order>,
}
impl OrderAggregate {
 fn apply(&mut self, event: OrderEvent) {
 match &event {
 OrderEvent::Created(order) => {
 self.current_state = Some(order.clone());
 }
 OrderEvent::Updated { changes, .. } => {
 if let Some(ref mut order) = self.current_state {
 if let Some(status) = &changes.status {
 order.status = status.clone();
 }
 }
 }
 OrderEvent::Deleted(_) => {
 self.current_state = None;
 }
 }
 self.events.push(event);
 }
 fn replay(events: Vec<OrderEvent>) -> Self {
 let mut aggregate = Self {
 events: Vec::new(),
 current_state: None,
 };
 for event in events {
 aggregate.apply(event);
 }
 aggregate
 }
}

与软删除配合

#[derive(Entity)]
#[entity(table = "documents", events, soft_delete)]
pub struct Document {
 #[id]
 pub id: Uuid,
 #[field(create, response)]
 pub title: String,
 #[field(skip)]
 pub deleted_at: Option<DateTime<Utc>>,
}

pub enum DocumentEvent {
 Created(Document),
 Updated { id: Uuid, changes: UpdateDocumentRequest },
 Deleted(Uuid), // 软删除
 Restored(Uuid), // 从软删除恢复
 HardDeleted(Uuid), // 永久删除
}

最佳实践

与钩子结合

#[derive(Entity)]
#[entity(table = "orders", events, hooks)]
pub struct Order { /* ... */ }
struct OrderService {
 repo: PgPool,
 bus: EventBus,
}
#[async_trait]
impl OrderHooks for OrderService {
 type Error = AppError;
 async fn after_create(&self, entity: &Order) -> Result<(), Self::Error> {
 // 在钩子中发布事件
 self.bus.publish(OrderEvent::Created(entity.clone())).await;
 Ok(())
 }
 async fn after_update(&self, entity: &Order) -> Result<(), Self::Error> {
 // 事件也可以在这里发布
 Ok(())
 }
 async fn after_delete(&self, id: &Uuid) -> Result<(), Self::Error> {
 self.bus.publish(OrderEvent::Deleted(*id)).await;
 Ok(())
 }
}

另见