#![cfg_attr(coverage, allow(dead_code, unused_imports, unused_variables))] #![forbid(unsafe_code)] use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use tokio::sync::Mutex; use tonic::Status; use tracing::info; use crate::{camera, uvc_runtime, video}; struct CameraRelaySlot { profile: CameraRelayProfile, relay: Arc, } #[derive(Clone, Debug)] struct CameraRelayProfile { primary: camera::CameraConfig, hdmi_mirror: Option, } /// Manage the currently active camera relay instance. /// /// Inputs: camera configurations requested by incoming RPC streams. /// Outputs: a reusable relay handle plus a monotonically increasing session id. /// Why: only one camera output should own the physical sink at a time, but we /// still want identical stream requests to reuse the existing pipeline. pub struct CameraRuntime { generation: AtomicU64, slot: Mutex>, } impl CameraRuntime { /// Create an empty runtime with no active relay. /// /// Inputs: none. /// Outputs: a fresh runtime with generation zero. /// Why: keeping construction trivial lets the main server handler create /// camera state early and share it across RPCs. #[must_use] pub fn new() -> Self { Self { generation: AtomicU64::new(0), slot: Mutex::new(None), } } /// Activate the relay matching the current configuration. /// /// Inputs: the desired camera configuration selected from the environment. /// Outputs: a session id plus a relay that is either reused or recreated. /// Why: UVC/HDMI sinks are expensive to churn, so identical requests should /// reuse the active pipeline instead of rebuilding it every time. #[cfg(coverage)] pub async fn activate( &self, cfg: &camera::CameraConfig, ) -> Result<(u64, Arc, bool), Status> { let session_id = self.generation.fetch_add(1, Ordering::SeqCst) + 1; if matches!(cfg.output, camera::CameraOutput::Uvc) && std::env::var("LESAVKA_DISABLE_UVC").is_ok() { return Err(Status::failed_precondition( "UVC output disabled (LESAVKA_DISABLE_UVC set)", )); } Ok((session_id, Arc::new(video::CameraRelay::new_noop(0)), false)) } #[cfg(not(coverage))] pub async fn activate( &self, cfg: &camera::CameraConfig, ) -> Result<(u64, Arc, bool), Status> { let session_id = self.generation.fetch_add(1, Ordering::SeqCst) + 1; let requested_profile = camera_relay_profile(cfg); let mut slot = self.slot.lock().await; let mut reused = false; let relay = if let Some(existing) = slot.as_ref() { if camera_relay_profile_eq(&existing.profile, &requested_profile) { reused = true; existing.relay.clone() } else { self.make_relay(&requested_profile)? } } else { self.make_relay(&requested_profile)? }; if !reused { *slot = Some(CameraRelaySlot { profile: requested_profile.clone(), relay: relay.clone(), }); info!( session_id, output = requested_profile.primary.output.as_str(), codec = requested_profile.primary.codec.as_str(), width = requested_profile.primary.width, height = requested_profile.primary.height, fps = requested_profile.primary.fps, hdmi_mirror = requested_profile.hdmi_mirror.is_some(), "🎥 camera relay (re)created" ); } else { info!(session_id, "🎥 camera relay reused"); } Ok((session_id, relay, reused)) } /// Check whether a previously issued session id is still current. /// /// Inputs: a session id returned by `activate`. /// Outputs: `true` only when the session is still the most recent owner of /// the active camera relay. /// Why: superseded streams must stop writing frames into a sink that has /// already been reconfigured for a newer client session. #[must_use] pub fn is_active(&self, session_id: u64) -> bool { self.generation.load(Ordering::Relaxed) == session_id } /// Release the active relay when the owning stream has ended. /// /// Inputs: the camera session id returned by `activate`. /// Outputs: true only when that session was still current and was /// superseded. Why: keeping a completed UVC session's GStreamer pipeline /// alive preserves native buffers and makes allocator retention look like /// a server leak after the client UI disconnects. pub async fn release_if_active(&self, session_id: u64) -> bool { if self .generation .compare_exchange( session_id, session_id.saturating_add(1), Ordering::SeqCst, Ordering::Relaxed, ) .is_err() { return false; } let mut slot = self.slot.lock().await; let released = slot.take().is_some(); info!( session_id, released, "🎥 camera relay released after stream lifecycle ended" ); true } /// Supersede the active camera stream and drop the userspace relay sink. /// /// Inputs: none. /// Outputs: none. /// Why: UVC recovery should make the client reconnect and recreate the /// spool/appsrc pipeline without cycling the USB controller while a browser /// may still own the gadget. #[cfg(coverage)] pub async fn soft_recover(&self) { self.generation.fetch_add(1, Ordering::SeqCst); } #[cfg(not(coverage))] pub async fn soft_recover(&self) { self.generation.fetch_add(1, Ordering::SeqCst); let mut slot = self.slot.lock().await; let _dropped = slot.take(); } #[allow(clippy::result_large_err)] #[cfg(not(coverage))] fn make_relay(&self, profile: &CameraRelayProfile) -> Result, Status> { let cfg = &profile.primary; let relay = match cfg.output { camera::CameraOutput::Uvc => { if std::env::var("LESAVKA_DISABLE_UVC").is_ok() { return Err(Status::failed_precondition( "UVC output disabled (LESAVKA_DISABLE_UVC set)", )); } let uvc = uvc_runtime::pick_uvc_device() .map_err(|e| Status::internal(format!("{e:#}")))?; if let Some(hdmi_cfg) = profile.hdmi_mirror.as_ref() { let display_size = hdmi_cfg.hdmi_display_size(); info!( %uvc, hdmi = hdmi_cfg .hdmi .as_ref() .map(|h| h.name.as_str()) .unwrap_or("none"), display_width = display_size.0, display_height = display_size.1, "🎥 stream_camera using UVC sink with HDMI mirror" ); video::CameraRelay::new_uvc_with_hdmi_mirror(0, &uvc, cfg, hdmi_cfg) .map_err(|e| Status::internal(format!("{e:#}")))? } else { info!(%uvc, "🎥 stream_camera using UVC sink"); video::CameraRelay::new_uvc(0, &uvc, cfg) .map_err(|e| Status::internal(format!("{e:#}")))? } } camera::CameraOutput::Hdmi => video::CameraRelay::new_hdmi(0, cfg) .map_err(|e| Status::internal(format!("{e:#}")))?, }; Ok(Arc::new(relay)) } } impl Default for CameraRuntime { fn default() -> Self { Self::new() } } /// Compare two camera configurations for sink reuse. /// /// Inputs: the currently active camera config and the requested config. /// Outputs: `true` when both configs target the same sink and stream profile. /// Why: reusing a pipeline is only safe when both the transport parameters and /// the HDMI connector identity still match. #[must_use] pub fn camera_cfg_eq(a: &camera::CameraConfig, b: &camera::CameraConfig) -> bool { if a.output != b.output || a.codec != b.codec || a.width != b.width || a.height != b.height || a.fps != b.fps { return false; } if a.output == camera::CameraOutput::Hdmi && a.hdmi_display_size() != b.hdmi_display_size() { return false; } match (&a.hdmi, &b.hdmi) { (Some(left), Some(right)) => left.name == right.name && left.id == right.id, (None, None) => true, _ => false, } } fn camera_relay_profile(cfg: &camera::CameraConfig) -> CameraRelayProfile { let hdmi_mirror = (cfg.output == camera::CameraOutput::Uvc) .then(|| match hdmi_mirror_setting() { HdmiMirrorSetting::Enabled => Some(camera::hdmi_mirror_config(cfg)), HdmiMirrorSetting::Auto if hdmi_mirror_connector_is_configured() => { let mirror = camera::hdmi_mirror_config(cfg); mirror.hdmi.is_some().then_some(mirror) } HdmiMirrorSetting::Auto | HdmiMirrorSetting::Disabled => None, }) .flatten(); CameraRelayProfile { primary: cfg.clone(), hdmi_mirror, } } fn camera_relay_profile_eq(a: &CameraRelayProfile, b: &CameraRelayProfile) -> bool { camera_cfg_eq(&a.primary, &b.primary) && match (&a.hdmi_mirror, &b.hdmi_mirror) { (Some(left), Some(right)) => camera_cfg_eq(left, right), (None, None) => true, _ => false, } } #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum HdmiMirrorSetting { Disabled, Enabled, Auto, } fn hdmi_mirror_setting() -> HdmiMirrorSetting { std::env::var("LESAVKA_CAM_HDMI_MIRROR") .ok() .map(|value| match value.trim().to_ascii_lowercase().as_str() { "1" | "true" | "yes" | "on" => HdmiMirrorSetting::Enabled, "auto" => HdmiMirrorSetting::Auto, _ => HdmiMirrorSetting::Disabled, }) .unwrap_or(HdmiMirrorSetting::Disabled) } fn hdmi_mirror_connector_is_configured() -> bool { std::env::var("LESAVKA_HDMI_CONNECTOR") .ok() .is_some_and(|value| !value.trim().is_empty()) } #[cfg(test)] mod tests { use super::{HdmiMirrorSetting, camera_cfg_eq, camera_relay_profile, camera_relay_profile_eq}; use crate::camera::{CameraCodec, CameraConfig, CameraOutput, HdmiConnector}; use serial_test::serial; #[test] fn camera_cfg_eq_requires_matching_sink_profile() { let base = CameraConfig { output: CameraOutput::Hdmi, codec: CameraCodec::H264, width: 1920, height: 1080, fps: 30, hdmi: Some(HdmiConnector { name: String::from("HDMI-A-1"), id: Some(42), modes: Vec::new(), }), }; let same = base.clone(); assert!(camera_cfg_eq(&base, &same)); let mut changed = base.clone(); changed.fps = 25; assert!(!camera_cfg_eq(&base, &changed)); changed = base.clone(); changed.hdmi = Some(HdmiConnector { name: String::from("HDMI-A-2"), id: Some(42), modes: Vec::new(), }); assert!(!camera_cfg_eq(&base, &changed)); } #[test] #[serial] fn camera_relay_profile_tracks_hdmi_mirror_flag() { let base = CameraConfig { output: CameraOutput::Uvc, codec: CameraCodec::Mjpeg, width: 1280, height: 720, fps: 20, hdmi: None, }; temp_env::with_var("LESAVKA_CAM_HDMI_MIRROR", None::<&str>, || { let profile = camera_relay_profile(&base); assert!(profile.hdmi_mirror.is_none()); }); temp_env::with_var("LESAVKA_CAM_HDMI_MIRROR", Some("1"), || { let profile = camera_relay_profile(&base); let mirror = profile.hdmi_mirror.as_ref().expect("HDMI mirror config"); assert_eq!(mirror.output, CameraOutput::Hdmi); assert_eq!(mirror.codec, CameraCodec::Mjpeg); assert_eq!((mirror.width, mirror.height, mirror.fps), (1280, 720, 20)); }); } #[test] #[serial] fn camera_relay_profile_auto_mirrors_only_when_hdmi_is_detected() { let base = CameraConfig { output: CameraOutput::Uvc, codec: CameraCodec::Mjpeg, width: 1280, height: 720, fps: 20, hdmi: None, }; temp_env::with_vars( [ ("LESAVKA_CAM_HDMI_MIRROR", Some("auto")), ("LESAVKA_HDMI_CONNECTOR", None::<&str>), ], || { assert_eq!(super::hdmi_mirror_setting(), HdmiMirrorSetting::Auto); assert!(camera_relay_profile(&base).hdmi_mirror.is_none()); }, ); temp_env::with_vars( [ ("LESAVKA_CAM_HDMI_MIRROR", Some("auto")), ("LESAVKA_HDMI_CONNECTOR", Some("card1-HDMI-A-1")), ], || { let profile = camera_relay_profile(&base); let mirror = profile .hdmi_mirror .as_ref() .expect("auto HDMI mirror config"); assert_eq!(mirror.output, CameraOutput::Hdmi); assert_eq!(mirror.hdmi.as_ref().unwrap().name, "card1-HDMI-A-1"); }, ); } #[test] #[serial] fn camera_relay_profile_reuse_requires_same_mirror_connector() { let base = CameraConfig { output: CameraOutput::Uvc, codec: CameraCodec::Mjpeg, width: 1280, height: 720, fps: 20, hdmi: None, }; let left = CameraConfig { output: CameraOutput::Hdmi, hdmi: Some(HdmiConnector { name: String::from("HDMI-A-1"), id: Some(41), modes: Vec::new(), }), ..base.clone() }; let right = CameraConfig { output: CameraOutput::Hdmi, hdmi: Some(HdmiConnector { name: String::from("HDMI-A-2"), id: Some(42), modes: Vec::new(), }), ..base.clone() }; assert!(!camera_relay_profile_eq( &super::CameraRelayProfile { primary: base.clone(), hdmi_mirror: Some(left), }, &super::CameraRelayProfile { primary: base, hdmi_mirror: Some(right), }, )); } }