Postgres LISTEN/NOTIFY를 사용하여 엔티티 변경사항을 실시간으로 구독합니다. 스트림을 통해 실시간 대시보드, 즉시 알림, 캐시 무효화, 이벤트 기반 아키텍처를 구현할 수 있습니다.
빠른 시작
#[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,
}요구사항:
- 엔티티는
Serialize와Deserialize를 구현해야 합니다 (JSON 페이로드용) events와streams속성이 모두 필요합니다- Cargo.toml에서
streams기능을 활성화하세요
[dependencies]
entity-derive = { version = "0.3", features = ["postgres", "streams"] }
serde = { version = "1", features = ["derive"] }
생성되는 코드
streams 속성은 다음을 생성합니다:
채널 상수
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>>;
}자동 알림
CRUD 작업은 자동으로 이벤트를 발행합니다:
// 생성된 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 캐시
대시보드 무효화기
모범 사례
- 재연결 — PgListener는 자동 재연결됩니다; 일시적 실패를 처리하도록 루프를 설계하세요
- 멱등성 — 이벤트가 여러 번 전달될 수 있습니다; 핸들러는 멱등해야 합니다
- 페이로드 크기 — 엔티티를 작게 유지하세요; 큰 페이로드는 Postgres 제한에 걸릴 수 있습니다
- 별도 풀 — 리스너용 전용 연결 풀을 사용하세요
- 모니터링 — 스트림 오류를 로그하고 이벤트 처리 지연을 추적하세요
- Graceful shutdown — 리소스를 정리하기 위해 shutdown 신호와 함께 select!를 사용하세요
소프트 삭제와 함께
soft_delete가 활성화되면 추가 이벤트를 사용할 수 있습니다:
#[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 }참고
- [[이벤트|이벤트]] — 실시간 스트리밍 없는 이벤트 열거형
- [[훅|훅]] — 라이프사이클 이벤트에서 커스텀 로직 실행
- [[모범 사례|모범-사례]] — 프로덕션 팁