howler_core/
streaming.rs

1use crate::models::Sighting;
2use serde::{Deserialize, Serialize};
3use tokio::sync::broadcast;
4
5#[derive(Debug, Clone, Serialize, Deserialize)]
6#[serde(tag = "type")]
7pub enum StreamingEvent {
8    SightingCreated(Sighting),
9    SightingUpdated(Sighting),
10    AnalysisComplete(String),
11}
12
13#[derive(Clone)]
14pub struct Broadcast {
15    sender: broadcast::Sender<StreamingEvent>,
16}
17
18impl Broadcast {
19    pub fn new(capacity: usize) -> Self {
20        let (sender, _) = broadcast::channel(capacity);
21        Broadcast { sender }
22    }
23
24    pub fn subscribe(&self) -> broadcast::Receiver<StreamingEvent> {
25        self.sender.subscribe()
26    }
27
28    pub fn publish(
29        &self,
30        event: StreamingEvent,
31    ) -> Result<(), Box<broadcast::error::SendError<StreamingEvent>>> {
32        self.sender.send(event).map(|_| ()).map_err(Box::new)
33    }
34}
35
36#[cfg(test)]
37mod tests {
38    use super::*;
39    use crate::models::Source;
40    use chrono::Utc;
41
42    fn test_sighting() -> Sighting {
43        Sighting {
44            id: Some(1),
45            species: "Canis lupus".to_string(),
46            scientific_name: Some("Canis lupus".to_string()),
47            latitude: 45.0,
48            longitude: -122.0,
49            observed_on: Utc::now(),
50            source: Source::GBIF,
51            source_id: "test_1".to_string(),
52            details: Some("Test sighting".to_string()),
53        }
54    }
55
56    #[tokio::test]
57    async fn test_broadcast_publish_subscribe() {
58        let broadcast = Broadcast::new(32);
59        let mut rx = broadcast.subscribe();
60
61        let sighting = test_sighting();
62        let event = StreamingEvent::SightingCreated(sighting.clone());
63        broadcast.publish(event).unwrap();
64
65        let received = rx.recv().await.unwrap();
66        match received {
67            StreamingEvent::SightingCreated(s) => {
68                assert_eq!(s.species, "Canis lupus");
69                assert_eq!(s.id, Some(1));
70            }
71            _ => panic!("Expected SightingCreated"),
72        }
73    }
74
75    #[tokio::test]
76    async fn test_broadcast_multiple_subscribers() {
77        let broadcast = Broadcast::new(32);
78        let mut rx1 = broadcast.subscribe();
79        let mut rx2 = broadcast.subscribe();
80
81        let event = StreamingEvent::AnalysisComplete("done".to_string());
82        broadcast.publish(event).unwrap();
83
84        let received1 = rx1.recv().await.unwrap();
85        let received2 = rx2.recv().await.unwrap();
86
87        match (&received1, &received2) {
88            (StreamingEvent::AnalysisComplete(m1), StreamingEvent::AnalysisComplete(m2)) => {
89                assert_eq!(m1, "done");
90                assert_eq!(m2, "done");
91            }
92            _ => panic!("Expected AnalysisComplete"),
93        }
94    }
95}
howler_core/streaming.rs

howler_core/
streaming.rs
