ecr_server/
events.rs

1use ecr_core::revision::Revision;
2use serde::{Deserialize, Serialize};
3use tokio::sync::broadcast;
4
5const CHANNEL_CAPACITY: usize = 256;
6
7#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
8#[serde(tag = "type", rename_all = "snake_case")]
9pub enum ServerEvent {
10    MailChanged {
11        revision: Revision,
12    },
13    TagsChanged {
14        revision: Revision,
15        ids: Vec<String>,
16    },
17    SyncStarted {
18        accounts: Vec<String>,
19    },
20    SyncProgress {
21        line: String,
22    },
23    SyncFinished {
24        new_messages: usize,
25        revision: Revision,
26    },
27    Error {
28        detail: String,
29    },
30}
31
32impl ServerEvent {
33    pub fn name(&self) -> &'static str {
34        match self {
35            ServerEvent::MailChanged { .. } => "mail:changed",
36            ServerEvent::TagsChanged { .. } => "tags:changed",
37            ServerEvent::SyncStarted { .. } => "sync:started",
38            ServerEvent::SyncProgress { .. } => "sync:progress",
39            ServerEvent::SyncFinished { .. } => "sync:finished",
40            ServerEvent::Error { .. } => "error",
41        }
42    }
43}
44
45#[derive(Clone)]
46pub struct EventBus {
47    sender: broadcast::Sender<ServerEvent>,
48}
49
50impl Default for EventBus {
51    fn default() -> Self {
52        Self::new()
53    }
54}
55
56impl EventBus {
57    pub fn new() -> Self {
58        let (sender, _) = broadcast::channel(CHANNEL_CAPACITY);
59        Self { sender }
60    }
61
62    pub fn publish(&self, event: ServerEvent) {
63        let _ = self.sender.send(event);
64    }
65
66    pub fn subscribe(&self) -> broadcast::Receiver<ServerEvent> {
67        self.sender.subscribe()
68    }
69
70    pub fn subscriber_count(&self) -> usize {
71        self.sender.receiver_count()
72    }
73}
74
75pub struct SyncProgress {
76    bus: EventBus,
77}
78
79impl SyncProgress {
80    pub fn new(bus: EventBus) -> Self {
81        Self { bus }
82    }
83}
84
85impl ecr_store::ProgressSink for SyncProgress {
86    fn line(&self, text: &str) {
87        self.bus.publish(ServerEvent::SyncProgress {
88            line: text.to_string(),
89        });
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96
97    fn revision() -> Revision {
98        Revision::new("uuid", 1)
99    }
100
101    #[test]
102    fn events_carry_stable_wire_names() {
103        assert_eq!(
104            ServerEvent::MailChanged {
105                revision: revision()
106            }
107            .name(),
108            "mail:changed"
109        );
110        assert_eq!(
111            ServerEvent::SyncFinished {
112                new_messages: 0,
113                revision: revision()
114            }
115            .name(),
116            "sync:finished"
117        );
118    }
119
120    #[test]
121    fn events_serialize_with_a_discriminating_type_field() {
122        let json = serde_json::to_string(&ServerEvent::MailChanged {
123            revision: revision(),
124        })
125        .unwrap();
126
127        assert!(json.contains(r#""type":"mail_changed""#), "{json}");
128        assert!(json.contains(r#""lastmod":1"#), "{json}");
129    }
130
131    #[tokio::test]
132    async fn subscribers_receive_published_events() {
133        let bus = EventBus::new();
134        let mut rx = bus.subscribe();
135
136        bus.publish(ServerEvent::MailChanged {
137            revision: revision(),
138        });
139
140        let received = rx.recv().await.unwrap();
141        assert_eq!(
142            received,
143            ServerEvent::MailChanged {
144                revision: revision()
145            }
146        );
147    }
148
149    #[tokio::test]
150    async fn publishing_with_no_subscribers_does_not_fail() {
151        let bus = EventBus::new();
152        bus.publish(ServerEvent::Error {
153            detail: "nobody listening".to_string(),
154        });
155        assert_eq!(bus.subscriber_count(), 0);
156    }
157
158    #[tokio::test]
159    async fn every_subscriber_sees_the_same_event() {
160        let bus = EventBus::new();
161        let mut a = bus.subscribe();
162        let mut b = bus.subscribe();
163
164        bus.publish(ServerEvent::SyncStarted {
165            accounts: vec!["main".to_string()],
166        });
167
168        assert_eq!(a.recv().await.unwrap(), b.recv().await.unwrap());
169    }
170
171    #[tokio::test]
172    async fn the_progress_sink_publishes_each_line() {
173        use ecr_store::ProgressSink;
174
175        let bus = EventBus::new();
176        let mut rx = bus.subscribe();
177        let sink = SyncProgress::new(bus.clone());
178
179        sink.line("C: 1/2");
180
181        assert_eq!(
182            rx.recv().await.unwrap(),
183            ServerEvent::SyncProgress {
184                line: "C: 1/2".to_string()
185            }
186        );
187    }
188}
ecr_server/events.rs

ecr_server/
events.rs
