dahua_camera_rtsp/

camera_service.rs

//! The service: owns every camera, and is the only thing HTTP talks to.

use crate::frame_router::{FrameRouter, Subscription};
use crate::ports::VideoSourceFactory;
use dahua_camera_core::ports::{CameraControl, CameraControlFactory, EventSourceFactory};
use crate::supervisor::{supervise, supervise_link, RestartPolicy, SupervisedLink, SupervisorCtx};
use dahua_camera_core::{
    AppConfig, CameraConfig, CameraStats, CameraStatus, CapabilitySet, Event, SharedParams,
};
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::broadcast;
use tokio_util::sync::CancellationToken;

/// Depth of the shared analytics-event ring.
const EVENT_BUS_CAPACITY: usize = 128;


/// The adapters a [`CameraService`] runs on.
///
/// This is the seam where concrete implementations enter the application
/// layer. Production wiring lives in the `dahua-camera` facade; tests substitute the
/// doubles in [`testing`](crate::testing).
#[derive(Clone)]
pub struct ServiceDeps {
    /// Builds video sources. Required.
    pub source_factory: Arc<dyn VideoSourceFactory>,
    /// Builds out-of-band control clients. Required.
    pub control_factory: Arc<dyn CameraControlFactory>,
    /// Builds analytics-event sources. `None` disables event collection.
    pub event_factory: Option<Arc<dyn EventSourceFactory>>,
    /// Builds the decode lane's codecs.
    #[cfg(feature = "decode")]
    pub codec_factory: Arc<dyn crate::ports::CodecFactory>,
    /// Restart timings applied to every camera.
    pub policy: RestartPolicy,
}

/// One camera, as the rest of the program sees it.
///
/// Holds no I/O of its own: the supervisor writes into the router, consumers
/// read out of it, and control commands go straight to the adapter.
pub struct Camera {
    id: String,
    config: CameraConfig,
    stats: Arc<CameraStats>,
    router: Arc<FrameRouter>,
    control: Arc<dyn CameraControl>,
    /// What the device reports it can do.
    ///
    /// Starts **unprobed**, so every capability-gated call refuses until a
    /// probe fills it in. A `RwLock` because reads are frequent (every HTTP
    /// status poll) and writes happen once, at probe time.
    capabilities: std::sync::RwLock<CapabilitySet>,
    #[cfg(feature = "decode")]
    jpeg: tokio::sync::Mutex<Option<crate::decoder::JpegLane>>,
    #[cfg(feature = "decode")]
    codec_factory: Arc<dyn crate::ports::CodecFactory>,
}

impl Camera {
    /// Stable identifier.
    pub fn id(&self) -> &str {
        &self.id
    }

    /// This camera's configuration.
    pub fn config(&self) -> &CameraConfig {
        &self.config
    }

    /// Live counters.
    pub fn stats(&self) -> &CameraStats {
        &self.stats
    }

    /// Current lifecycle state.
    pub fn status(&self) -> CameraStatus {
        self.router.status()
    }

    /// Active codec parameters, if a session has published any.
    pub fn params(&self) -> Option<SharedParams> {
        self.router.params()
    }

    /// How many consumers are attached.
    pub fn viewer_count(&self) -> usize {
        self.router.viewer_count()
    }

    /// Out-of-band control for this camera.
    pub fn control(&self) -> &Arc<dyn CameraControl> {
        &self.control
    }

    /// What this camera reports it can do.
    ///
    /// Empty and unprobed until something calls
    /// [`set_capabilities`](Self::set_capabilities). Callers must treat an
    /// unprobed profile as "refuse", never as "assume yes".
    pub fn capabilities(&self) -> CapabilitySet {
        self.capabilities.read().expect("capabilities lock").clone()
    }

    /// Record the result of a capability probe.
    pub fn set_capabilities(&self, capabilities: CapabilitySet) {
        *self.capabilities.write().expect("capabilities lock") = capabilities;
    }

    /// Attach to the frame bus.
    ///
    /// For a camera without `autostart`, this is what causes the supervisor to
    /// connect. Dropping the returned [`Subscription`] releases the viewer
    /// slot and eventually lets the session be torn down.
    pub fn subscribe(&self) -> Subscription {
        self.router.subscribe(self.stats.clone())
    }

    /// Wait for codec parameters, subscribing first so a lazy camera starts.
    ///
    /// Returns `None` if nothing arrived within `timeout`, which normally
    /// means the camera is unreachable.
    pub async fn wait_for_params(&self, timeout: Duration) -> Option<SharedParams> {
        self.router.wait_for_params(timeout).await
    }
}

#[cfg(feature = "decode")]
impl Camera {
    /// Get or start this camera's JPEG decode lane.
    ///
    /// The lane is created on first use and shared by every MJPEG and snapshot
    /// consumer, so N viewers cost one decode, not N. It shuts itself down
    /// once the last handle is dropped.
    pub async fn jpeg_lane(&self) -> crate::decoder::JpegLane {
        let mut slot = self.jpeg.lock().await;
        if let Some(lane) = slot.as_ref() {
            if lane.is_alive() {
                return lane.clone();
            }
        }
        let lane = crate::decoder::spawn_lane(
            &self.id,
            self.config.jpeg_quality,
            self.subscribe(),
            self.stats.clone(),
            self.codec_factory.clone(),
        );
        *slot = Some(lane.clone());
        lane
    }
}

