dahua_camera_server/
state.rs

1//! Shared state for every handler.
2//!
3//! Wraps the camera service and, when the `alarms` feature is on, the alarm
4//! registry. [`Deref`](std::ops::Deref) to the service keeps every existing
5//! handler that says `state.get(id)` working unchanged.
6
7use dahua_camera_rtsp::CameraService;
8use std::sync::Arc;
9
10/// What handlers are given.
11pub type AppState = Arc<AppStateInner>;
12
13/// The state behind [`AppState`].
14pub struct AppStateInner {
15    /// The cameras.
16    pub service: Arc<CameraService>,
17
18    /// Latched alarms, fed by a single aggregator task.
19    ///
20    /// A [`std::sync::Mutex`], not tokio's: every critical section is a few
21    /// map operations with no `.await` inside, so an async mutex would add a
22    /// scheduling hop for nothing.
23    #[cfg(feature = "alarms")]
24    pub alarms: Arc<std::sync::Mutex<dahua_camera_alarms::AlarmRegistry>>,
25}
26
27impl AppStateInner {
28    /// State for a service, with alarm tracking started if the feature is on.
29    ///
30    /// # Task budget
31    ///
32    /// The alarm aggregator is **one task for the whole service**, not one per
33    /// camera: it reads the already-merged event bus. The workspace budget
34    /// allows no feature more than one task per camera, and `events` already
35    /// spends that one.
36    pub fn new(service: Arc<CameraService>) -> AppState {
37        #[cfg(feature = "alarms")]
38        let alarms = {
39            let registry = Arc::new(std::sync::Mutex::new(
40                dahua_camera_alarms::AlarmRegistry::new(Default::default()),
41            ));
42            spawn_alarm_aggregator(service.clone(), registry.clone());
43            registry
44        };
45
46        Arc::new(Self {
47            service,
48            #[cfg(feature = "alarms")]
49            alarms,
50        })
51    }
52}
53
54/// Fold the merged event stream into the alarm registry.
55///
56/// One task total. Reads only; never touches the video path.
57#[cfg(feature = "alarms")]
58fn spawn_alarm_aggregator(
59    service: Arc<CameraService>,
60    registry: Arc<std::sync::Mutex<dahua_camera_alarms::AlarmRegistry>>,
61) {
62    let mut events = service.subscribe_events();
63
64    tokio::spawn(async move {
65        loop {
66            match events.recv().await {
67                Ok(event) => {
68                    let now = now_unix_ms();
69                    if let Ok(mut registry) = registry.lock() {
70                        registry.observe(&event, now);
71                    }
72                }
73                // Alarms are edge-triggered, so a dropped burst can leave a
74                // condition latched with no STOP coming. The auto-clear
75                // timeout is what recovers from exactly this.
76                Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
77                    tracing::warn!(
78                        skipped,
79                        "alarm aggregator lagged; any alarm whose STOP was dropped \
80                         will auto-clear on its timeout"
81                    );
82                }
83                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
84            }
85        }
86    });
87}
88
89/// Host wall-clock time in Unix milliseconds.
90#[cfg(feature = "alarms")]
91pub(crate) fn now_unix_ms() -> u64 {
92    std::time::SystemTime::now()
93        .duration_since(std::time::UNIX_EPOCH)
94        .unwrap_or_default()
95        .as_millis() as u64
96}
97
98/// So a handler written against the service keeps working unchanged.
99impl std::ops::Deref for AppStateInner {
100    type Target = CameraService;
101    fn deref(&self) -> &CameraService {
102        &self.service
103    }
104}
dahua_camera_server/state.rs

dahua_camera_server/
state.rs
