스트림

빠른 시작

#[derive(Entity, Serialize, Deserialize)]
#[entity(table = "orders", events, streams)]
pub struct Order {
 #[id]
 pub id: Uuid,
 #[field(create, update, response)]
 pub status: String,
 #[field(create, response)]
 pub customer_id: Uuid,
}

[dependencies]
entity-derive = { version = "0.3", features = ["postgres", "streams"] }
serde = { version = "1", features = ["derive"] }

생성되는 코드

채널 상수

impl Order {
 /// Postgres NOTIFY 채널 이름.
 pub const CHANNEL: &'static str = "entity_orders";
}

구독자 구조체

/// 실시간 Order 변경 구독자.
pub struct OrderSubscriber {
 listener: PgListener,
}
impl OrderSubscriber {
 /// 연결하고 채널을 구독합니다.
 pub async fn new(pool: &PgPool) -> Result<Self, sqlx::Error>;
 /// 다음 이벤트를 대기합니다 (블로킹).
 pub async fn recv(&mut self) -> Result<OrderEvent, StreamError<sqlx::Error>>;
 /// 블로킹 없이 이벤트를 확인합니다.
 pub async fn try_recv(&mut self) -> Result<Option<OrderEvent>, StreamError<sqlx::Error>>;
}

자동 알림

// 생성된 create() 메서드 내부:
async fn create(&self, dto: CreateOrderRequest) -> Result<Order, Self::Error> {
 let order = /* insert */;
 // 자동 생성된 알림
 let event = OrderEvent::created(order.clone());
 let payload = serde_json::to_string(&event)?;
 sqlx::query("SELECT pg_notify(1,ドル 2ドル)")
 .bind(Order::CHANNEL)
 .bind(&payload)
 .execute(self)
 .await?;
 Ok(order)
}

사용 예제

기본 구독

use entity_derive::StreamError;
async fn watch_orders(pool: &PgPool) -> Result<(), Box<dyn std::error::Error>> {
 let mut subscriber = OrderSubscriber::new(pool).await?;
 loop {
 match subscriber.recv().await {
 Ok(event) => {
 match event {
 OrderEvent::Created(order) => {
 println!("새 주문: {}", order.id);
 }
 OrderEvent::Updated { old, new } => {
 println!("주문 {} 업데이트: {} -> {}", new.id, old.status, new.status);
 }
 OrderEvent::HardDeleted { id } => {
 println!("주문 {} 삭제됨", id);
 }
 _ => {}
 }
 }
 Err(StreamError::Database(e)) => {
 eprintln!("데이터베이스 오류: {}", e);
 break;
 }
 Err(StreamError::Deserialize(e)) => {
 eprintln!("잘못된 이벤트 페이로드: {}", e);
 }
 }
 }
 Ok(())
}

실시간 대시보드 (Axum WebSocket)

use axum::{
 extract::{State, WebSocketUpgrade, ws::{Message, WebSocket}},
 response::IntoResponse,
};
async fn ws_handler(
 ws: WebSocketUpgrade,
 State(pool): State<PgPool>,
) -> impl IntoResponse {
 ws.on_upgrade(|socket| handle_socket(socket, pool))
}
async fn handle_socket(mut socket: WebSocket, pool: PgPool) {
 let mut subscriber = match OrderSubscriber::new(&pool).await {
 Ok(s) => s,
 Err(_) => return,
 };
 loop {
 match subscriber.recv().await {
 Ok(event) => {
 let json = serde_json::to_string(&event).unwrap();
 if socket.send(Message::Text(json)).await.is_err() {
 break;
 }
 }
 Err(_) => break,
 }
 }
}

캐시 무효화

struct CacheInvalidator {
 cache: Redis,
 pool: PgPool,
}
impl CacheInvalidator {
 async fn run(&self) -> Result<(), StreamError<sqlx::Error>> {
 let mut subscriber = OrderSubscriber::new(&self.pool).await
 .map_err(StreamError::Database)?;
 loop {
 let event = subscriber.recv().await?;
 let key = format!("order:{}", event.entity_id());
 match event {
 OrderEvent::Created(_) | OrderEvent::Updated { .. } => {
 self.cache.del(&key).await.ok();
 }
 OrderEvent::HardDeleted { id } | OrderEvent::SoftDeleted { id } => {
 self.cache.del(&format!("order:{}", id)).await.ok();
 }
 _ => {}
 }
 }
 }
}

Graceful Shutdown이 있는 백그라운드 워커

use tokio::sync::watch;
async fn notification_worker(
 pool: PgPool,
 mut shutdown: watch::Receiver<bool>,
) {
 let mut subscriber = OrderSubscriber::new(&pool).await.unwrap();
 loop {
 tokio::select! {
 result = subscriber.recv() => {
 match result {
 Ok(event) => process_event(event).await,
 Err(e) => {
 eprintln!("스트림 오류: {:?}", e);
 tokio::time::sleep(Duration::from_secs(1)).await;
 }
 }
 }
 _ = shutdown.changed() => {
 println!("알림 워커 종료 중");
 break;
 }
 }
 }
}

오류 처리

use entity_derive::StreamError;
match subscriber.recv().await {
 Ok(event) => { /* 처리 */ }
 Err(StreamError::Database(sqlx_error)) => {
 // 연결 끊김, 쿼리 실패 등
 // 구독자는 다음 recv()에서 자동 재연결됩니다
 }
 Err(StreamError::Deserialize(message)) => {
 // 잘못된 JSON 페이로드
 // 로그하고 계속 - 루프를 중단하지 마세요
 }
}

아키텍처

CRUD 작업 (create/update/delete)
 │
 ▼
 pg_notify(channel, event_json)
 │
 ▼
 Postgres NOTIFY
 │
 ┌────┴────┐
 ▼ ▼
구독자 구독자 (여러 리스너)
 │ │
 ▼ ▼
 WebSocket 캐시
 대시보드 무효화기

모범 사례

소프트 삭제와 함께

#[derive(Entity, Serialize, Deserialize)]
#[entity(table = "documents", events, streams, soft_delete)]
pub struct Document {
 #[id]
 pub id: Uuid,
 #[field(create, response)]
 pub title: String,
 #[field(skip)]
 pub deleted_at: Option<DateTime<Utc>>,
}
// 이벤트에 포함:
// - DocumentEvent::SoftDeleted { id }
// - DocumentEvent::Restored { id }
// - DocumentEvent::HardDeleted { id }

참고