/// Owns every configured camera and their supervisors.
pub struct CameraService {
    cameras: BTreeMap<String, Arc<Camera>>,
    events_tx: broadcast::Sender<Event>,
    cancel: CancellationToken,
    /// Long-lived tasks this service spawned, for the cost budget.
    ///
    /// Counted at the spawn sites rather than read from the runtime, because
    /// `RuntimeMetrics::spawned_tasks_count` needs `tokio_unstable`, and
    /// because this counts exactly what the budget is about — supervised links
    /// — rather than every transient task on the runtime.
    spawned_tasks: std::sync::atomic::AtomicUsize,
}

impl CameraService {
    /// Build the service and spawn one supervisor per camera.
    ///
    /// Returns immediately: cameras with `autostart` begin connecting in the
    /// background, the rest stay idle until something subscribes.
    pub fn start(config: &AppConfig, deps: ServiceDeps) -> Arc<Self> {
        let cancel = CancellationToken::new();
        let (events_tx, _) = broadcast::channel(EVENT_BUS_CAPACITY);
        let mut cameras = BTreeMap::new();
        let mut spawned = 0usize;

        for (id, camera_config) in &config.cameras {
            let router = Arc::new(FrameRouter::new());
            let stats = Arc::new(CameraStats::default());

            cameras.insert(
                id.clone(),
                Arc::new(Camera {
                    id: id.clone(),
                    config: camera_config.clone(),
                    stats: stats.clone(),
                    router: router.clone(),
                    control: deps.control_factory.create(camera_config),
                    capabilities: std::sync::RwLock::new(CapabilitySet::default()),
                    #[cfg(feature = "decode")]
                    jpeg: tokio::sync::Mutex::new(None),
                    #[cfg(feature = "decode")]
                    codec_factory: deps.codec_factory.clone(),
                }),
            );

            spawned += 1;
            tokio::spawn(supervise(SupervisorCtx {
                id: id.clone(),
                config: camera_config.clone(),
                router,
                stats,
                source_factory: deps.source_factory.clone(),
                cancel: cancel.child_token(),
                policy: deps.policy.clone(),
            }));

            // The event channel is a link like any other: same loop, same
            // backoff, same panic isolation — and its own child token, so it
            // reconnects entirely independently of video.
            if let Some(factory) = &deps.event_factory {
                let link = Arc::new(EventLink {
                    source: factory.create(camera_config),
                    sink: events_tx.clone(),
                });
                spawned += 1;
                tokio::spawn(supervise_link(
                    id.clone(),
                    link,
                    cancel.child_token(),
                    deps.policy.clone(),
                ));
            }
        }

        Arc::new(Self {
            cameras,
            events_tx,
            cancel,
            spawned_tasks: std::sync::atomic::AtomicUsize::new(spawned),
        })
    }

    /// Look up one camera.
    pub fn get(&self, id: &str) -> Option<Arc<Camera>> {
        self.cameras.get(id).cloned()
    }

    /// Every camera, in stable id order.
    pub fn all(&self) -> impl Iterator<Item = &Arc<Camera>> {
        self.cameras.values()
    }

    /// Attach to the merged analytics-event stream from all cameras.
    pub fn subscribe_events(&self) -> broadcast::Receiver<Event> {
        self.events_tx.subscribe()
    }

    /// How many long-lived tasks this service spawned.
    ///
    /// One supervised link per task. The workspace budget allows **no feature
    /// more than one task per camera**, and `tests/task_budget.rs` asserts it
    /// against this number.
    pub fn spawned_task_count(&self) -> usize {
        self.spawned_tasks.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Stop every camera. Supervisors wind down asynchronously.
    pub fn shutdown(&self) {
        self.cancel.cancel();
    }
}

/// One camera's analytics event channel, as a [`SupervisedLink`].
///
/// This exists instead of a bespoke retry loop. The previous implementation
/// was a bare `tokio::spawn` with a flat ten-second retry: no exponential
/// backoff, no distinction between a dropped connection and rejected
/// credentials, and — because nothing awaited its `JoinHandle` — a panic in
/// the parser killed event reporting for that camera silently, for the life of
/// the process.
struct EventLink {
    source: Arc<dyn dahua_camera_core::ports::EventSource>,
    sink: broadcast::Sender<Event>,
}

#[async_trait::async_trait]
impl SupervisedLink for EventLink {
    fn name(&self) -> &'static str {
        "events"
    }

    async fn run_once(&self, cancel: CancellationToken) -> dahua_camera_core::Result<()> {
        let mut stream = self.source.connect().await?;
        tracing::info!("event stream connected");

        loop {
            let next = tokio::select! {
                biased;
                _ = cancel.cancelled() => return Ok(()),
                r = stream.next() => r,
            };

            match next? {
                // A send error means nothing is subscribed to the event bus.
                // That is not a fault: events are advisory, and a service with
                // no event consumers should not be reconnecting over it.
                Some(event) => {
                    let _ = self.sink.send(event);
                }
                None => {
                    tracing::debug!("event stream closed by peer");
                    return Err(dahua_camera_core::CameraError::Cgi {
                        camera_id: String::new(),
                        detail: "event stream closed".to_owned(),
                    });
                }
            }
        }
    }
}