use std::collections::HashMap; use serde::Serialize; use tokio::sync::{RwLock, broadcast}; #[derive(Debug, Clone, Serialize)] #[serde(tag = "type", content = "data")] #[allow(dead_code)] pub enum MailEvent { NewMessages { folder: String, count: u32, latest_uid: Option, latest_sender: Option, latest_subject: Option, }, FlagsChanged { folder: String, }, FolderUpdated, } pub struct EventBus { channels: RwLock>>, } impl EventBus { pub fn new() -> Self { Self { channels: RwLock::new(HashMap::new()), } } pub async fn subscribe(&self, user_hash: &str) -> broadcast::Receiver { { let channels = self.channels.read().await; if let Some(tx) = channels.get(user_hash) { return tx.subscribe(); } } let mut channels = self.channels.write().await; let tx = channels.entry(user_hash.to_string()).or_insert_with(|| { let (tx, _) = broadcast::channel(64); tx }); tx.subscribe() } pub async fn publish(&self, user_hash: &str, event: MailEvent) { let channels = self.channels.read().await; if let Some(tx) = channels.get(user_hash) { let _ = tx.send(event); } } pub async fn cleanup(&self, user_hash: &str) { let mut channels = self.channels.write().await; if let Some(tx) = channels.get(user_hash) && tx.receiver_count() == 0 { channels.remove(user_hash); } } } impl Default for EventBus { fn default() -> Self { Self::new() } } #[cfg(test)] mod tests { use super::*; use std::sync::Arc; #[tokio::test] async fn subscribe_and_publish() { let bus = Arc::new(EventBus::new()); let mut rx: tokio::sync::broadcast::Receiver = bus.subscribe("user1").await; bus.publish( "user1", MailEvent::NewMessages { folder: "INBOX".to_string(), count: 1, latest_uid: Some(42), latest_sender: Some("alice@example.com".to_string()), latest_subject: Some("Hello".to_string()), }, ) .await; let event: MailEvent = rx.recv().await.unwrap(); match event { MailEvent::NewMessages { folder, count, .. } => { assert_eq!(folder, "INBOX"); assert_eq!(count, 1); } _ => panic!("unexpected event type"), } } #[tokio::test] async fn publish_to_nonexistent_user_is_noop() { let bus = EventBus::new(); bus.publish("ghost", MailEvent::FolderUpdated).await; } #[tokio::test] async fn multiple_subscribers_receive_same_event() { let bus = Arc::new(EventBus::new()); let mut rx1: tokio::sync::broadcast::Receiver = bus.subscribe("user1").await; let mut rx2: tokio::sync::broadcast::Receiver = bus.subscribe("user1").await; bus.publish("user1", MailEvent::FolderUpdated).await; let e1: MailEvent = rx1.recv().await.unwrap(); assert!(matches!(e1, MailEvent::FolderUpdated)); let e2: MailEvent = rx2.recv().await.unwrap(); assert!(matches!(e2, MailEvent::FolderUpdated)); } #[tokio::test] async fn cleanup_removes_empty_channel() { let bus = EventBus::new(); let rx: tokio::sync::broadcast::Receiver = bus.subscribe("user1").await; drop(rx); bus.cleanup("user1").await; let channels = bus.channels.read().await; assert!(!channels.contains_key("user1")); } #[tokio::test] async fn cleanup_keeps_channel_with_active_subscribers() { let bus = EventBus::new(); let _rx: tokio::sync::broadcast::Receiver = bus.subscribe("user1").await; bus.cleanup("user1").await; let channels = bus.channels.read().await; assert!(channels.contains_key("user1")); } }