howler-core 0.1.0

Core library for the Howler wolf-tracking platform: database, data ingestion, ML, web server, and real-time streaming
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

use crate::models::{Sighting, Source};
use crate::streaming::{Broadcast, StreamingEvent};
use chrono::Utc;

fn test_sighting(id: i64) -> Sighting {
    Sighting {
        id: Some(id),
        species: "Canis lupus".to_string(),
        scientific_name: Some("Canis lupus".to_string()),
        latitude: 45.0 + id as f64,
        longitude: -122.0 - id as f64,
        observed_on: Utc::now(),
        source: Source::GBIF,
        source_id: format!("test_{}", id),
        details: Some("Test sighting".to_string()),
    }
}

#[tokio::test]
async fn test_streaming_broadcast_publish_subscribe() {
    let broadcast = Broadcast::new(32);
    let mut rx = broadcast.subscribe();

    let sighting = test_sighting(1);
    let event = StreamingEvent::SightingCreated(sighting);
    broadcast.publish(event).unwrap();

    let received = rx.recv().await.unwrap();
    match received {
        StreamingEvent::SightingCreated(s) => {
            assert_eq!(s.species, "Canis lupus");
            assert_eq!(s.id, Some(1));
        }
        _ => panic!("Expected SightingCreated"),
    }
}

#[tokio::test]
async fn test_streaming_broadcast_multiple_events() {
    let broadcast = Broadcast::new(32);
    let mut rx = broadcast.subscribe();

    for i in 1..=3 {
        let sighting = test_sighting(i);
        broadcast
            .publish(StreamingEvent::SightingCreated(sighting))
            .unwrap();
    }

    for i in 1..=3 {
        let received = rx.recv().await.unwrap();
        match received {
            StreamingEvent::SightingCreated(s) => assert_eq!(s.id, Some(i)),
            _ => panic!("Expected SightingCreated"),
        }
    }
}

#[tokio::test]
async fn test_streaming_broadcast_analysis_complete() {
    let broadcast = Broadcast::new(32);
    let mut rx = broadcast.subscribe();

    let event = StreamingEvent::AnalysisComplete("Cluster analysis done".to_string());
    broadcast.publish(event).unwrap();

    let received = rx.recv().await.unwrap();
    match received {
        StreamingEvent::AnalysisComplete(msg) => {
            assert_eq!(msg, "Cluster analysis done");
        }
        _ => panic!("Expected AnalysisComplete"),
    }
}

#[tokio::test]
async fn test_streaming_broadcast_multiple_subscribers() {
    let broadcast = Broadcast::new(32);
    let mut rx1 = broadcast.subscribe();
    let mut rx2 = broadcast.subscribe();

    let sighting = test_sighting(1);
    let event = StreamingEvent::SightingUpdated(sighting);
    broadcast.publish(event).unwrap();

    let r1 = rx1.recv().await.unwrap();
    let r2 = rx2.recv().await.unwrap();

    match (&r1, &r2) {
        (StreamingEvent::SightingUpdated(s1), StreamingEvent::SightingUpdated(s2)) => {
            assert_eq!(s1.id, s2.id);
            assert_eq!(s1.species, s2.species);
        }
        _ => panic!("Expected SightingUpdated for both"),
    }
}

#[tokio::test]
async fn test_streaming_event_serialization() {
    let sighting = test_sighting(42);
    let event = StreamingEvent::SightingCreated(sighting);

    let json = serde_json::to_string(&event).unwrap();
    assert!(json.contains("SightingCreated"));
    assert!(json.contains("Canis lupus"));

    let deserialized: StreamingEvent = serde_json::from_str(&json).unwrap();
    match deserialized {
        StreamingEvent::SightingCreated(s) => {
            assert_eq!(s.id, Some(42));
            assert_eq!(s.species, "Canis lupus");
        }
        _ => panic!("Expected SightingCreated"),
    }
}

#[tokio::test]
async fn test_streaming_broadcast_event_types_roundtrip() {
    let broadcast = Broadcast::new(32);
    let mut rx = broadcast.subscribe();

    let sighting = test_sighting(10);
    let created = StreamingEvent::SightingCreated(sighting.clone());
    let updated = StreamingEvent::SightingUpdated(sighting.clone());
    let analysis = StreamingEvent::AnalysisComplete("done".to_string());

    broadcast.publish(created).unwrap();
    broadcast.publish(updated).unwrap();
    broadcast.publish(analysis).unwrap();

    let e1 = rx.recv().await.unwrap();
    let e2 = rx.recv().await.unwrap();
    let e3 = rx.recv().await.unwrap();

    assert!(matches!(e1, StreamingEvent::SightingCreated(_)));
    assert!(matches!(e2, StreamingEvent::SightingUpdated(_)));
    assert!(matches!(e3, StreamingEvent::AnalysisComplete(_)));

    let json1 = serde_json::to_string(&e1).unwrap();
    let d1: StreamingEvent = serde_json::from_str(&json1).unwrap();
    assert!(matches!(d1, StreamingEvent::SightingCreated(_)));
}