rickub
nandi/jolt-nativepublic⑂ Fork 0
⑂ c41903b106d04e5cf80828c49e2d7d456a54739d
Branches
main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/jolt-native.git
git clone ssh://git@rickub.com/nandi/jolt-native.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

av_media.rs · 2807 lines · 106.1 KBRust Blame HistoryRaw
Lift freeq's AV media plane out of sleek 90f8b89 nandi 20d ago1//! Native MoQ media plane for freeq AV calls.
2//!
3//! Publishes mic Opus (+ optional camera H.264) to the SFU and plays remote
4//! audio/video via `iroh-live` + `moq-native` — same stack as freeq-av /
5//! freeq-sdk-ffi.
6//!
7//! Android: mic/speaker via cpal aaudio; local camera via Camera2
8//! (`CameraCapture` Java helper in APK `classes.dex`) pushing NV12 into a
9//! [`VideoSource`]. Remote H.264 decodes in-software (or MediaCodec when
10//! available) when peers publish video.
11
12use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
13use std::sync::Arc;
14
15use anyhow::{Context, Result};
16use iroh_live::media::{
17 audio_backend::AudioBackend,
18 codec::{AudioCodec, VideoCodec},
19 format::{AudioPreset, VideoPreset},
20 publish::LocalBroadcast,
21 subscribe::RemoteBroadcast,
22 traits::VideoSource,
23};
24use tokio::sync::{mpsc, oneshot, watch};
25
26use crate::av::{
27 broadcast_path, path_key, should_tap, MicLevel, VideoFrameStore, LOCAL_PREVIEW_KEY,
28};
29
30#[cfg(not(target_os = "android"))]
31use iroh_live::media::{
32 audio_backend::{AudioBackendOpts, DeviceId},
33 capture::{CameraCapturer, CameraConfig, CameraSelector},
34};
35
36// ── Device enumeration (desktop) ───────────────────────────────────────────
37
38/// One selectable capture/playback device for the UI.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub struct MediaDevice {
41 /// Stable-ish key stored in prefs (`CameraInfo.id` or audio device name).
42 pub id: String,
43 /// Human label for combo boxes.
44 pub name: String,
45 pub is_default: bool,
46}
47
48/// Heuristic: virtual / loopback cameras (prefer real USB cams when falling back).
49fn is_virtual_camera(name: &str, id: &str) -> bool {
50 let s = format!("{name} {id}").to_ascii_lowercase();
51 s.contains("virtual")
52 || s.contains("obs")
53 || s.contains("loopback")
54 || s.contains("dummy")
55 || s.contains("v4l2loopback")
56}
57
58/// List cameras. Hardware first; virtual (OBS/loopback) last on desktop.
59/// On Android, front-facing cameras are listed first.
60pub fn list_cameras() -> Vec<MediaDevice> {
61 #[cfg(not(target_os = "android"))]
62 {
63 match CameraCapturer::list() {
64 Ok(mut cams) => {
65 cams.sort_by(|a, b| {
66 let av = is_virtual_camera(&a.name, &a.id);
67 let bv = is_virtual_camera(&b.name, &b.id);
68 av.cmp(&bv)
69 .then_with(|| a.name.to_lowercase().cmp(&b.name.to_lowercase()))
70 });
71 cams.into_iter()
72 .map(|c| {
73 let label = if c.id.is_empty() || c.id == c.name {
74 c.name.clone()
75 } else {
76 format!("{} · {}", c.name, c.id)
77 };
78 MediaDevice {
79 id: c.id,
80 name: label,
81 is_default: false,
82 }
83 })
84 .collect()
85 }
86 Err(e) => {
87 log::debug!("av-media: list cameras: {e}");
88 Vec::new()
89 }
90 }
91 }
92 #[cfg(target_os = "android")]
93 {
94 crate::android_camera::list_cameras()
95 .into_iter()
96 .enumerate()
97 .map(|(i, (id, name))| MediaDevice {
98 id,
99 name,
100 is_default: i == 0,
101 })
102 .collect()
103 }
104}
105
106/// ALSA plugin / exclusive-card names that fight PipeWire (dmix slave busy, or
107/// exclusive `hw`/`sysdefault:CARD=` while WirePlumber already owns the card).
108/// Prefer the native PipeWire host (or ALSA `pipewire` PCM) instead.
109fn is_alsa_virtual_pcm(name: &str) -> bool {
110 let n = name.trim();
111 // Bare plugins.
112 if matches!(
113 n,
114 "sysdefault"
115 | "default"
116 | "dmix"
117 | "dsnoop"
118 | "hw"
119 | "plughw"
120 | "null"
121 | "pulse"
122 | "jack"
123 | "upmix"
124 | "vdownmix"
125 | "surround21"
126 | "surround40"
127 | "surround41"
128 | "surround50"
129 | "surround51"
130 | "surround71"
131 // ALSA→PipeWire bridge name; use system default / native PW host.
132 | "pipewire"
133 ) {
134 return true;
135 }
136 // Card-qualified exclusive ALSA paths (e.g. sysdefault:CARD=C960 for EMEET).
137 // Opening these bypasses PipeWire and fails when PW/OBS already hold the card.
138 n.starts_with("sysdefault:")
139 || n.starts_with("default:")
140 || n.starts_with("dmix:")
141 || n.starts_with("dsnoop:")
142 || n.starts_with("front:")
143 || n.starts_with("rear:")
144 || n.starts_with("center_lfe:")
145 || n.starts_with("side:")
146 || n.starts_with("hw:")
147 || n.starts_with("plughw:")
148 || n.starts_with("surround")
149 || n.starts_with("iec958:")
150 || n.starts_with("hdmi:")
151}
152
153/// Drop persisted mic/speaker ids that are known-broken or redundant under
154/// PipeWire so the UI shows "System default" (OS default source/sink — same as
155/// the browser: currently the EMEET SmartCam mic when that is the default).
156pub fn sanitize_audio_device_pref(id: Option<String>) -> Option<String> {
157 id.filter(|s| !s.is_empty()).and_then(|s| {
158 if is_alsa_virtual_pcm(&s) {
159 None
160 } else {
161 Some(s)
162 }
163 })
164}
165
166/// Prefer the native PipeWire host when available (lists real sources like
167/// "EMEET SmartCam C960 Mono" the same way Chromium does).
168#[cfg(not(target_os = "android"))]
169fn preferred_audio_host() -> Option<String> {
170 let hosts = AudioBackend::available_hosts();
171 if hosts
172 .iter()
173 .any(|h| h.eq_ignore_ascii_case("pipewire"))
174 {
175 Some("PipeWire".into())
176 } else {
177 None
178 }
179}
180
181/// Sort for UI: system default first, then remaining real devices.
182#[cfg(not(target_os = "android"))]
183fn rank_audio_device(name: &str, is_default: bool) -> (u8, String) {
184 let n = name.to_ascii_lowercase();
185 let rank = if is_default {
186 0
187 } else if is_alsa_virtual_pcm(name) {
188 9
189 } else {
190 5
191 };
192 (rank, n)
193}
194
195/// Names that are cpal PipeWire placeholders, stream monitors, or sinks
196/// mis-listed as capture (Duplex sinks show up as inputs).
197fn is_junk_capture_name(name: &str) -> bool {
198 let n = name.trim().to_ascii_lowercase();
199 if n.is_empty() || n == "unknown" {
200 return true;
201 }
202 matches!(
203 n.as_str(),
204 "default_input"
205 | "default_output"
206 | "default_sink"
207 | "sink_default"
208 | "input_default"
209 | "output_default"
210 ) || n.starts_with("alsa_output.")
211 || n.contains("monitor of")
212}
213
214#[cfg(not(target_os = "android"))]
215fn map_audio_devices(
216 raw: impl IntoIterator<Item = iroh_live::media::audio_backend::AudioDevice>,
217 inputs: bool,
218) -> Vec<MediaDevice> {
219 let mut devices: Vec<MediaDevice> = raw
220 .into_iter()
221 .filter(|d| !is_alsa_virtual_pcm(&d.name))
222 .filter(|d| !is_junk_capture_name(&d.name))
223 // Prefer unique names (cpal PW can list the same nick twice).
224 .map(|d| MediaDevice {
225 id: d.name.clone(),
226 name: if d.is_default {
227 format!("{} (system default)", d.name)
228 } else {
229 d.name
230 },
231 is_default: d.is_default,
232 })
233 .collect();
234 // Dedupe by id, keep first (defaults sorted later).
235 let mut seen = std::collections::HashSet::new();
236 devices.retain(|d| seen.insert(d.id.clone()));
237 devices.sort_by(|a, b| {
238 rank_audio_device(&a.id, a.is_default).cmp(&rank_audio_device(&b.id, b.is_default))
239 });
240 let _ = inputs;
241 devices
242}
243
244/// List microphones.
245pub fn list_microphones() -> Vec<MediaDevice> {
246 #[cfg(not(target_os = "android"))]
247 {
248 map_audio_devices(AudioBackend::list_inputs(), true)
249 }
250 #[cfg(target_os = "android")]
251 {
252 Vec::new()
253 }
254}
255
256/// List speakers / output devices.
257pub fn list_speakers() -> Vec<MediaDevice> {
258 #[cfg(not(target_os = "android"))]
259 {
260 map_audio_devices(AudioBackend::list_outputs(), false)
261 }
262 #[cfg(target_os = "android")]
263 {
264 Vec::new()
265 }
266}
267
268#[cfg(not(target_os = "android"))]
269fn resolve_audio_device_id(name: Option<&str>, inputs: bool) -> Option<DeviceId> {
270 let name = name.filter(|s| !s.is_empty())?;
271 // ALSA virtual / exclusive PCMs (incl. bare "pipewire") → system default so
272 // moq-media uses the native PipeWire host default (same as the browser).
273 if is_alsa_virtual_pcm(name) {
274 log::info!(
275 "av-media: preferred audio {name:?} is an ALSA bridge/virtual PCM; \
276 using system default (PipeWire when available)"
277 );
278 return None;
279 }
280 let list = if inputs {
281 AudioBackend::list_inputs()
282 } else {
283 AudioBackend::list_outputs()
284 };
285 // Exact name first (avoid "default" substring matches), then case-insensitive
286 // contains so "EMEET" matches "EMEET SmartCam C960 Mono". Skip junk/sinks.
287 let name_l = name.to_ascii_lowercase();
288 let usable: Vec<_> = list
289 .into_iter()
290 .filter(|d| !is_alsa_virtual_pcm(&d.name) && !is_junk_capture_name(&d.name))
291 .collect();
292 usable
293 .iter()
294 .find(|d| d.name == name)
295 .or_else(|| {
296 usable
297 .iter()
298 .find(|d| d.name.to_ascii_lowercase() == name_l)
299 })
300 .or_else(|| {
301 usable
302 .iter()
303 .find(|d| d.name.to_ascii_lowercase().contains(&name_l))
304 })
305 .map(|d| d.id.clone())
306}
307
308// ── Config / session ───────────────────────────────────────────────────────
309
310#[derive(Clone)]
311pub struct AvMediaConfig {
312 pub sfu_url: url::Url,
313 pub session_id: String,
314 pub nick: String,
315 pub instance: String,
316 /// Initial mic mute (pre-call preference).
317 pub muted: bool,
318 /// Initial speaker mute — remote playback volume 0 (pre-call preference).
319 pub speaker_muted: bool,
320 /// Initial camera publish when hardware is available.
321 pub camera_enabled: bool,
322 /// Preferred camera id (`CameraInfo.id` / name). `None` = first available.
323 pub camera_id: Option<String>,
324 /// Preferred mic name. `None` = system default.
325 pub mic_id: Option<String>,
326 /// Preferred speaker name. `None` = system default.
327 pub speaker_id: Option<String>,
328}
329
330/// Runtime controls from the UI thread (device switches, mute, camera).
331#[derive(Debug, Clone)]
332pub enum MediaControl {
333 SetMuted(bool),
334 /// Mute / unmute remote audio playback (speaker).
335 SetSpeakerMuted(bool),
336 SetCameraEnabled(bool),
337 /// Re-open camera by id (`None` = default / first).
338 SetCameraDevice(Option<String>),
339 /// Switch mic by display name (`None` = system default).
340 SetMicDevice(Option<String>),
341 /// Switch speaker by display name (`None` = system default).
342 SetSpeakerDevice(Option<String>),
343}
344
345/// Status updates from the media task.
346#[derive(Debug, Clone)]
347pub enum AvMediaUpdate {
348 /// MoQ connected and publishing. `has_camera` is true when a capture
349 /// device is available for this call (opened now, or listable so the
350 /// user can turn the camera on later without holding the device).
351 Live {
352 video: VideoFrameStore,
353 has_camera: bool,
354 /// Live mic level (0..=1) written by the capture path.
355 mic_level: MicLevel,
356 /// True when a real capture device is feeding the Opus track.
357 /// False = listen-only / silence publish (still advertises audio).
358 has_mic: bool,
359 },
360 /// Session ended cleanly (stop / transport closed).
361 Ended,
362 /// Connect or runtime failure.
363 Failed(String),
364}
365
366/// Background handle: drop or send on `stop` to tear down the MoQ session.
367pub struct AvMediaSession {
368 stop: Option<oneshot::Sender<()>>,
369 control: Option<mpsc::UnboundedSender<MediaControl>>,
370 pub muted: Arc<AtomicBool>,
371 pub speaker_muted: Arc<AtomicBool>,
372 pub camera_enabled: Arc<AtomicBool>,
373 pub video: VideoFrameStore,
374 pub mic_level: MicLevel,
375 task: Option<tokio::task::JoinHandle<()>>,
376 abort: Option<tokio::task::AbortHandle>,
377}
378
379impl AvMediaSession {
380 pub fn start<F>(config: AvMediaConfig, on_status: F) -> Self
381 where
382 F: Fn(AvMediaUpdate) + Send + Sync + 'static,
383 {
384 let muted = Arc::new(AtomicBool::new(config.muted));
385 let speaker_muted = Arc::new(AtomicBool::new(config.speaker_muted));
386 // Respect pre-call camera pref; has_camera is reported after open.
387 let camera_enabled = Arc::new(AtomicBool::new(config.camera_enabled));
388 let video = VideoFrameStore::new();
389 let mic_level = MicLevel::new();
390 let (stop_tx, stop_rx) = oneshot::channel();
391 let (ctrl_tx, ctrl_rx) = mpsc::unbounded_channel();
392 let muted_task = muted.clone();
393 let camera_task = camera_enabled.clone();
394 let video_task = video.clone();
395 let mic_level_task = mic_level.clone();
396 let on_status = Arc::new(on_status);
397 let task = tokio::spawn(async move {
398 match run_media(
399 config,
400 muted_task,
401 camera_task,
402 video_task,
403 mic_level_task,
404 stop_rx,
405 ctrl_rx,
406 on_status.clone(),
407 )
408 .await
409 {
410 Ok(()) => on_status(AvMediaUpdate::Ended),
411 Err(e) => on_status(AvMediaUpdate::Failed(e.to_string())),
412 }
413 });
414 let abort = task.abort_handle();
415 Self {
416 stop: Some(stop_tx),
417 control: Some(ctrl_tx),
418 muted,
419 speaker_muted,
420 camera_enabled,
421 video,
422 mic_level,
423 task: Some(task),
424 abort: Some(abort),
425 }
426 }
427
428 pub fn set_muted(&self, muted: bool) {
429 self.muted.store(muted, Ordering::Relaxed);
430 if let Some(tx) = &self.control {
431 let _ = tx.send(MediaControl::SetMuted(muted));
432 }
433 }
434
435 pub fn set_speaker_muted(&self, muted: bool) {
436 self.speaker_muted.store(muted, Ordering::Relaxed);
437 if let Some(tx) = &self.control {
438 let _ = tx.send(MediaControl::SetSpeakerMuted(muted));
439 }
440 }
441
442 pub fn set_camera_enabled(&self, enabled: bool) {
443 self.camera_enabled.store(enabled, Ordering::Relaxed);
444 if let Some(tx) = &self.control {
445 let _ = tx.send(MediaControl::SetCameraEnabled(enabled));
446 }
447 }
448
449 pub fn set_camera_device(&self, id: Option<String>) {
450 if let Some(tx) = &self.control {
451 let _ = tx.send(MediaControl::SetCameraDevice(id));
452 }
453 }
454
455 pub fn set_mic_device(&self, id: Option<String>) {
456 if let Some(tx) = &self.control {
457 let _ = tx.send(MediaControl::SetMicDevice(id));
458 }
459 }
460
461 pub fn set_speaker_device(&self, id: Option<String>) {
462 if let Some(tx) = &self.control {
463 let _ = tx.send(MediaControl::SetSpeakerDevice(id));
464 }
465 }
466
467 /// Request a clean stop (unpublish MoQ, drop audio devices). Prefer
468 /// [`Self::stop_and_wait`] from async code so the task can finish teardown.
469 pub fn request_stop(&mut self) {
470 if let Some(tx) = self.stop.take() {
471 let _ = tx.send(());
472 }
473 self.control.take();
474 }
475
476 /// Soft-stop then wait up to `timeout` for the media task to exit.
477 /// Hard-aborts only if the task is still stuck — immediate abort skips
478 /// MoQ unpublish and leaves **zombie broadcasts** on the SFU that peers
479 /// may subscribe to (they see you online but hear silence).
480 pub async fn stop_and_wait(&mut self, timeout: std::time::Duration) {
481 self.request_stop();
482 if let Some(task) = self.task.take() {
483 match tokio::time::timeout(timeout, task).await {
484 Ok(Ok(())) => log::info!("av-media: session task exited cleanly"),
485 Ok(Err(e)) if e.is_cancelled() => {
486 log::debug!("av-media: session task cancelled");
487 }
488 Ok(Err(e)) => log::warn!("av-media: session task join: {e}"),
489 Err(_) => {
490 log::warn!(
491 "av-media: session task still running after {timeout:?}; aborting \
492 (may leave a brief SFU ghost)"
493 );
494 if let Some(a) = self.abort.take() {
495 a.abort();
496 }
497 }
498 }
499 }
500 self.abort.take();
501 // Do not clear `video` here — the UI may still be painting the last
502 // frames from this Arc while MoQ re-dials. Call end uses clear_av_media.
503 self.mic_level.clear();
504 }
505
506 /// Sync stop for Drop / non-async callers. Soft-stop + abort (best-effort).
507 pub fn stop(&mut self) {
508 self.request_stop();
509 if let Some(task) = self.task.take() {
510 task.abort();
511 }
512 self.abort.take();
513 // Keep last frames for reconnect UI; see stop_and_wait.
514 self.mic_level.clear();
515 }
516}
517
518impl Drop for AvMediaSession {
519 fn drop(&mut self) {
520 self.stop();
521 }
522}
523
524async fn run_media(
525 config: AvMediaConfig,
526 muted: Arc<AtomicBool>,
527 camera_enabled: Arc<AtomicBool>,
528 video_store: VideoFrameStore,
529 mic_level: MicLevel,
530 mut stop: oneshot::Receiver<()>,
531 mut control: mpsc::UnboundedReceiver<MediaControl>,
532 on_status: Arc<dyn Fn(AvMediaUpdate) + Send + Sync>,
533) -> Result<()> {
534 // Watch so each remote audio track can react instantly to speaker mute.
535 let (speaker_mute_tx, speaker_mute_rx) = watch::channel(config.speaker_muted);
536 let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
537
538 let our_broadcast = broadcast_path(&config.session_id, &config.nick, &config.instance);
539 log::info!(
540 "av-media: dialing {} as {our_broadcast}",
541 config.sfu_url
542 );
543
544 let mut client_config = moq_native::ClientConfig::default();
545 client_config.tls.disable_verify = Some(true);
546 client_config.backend = Some(moq_native::QuicBackend::Noq);
547 let client = client_config.init().context("moq client init")?;
548
549 let broadcast = LocalBroadcast::new();
550
551 #[cfg(not(target_os = "android"))]
552 let audio_backend = {
553 let host = preferred_audio_host();
554 let input_device = resolve_audio_device_id(config.mic_id.as_deref(), true);
555 let output_device = resolve_audio_device_id(config.speaker_id.as_deref(), false);
556 // Log what the OS thinks is available (helps confirm EMEET vs built-in).
557 let inputs = list_microphones();
558 let outputs = list_speakers();
559 log::info!(
560 "av-media: audio host={host:?} mic_pref={:?} → pinned={} | \
561 speaker_pref={:?} → pinned={}",
562 config.mic_id,
563 input_device.is_some(),
564 config.speaker_id,
565 output_device.is_some()
566 );
567 if !inputs.is_empty() {
568 log::info!(
569 "av-media: microphones: {}",
570 inputs
571 .iter()
572 .map(|d| {
573 if d.is_default {
574 format!("*{}", d.id)
575 } else {
576 d.id.clone()
577 }
578 })
579 .collect::<Vec<_>>()
580 .join(", ")
581 );
582 }
583 if !outputs.is_empty() {
584 log::info!(
585 "av-media: speakers: {}",
586 outputs
587 .iter()
588 .map(|d| {
589 if d.is_default {
590 format!("*{}", d.id)
591 } else {
592 d.id.clone()
593 }
594 })
595 .collect::<Vec<_>>()
596 .join(", ")
597 );
598 }
599 let ab = AudioBackend::new(AudioBackendOpts {
600 host,
601 input_device,
602 output_device,
603 ..Default::default()
604 });
605 ab.set_aec_enabled(false);
606 ab
607 };
608 #[cfg(target_os = "android")]
609 let audio_backend = {
610 let ab = AudioBackend::default();
611 ab.set_aec_enabled(false);
612 ab
613 };
614
615 // Always publish an Opus track so peers see audio in the catalog.
616 // Real mic when available; otherwise silence (listen-only). Prefer
617 // falling back to the system default when a preferred device fails —
618 // never leave the broadcast without audio (that looks like "not sending").
619 log::info!(
620 "av-media: outbound muted={} speaker_muted={} (peers hear silence while mic-muted; \
621 we hear silence while speaker-muted)",
622 muted.load(Ordering::Relaxed),
623 *speaker_mute_rx.borrow()
624 );
625 let has_mic = match open_microphone(&audio_backend).await {
626 Ok(mut mic) => {
627 // Warm-up: wait until the capture ring actually has energy so we
628 // don't advertise "mic open" while still InputNotReady→silence.
629 let peak = warm_up_mic(&mut *mic, std::time::Duration::from_millis(800));
630 log::info!(
631 "av-media: mic warm-up peak={peak:.4} (0 = capture silent / not ready)"
632 );
633 if peak < 1e-5 {
634 log::warn!(
635 "av-media: microphone opened but capture is silent so far — \
636 check PipeWire default source (EMEET), mute, and that OBS isn't \
637 exclusive; peers will hear silence until samples flow"
638 );
639 }
640 let muteable = MuteableSource {
641 inner: mic,
642 muted: muted.clone(),
643 level: mic_level.clone(),
644 pulls: 0,
645 voiced: 0,
646 gain: 1.0,
647 smooth_peak: 0.0,
648 };
649 match broadcast
650 .audio()
651 .set(muteable, AudioCodec::Opus, [AudioPreset::Hq])
652 {
653 Ok(()) => {
654 log::info!("av-media: microphone open, publishing Opus");
655 true
656 }
657 Err(e) => {
658 log::warn!("av-media: set mic audio source failed: {e}; publishing silence");
659 muted.store(true, Ordering::Relaxed);
660 if let Err(e2) = broadcast.audio().set(
661 MuteableSource::silence(muted.clone(), mic_level.clone()),
662 AudioCodec::Opus,
663 [AudioPreset::Hq],
664 ) {
665 log::error!("av-media: silence audio set also failed: {e2}");
666 }
667 false
668 }
669 }
670 }
671 Err(e) => {
672 log::warn!(
673 "av-media: no microphone ({e}); publishing silence (listen-only)"
674 );
675 muted.store(true, Ordering::Relaxed);
676 if let Err(e2) = broadcast.audio().set(
677 MuteableSource::silence(muted.clone(), mic_level.clone()),
678 AudioCodec::Opus,
679 [AudioPreset::Hq],
680 ) {
681 log::error!("av-media: silence audio set failed: {e2}");
682 }
683 false
684 }
685 };
686 if !has_mic {
687 log::info!("av-media: outbound audio is silence (listen-only / no capture)");
688 }
689
690 // Local-preview frame counter (diagnostics).
691 let local_frame_count = Arc::new(AtomicU64::new(0));
692
693 // Camera: desktop V4L2 / platform capture; Android Camera2 → NV12 push source.
694 // Only claim hardware while publish is on — soft-gating alone keeps STREAMON /
695 // Camera2 repeating and lights the privacy LED even when the UI says off.
696 let mut preferred_camera_id = config.camera_id.clone();
697 let devices_present = camera_devices_present();
698 log::info!(
699 "av-media: camera_enabled={} preferred_id={:?} devices_present={devices_present}",
700 camera_enabled.load(Ordering::Relaxed),
701 preferred_camera_id
702 );
703 #[cfg(target_os = "android")]
704 let mut android_camera_guard: Option<crate::android_camera::CameraCaptureGuard> = None;
705
706 let mut preview_keepalive = None;
707 let mut preview_pump: Option<std::thread::JoinHandle<()>> = None;
708 let mut camera_open = false;
709
710 let has_camera = if should_claim_camera_hardware(config.camera_enabled) {
711 #[cfg(not(target_os = "android"))]
712 {
713 match attach_desktop_camera(
714 preferred_camera_id.as_deref(),
715 &broadcast,
716 camera_enabled.clone(),
717 video_store.clone(),
718 local_frame_count.clone(),
719 ) {
720 Ok(opened_id) => {
721 if let Some(id) = opened_id {
722 preferred_camera_id = Some(id);
723 }
724 camera_enabled.store(true, Ordering::Relaxed);
725 preview_keepalive = hold_preview_keepalive(&broadcast);
726 preview_pump = spawn_local_preview_pump(
727 &broadcast,
728 video_store.clone(),
729 camera_enabled.clone(),
730 );
731 camera_open = true;
732 true
733 }
734 Err(e) => {
735 log::warn!("av-media: no camera ({e}); audio-only");
736 camera_enabled.store(false, Ordering::Relaxed);
737 // Keep the in-call toggle when devices still enumerate
738 // (busy / transient open failure).
739 camera_devices_present()
740 }
741 }
742 }
743 #[cfg(target_os = "android")]
744 {
745 match open_android_camera(
746 preferred_camera_id.as_deref(),
747 &broadcast,
748 camera_enabled.clone(),
749 video_store.clone(),
750 local_frame_count.clone(),
751 true,
752 ) {
753 Ok(guard) => {
754 android_camera_guard = Some(guard);
755 preview_keepalive = hold_preview_keepalive(&broadcast);
756 preview_pump = spawn_local_preview_pump(
757 &broadcast,
758 video_store.clone(),
759 camera_enabled.clone(),
760 );
761 camera_open = true;
762 true
763 }
764 Err(e) => {
765 log::warn!("av-media: Android camera unavailable ({e}); audio-only");
766 camera_enabled.store(false, Ordering::Relaxed);
767 // Permission race / busy: still report devices so the
768 // in-call toggle can retry after CAMERA is granted.
769 camera_devices_present()
770 }
771 }
772 }
773 } else {
774 // Pref off: leave the device closed so the privacy light stays dark.
775 // Still report availability when cameras enumerate so the in-call
776 // toggle remains usable.
777 camera_enabled.store(false, Ordering::Relaxed);
778 if devices_present {
779 log::info!(
780 "av-media: camera present but publish off — device left closed \
781 (privacy light off until camera is turned on)"
782 );
783 } else {
784 log::info!("av-media: no cameras listed; audio-only");
785 }
786 devices_present
787 };
788 let mut has_camera = has_camera;
789 let _ = &mut preview_keepalive;
790
791 let origin = moq_lite::Origin::produce();
792 origin.publish_broadcast(&our_broadcast, broadcast.consume());
793
794 let sub_origin = moq_lite::Origin::produce();
795 let mut sub_consumer = sub_origin.consume();
796
797 let session_handle = client
798 .with_publish(origin.consume())
799 .with_consume(sub_origin)
800 .connect(config.sfu_url.clone())
801 .await
802 .context("MoQ connect")?;
803
804 log::info!(
805 "av-media: MoQ connected, publishing {our_broadcast} has_mic={has_mic} has_camera={has_camera}"
806 );
807 on_status(AvMediaUpdate::Live {
808 video: video_store.clone(),
809 has_camera,
810 mic_level: mic_level.clone(),
811 has_mic,
812 });
813
814 let audio_for_playback = audio_backend.clone();
815 let session_id = config.session_id.clone();
816 let my_nick = config.nick.clone();
817 let our_name = our_broadcast.clone();
818 let store_for_subs = video_store.clone();
819 // JoinSet so leave/stop aborts every remote tap (orphan spawns used to leak
820 // AudioBackend + PipeWire streams across calls).
821 let mut taps: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
822 let mut tap_keys: std::collections::HashMap<String, tokio::task::AbortHandle> =
823 std::collections::HashMap::new();
824
825 // Keep broadcast + audio_backend alive for mid-call device switches.
826 let _broadcast = broadcast;
827 let audio_backend_ctrl = audio_backend;
828
829 loop {
830 tokio::select! {
831 res = session_handle.closed() => {
832 if let Err(e) = res {
833 log::info!("av-media: session closed: {e}");
834 } else {
835 log::info!("av-media: session closed cleanly");
836 }
837 break;
838 }
839 _ = &mut stop => {
840 log::info!("av-media: stop requested");
841 break;
842 }
843 announce = sub_consumer.announced() => {
844 let Some((path, announce)) = announce else {
845 log::info!("av-media: announce stream ended");
846 break;
847 };
848 match announce {
849 Some(broadcast_consumer) => {
850 let path_str = path.to_string();
851 if !should_tap(&path_str, &session_id, &our_name, &my_nick) {
852 continue;
853 }
854 // Replace any prior tap for this path.
855 if let Some(h) = tap_keys.remove(&path_str) {
856 h.abort();
857 }
858 log::info!("av-media: + remote {path_str}");
859 let ab = audio_for_playback.clone();
860 let ps = path_str.clone();
861 let store = store_for_subs.clone();
862 let key = path_key(&path_str).to_string();
863 let spk_mute = speaker_mute_rx.clone();
864 let handle = taps.spawn(async move {
865 tap_remote(ps, key, broadcast_consumer, ab, store, spk_mute).await;
866 });
867 tap_keys.insert(path_str, handle);
868 }
869 None => {
870 let path_str = path.to_string();
871 // Keep last frame visible — SFU unannounce/reannounce
872 // blips used to clear the tile and flash black. Stale
873 // frames are wiped on call end via clear_av_media.
874 if let Some(h) = tap_keys.remove(&path_str) {
875 h.abort();
876 }
877 log::info!("av-media: - remote {path_str}");
878 }
879 }
880 }
881 // Reap finished taps so JoinSet doesn't grow forever.
882 Some(res) = taps.join_next() => {
883 if let Err(e) = res {
884 if !e.is_cancelled() {
885 log::debug!("av-media: tap task ended: {e}");
886 }
887 }
888 }
889 msg = control.recv() => {
890 let Some(msg) = msg else { break };
891 match msg {
892 MediaControl::SetMuted(m) => {
893 muted.store(m, Ordering::Relaxed);
894 log::info!("av-media: muted={m}");
895 }
896 MediaControl::SetSpeakerMuted(m) => {
897 if speaker_mute_tx.send(m).is_ok() {
898 log::info!("av-media: speaker_muted={m}");
899 }
900 }
901 MediaControl::SetCameraEnabled(en) => {
902 camera_enabled.store(en, Ordering::Relaxed);
903 if en {
904 if camera_open {
905 log::info!("av-media: camera already open (publish on)");
906 } else {
907 #[cfg(not(target_os = "android"))]
908 {
909 match attach_desktop_camera(
910 preferred_camera_id.as_deref(),
911 &_broadcast,
912 camera_enabled.clone(),
913 video_store.clone(),
914 local_frame_count.clone(),
915 ) {
916 Ok(opened_id) => {
917 if let Some(id) = opened_id {
918 preferred_camera_id = Some(id);
919 }
920 preview_keepalive = hold_preview_keepalive(&_broadcast);
921 preview_pump = spawn_local_preview_pump(
922 &_broadcast,
923 video_store.clone(),
924 camera_enabled.clone(),
925 );
926 camera_open = true;
927 has_camera = true;
928 on_status(AvMediaUpdate::Live {
929 video: video_store.clone(),
930 has_camera: true,
931 mic_level: mic_level.clone(),
932 has_mic,
933 });
934 }
935 Err(e) => {
936 log::warn!("av-media: enable camera failed: {e}");
937 camera_enabled.store(false, Ordering::Relaxed);
938 has_camera = camera_devices_present();
939 on_status(AvMediaUpdate::Live {
940 video: video_store.clone(),
941 has_camera,
942 mic_level: mic_level.clone(),
943 has_mic,
944 });
945 }
946 }
947 }
948 #[cfg(target_os = "android")]
949 {
950 match open_android_camera(
951 preferred_camera_id.as_deref(),
952 &_broadcast,
953 camera_enabled.clone(),
954 video_store.clone(),
955 local_frame_count.clone(),
956 true,
957 ) {
958 Ok(guard) => {
959 android_camera_guard = Some(guard);
960 preview_keepalive = hold_preview_keepalive(&_broadcast);
961 preview_pump = spawn_local_preview_pump(
962 &_broadcast,
963 video_store.clone(),
964 camera_enabled.clone(),
965 );
966 camera_open = true;
967 has_camera = true;
968 on_status(AvMediaUpdate::Live {
969 video: video_store.clone(),
970 has_camera: true,
971 mic_level: mic_level.clone(),
972 has_mic,
973 });
974 }
975 Err(e) => {
976 log::warn!(
977 "av-media: enable Android camera failed: {e}"
978 );
979 camera_enabled.store(false, Ordering::Relaxed);
980 has_camera = camera_devices_present();
981 on_status(AvMediaUpdate::Live {
982 video: video_store.clone(),
983 has_camera,
984 mic_level: mic_level.clone(),
985 has_mic,
986 });
987 }
988 }
989 }
990 }
991 } else if camera_open {
992 release_local_camera(
993 &_broadcast,
994 &mut preview_keepalive,
995 &mut preview_pump,
996 &video_store,
997 #[cfg(target_os = "android")]
998 &mut android_camera_guard,
999 );
1000 camera_open = false;
1001 local_frame_count.store(0, Ordering::Relaxed);
1002 // Keep the toggle: device is available, just not held.
1003 has_camera = camera_devices_present() || has_camera;
1004 log::info!(
1005 "av-media: camera released (publish off; privacy light off)"
1006 );
1007 on_status(AvMediaUpdate::Live {
1008 video: video_store.clone(),
1009 has_camera,
1010 mic_level: mic_level.clone(),
1011 has_mic,
1012 });
1013 } else {
1014 video_store.remove(LOCAL_PREVIEW_KEY);
1015 }
1016 }
1017 MediaControl::SetMicDevice(name) => {
1018 #[cfg(not(target_os = "android"))]
1019 {
1020 let id = resolve_audio_device_id(name.as_deref(), true);
1021 match audio_backend_ctrl.switch_input(id).await {
1022 Ok(()) => log::info!("av-media: mic switched to {name:?}"),
1023 Err(e) => log::warn!("av-media: switch mic failed: {e}"),
1024 }
1025 }
1026 #[cfg(target_os = "android")]
1027 {
1028 let _ = name;
1029 }
1030 }
1031 MediaControl::SetSpeakerDevice(name) => {
1032 #[cfg(not(target_os = "android"))]
1033 {
1034 let id = resolve_audio_device_id(name.as_deref(), false);
1035 match audio_backend_ctrl.switch_output(id).await {
1036 Ok(()) => log::info!("av-media: speaker switched to {name:?}"),
1037 Err(e) => log::warn!("av-media: switch speaker failed: {e}"),
1038 }
1039 }
1040 #[cfg(target_os = "android")]
1041 {
1042 let _ = name;
1043 }
1044 }
1045 MediaControl::SetCameraDevice(id) => {
1046 preferred_camera_id = id.clone();
1047 // Camera off: remember preference only — do not open
1048 // (would light the privacy LED while publish is off).
1049 if !camera_enabled.load(Ordering::Relaxed) {
1050 log::info!(
1051 "av-media: camera device preference set to {id:?} \
1052 (device closed; camera off)"
1053 );
1054 has_camera = camera_devices_present() || has_camera;
1055 continue;
1056 }
1057 release_local_camera(
1058 &_broadcast,
1059 &mut preview_keepalive,
1060 &mut preview_pump,
1061 &video_store,
1062 #[cfg(target_os = "android")]
1063 &mut android_camera_guard,
1064 );
1065 camera_open = false;
1066 local_frame_count.store(0, Ordering::Relaxed);
1067 #[cfg(not(target_os = "android"))]
1068 {
1069 match attach_desktop_camera(
1070 preferred_camera_id.as_deref(),
1071 &_broadcast,
1072 camera_enabled.clone(),
1073 video_store.clone(),
1074 local_frame_count.clone(),
1075 ) {
1076 Ok(opened_id) => {
1077 if let Some(oid) = opened_id {
1078 preferred_camera_id = Some(oid);
1079 }
1080 preview_keepalive = hold_preview_keepalive(&_broadcast);
1081 preview_pump = spawn_local_preview_pump(
1082 &_broadcast,
1083 video_store.clone(),
1084 camera_enabled.clone(),
1085 );
1086 camera_open = true;
1087 has_camera = true;
1088 log::info!(
1089 "av-media: camera device switched to {id:?}"
1090 );
1091 on_status(AvMediaUpdate::Live {
1092 video: video_store.clone(),
1093 has_camera: true,
1094 mic_level: mic_level.clone(),
1095 has_mic,
1096 });
1097 }
1098 Err(e) => {
1099 log::warn!("av-media: open camera {id:?}: {e}");
1100 camera_enabled.store(false, Ordering::Relaxed);
1101 has_camera = camera_devices_present();
1102 on_status(AvMediaUpdate::Live {
1103 video: video_store.clone(),
1104 has_camera,
1105 mic_level: mic_level.clone(),
1106 has_mic,
1107 });
1108 }
1109 }
1110 }
1111 #[cfg(target_os = "android")]
1112 {
1113 match open_android_camera(
1114 preferred_camera_id.as_deref(),
1115 &_broadcast,
1116 camera_enabled.clone(),
1117 video_store.clone(),
1118 local_frame_count.clone(),
1119 true,
1120 ) {
1121 Ok(guard) => {
1122 android_camera_guard = Some(guard);
1123 preview_keepalive = hold_preview_keepalive(&_broadcast);
1124 preview_pump = spawn_local_preview_pump(
1125 &_broadcast,
1126 video_store.clone(),
1127 camera_enabled.clone(),
1128 );
1129 camera_open = true;
1130 has_camera = true;
1131 log::info!(
1132 "av-media: Android camera switched to {id:?}"
1133 );
1134 on_status(AvMediaUpdate::Live {
1135 video: video_store.clone(),
1136 has_camera: true,
1137 mic_level: mic_level.clone(),
1138 has_mic,
1139 });
1140 }
1141 Err(e) => {
1142 log::warn!("av-media: Android camera switch {id:?}: {e}");
1143 camera_enabled.store(false, Ordering::Relaxed);
1144 has_camera = camera_devices_present();
1145 on_status(AvMediaUpdate::Live {
1146 video: video_store.clone(),
1147 has_camera,
1148 mic_level: mic_level.clone(),
1149 has_mic,
1150 });
1151 }
1152 }
1153 }
1154 }
1155 }
1156 }
1157 }
1158 }
1159
1160 drop(session_handle);
1161 // Tear down every remote tap so AudioBackend / PW streams die with us.
1162 for (_, h) in tap_keys.drain() {
1163 h.abort();
1164 }
1165 taps.abort_all();
1166 while taps.join_next().await.is_some() {}
1167 // Drop preview tracks first so the pump thread sees is_closed and exits.
1168 drop(preview_keepalive);
1169 if let Some(h) = preview_pump.take() {
1170 // Best-effort join; don't block teardown if the pump is wedged.
1171 let _ = h.join();
1172 }
1173 drop(audio_backend_ctrl);
1174 drop(audio_for_playback);
1175 // Keep last frames in the shared store so the call UI can paint stale tiles
1176 // across MoQ transport drops / re-dials. Call end clears via clear_av_media.
1177 mic_level.clear();
1178 // Brief yield so aborted tasks drop cpal/PW resources before the next dial.
1179 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1180 log::info!("av-media: teardown complete for {our_broadcast}");
1181 Ok(())
1182}
1183
1184/// Subscribe audio + video for one remote broadcast until it ends or is aborted.
1185///
1186/// `key` is the frame-store id (`nick` or `nick~instance` from [`path_key`]).
1187///
1188/// Audio and video are independent: a catalog race that delays the audio
1189/// rendition must not tear down video (and vice versa). We wait with
1190/// `audio_ready` / `video_ready` and retry so late-advertised tracks still play.
1191///
1192/// `speaker_mute` drives remote playback volume (`0.0` when muted, `1.0` otherwise).
1193async fn tap_remote(
1194 path: String,
1195 key: String,
1196 broadcast_consumer: moq_lite::BroadcastConsumer,
1197 audio_backend: AudioBackend,
1198 video_store: VideoFrameStore,
1199 speaker_mute: watch::Receiver<bool>,
1200) {
1201 // Match freeq-sdk-ffi: tighter latency than the 150ms streaming default.
1202 let policy = iroh_live::media::playout::PlaybackPolicy::default()
1203 .with_max_latency(std::time::Duration::from_millis(60));
1204 let remote = match RemoteBroadcast::with_playback_policy(
1205 &path,
1206 broadcast_consumer,
1207 policy,
1208 )
1209 .await
1210 {
1211 Ok(r) => r,
1212 Err(e) => {
1213 log::warn!("av-media: catalog {path}: {e}");
1214 return;
1215 }
1216 };
1217
1218 let audio_task = {
1219 let remote = remote.clone();
1220 let ab = audio_backend;
1221 let ps = path.clone();
1222 let mut speaker_mute = speaker_mute;
1223 tokio::spawn(async move {
1224 let mut consecutive_errs = 0u32;
1225 loop {
1226 match remote.audio_ready(&ab).await {
1227 Ok(track) => {
1228 consecutive_errs = 0;
1229 log::info!("av-media: receiving audio from {ps}");
1230 // Apply current speaker mute, then hold until the track
1231 // ends or mute toggles (volume 0 = silence, keeps decode).
1232 apply_speaker_volume(&track, *speaker_mute.borrow());
1233 loop {
1234 tokio::select! {
1235 _ = track.stopped() => {
1236 log::info!("av-media: audio track ended for {ps}");
1237 break;
1238 }
1239 changed = speaker_mute.changed() => {
1240 if changed.is_err() {
1241 // Session tearing down — keep track until stop.
1242 track.stopped().await;
1243 break;
1244 }
1245 apply_speaker_volume(&track, *speaker_mute.borrow());
1246 }
1247 }
1248 }
1249 }
1250 Err(e) => {
1251 consecutive_errs = consecutive_errs.saturating_add(1);
1252 // Keep retrying while the remote broadcast is open — a
1253 // transient audio catalog/transport blip must not kill
1254 // the independent video pipeline (tap_remote waits on
1255 // `remote.closed()`, not this task exiting).
1256 if consecutive_errs <= 3 || consecutive_errs % 10 == 0 {
1257 log::warn!(
1258 "av-media: audio sub {ps}: {e} (retry {consecutive_errs})"
1259 );
1260 }
1261 let backoff_ms = (500u64 * u64::from(consecutive_errs.min(8))).min(4_000);
1262 tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)).await;
1263 }
1264 }
1265 }
1266 })
1267 };
1268
1269 let video_task = {
1270 let remote = remote.clone();
1271 let store = video_store.clone();
1272 let key = key.clone();
1273 let path = path.clone();
1274 tokio::spawn(async move {
1275 loop {
1276 match remote.video_ready().await {
1277 Ok(mut vtrack) => {
1278 log::info!("av-media: receiving video from {path}");
1279 while let Some(frame) = vtrack.next_frame().await {
1280 let (w, h) = (frame.width(), frame.height());
1281 let rgba = frame.rgba_image();
1282 let bytes: Arc<[u8]> = rgba.as_raw().as_slice().into();
1283 store.set(key.clone(), w, h, bytes);
1284 }
1285 log::info!("av-media: video track ended for {path}");
1286 }
1287 Err(e) => {
1288 log::debug!("av-media: video wait {path}: {e}");
1289 tokio::time::sleep(std::time::Duration::from_millis(500)).await;
1290 }
1291 }
1292 }
1293 })
1294 };
1295
1296 // Audio and video are independent: keep both taps alive until the remote
1297 // broadcast catalog entry closes. Do not abort one when the other retries
1298 // or hits a transient transport error — that dropped video while audio
1299 // (and IRC call state) continued.
1300 let close_err = remote.closed().await;
1301 log::info!("av-media: remote closed {path}: {close_err}");
1302 audio_task.abort();
1303 video_task.abort();
1304 // Frame removal is driven by announce `None` (participant left). Do not
1305 // clear here — tap replacement would flash black until the next frame.
1306}
1307
1308/// Set remote track volume from speaker-mute (`0.0` silence, `1.0` full).
1309fn apply_speaker_volume(track: &iroh_live::media::subscribe::AudioTrack, muted: bool) {
1310 track.set_volume(if muted { 0.0 } else { 1.0 });
1311}
1312
1313/// Open the default mic input, retrying after clearing preferred devices if
1314/// the first open fails.
1315///
1316/// moq-media starts **output and input as a pair**. A bad speaker pref
1317/// (`sysdefault` busy under PipeWire) fails the whole pair, so we must clear
1318/// **output** as well as input before retrying.
1319async fn open_microphone(
1320 audio_backend: &AudioBackend,
1321) -> Result<Box<dyn iroh_live::media::traits::AudioSource>> {
1322 match audio_backend.default_input().await {
1323 Ok(mic) => Ok(Box::new(mic)),
1324 Err(first) => {
1325 log::warn!(
1326 "av-media: default_input failed ({first}); \
1327 retry after clearing preferred I/O devices"
1328 );
1329 #[cfg(not(target_os = "android"))]
1330 {
1331 // Output first: start_cpal_streams requires a working speaker.
1332 if let Err(e) = audio_backend.switch_output(None).await {
1333 log::warn!("av-media: switch_output(None) failed: {e}");
1334 }
1335 if let Err(e) = audio_backend.switch_input(None).await {
1336 log::warn!("av-media: switch_input(None) failed: {e}");
1337 }
1338 }
1339 match audio_backend.default_input().await {
1340 Ok(mic) => Ok(Box::new(mic)),
1341 Err(e) => Err(e),
1342 }
1343 }
1344 }
1345}
1346
1347/// Pull mic frames until we see energy or `budget` elapses. Returns peak abs.
1348fn warm_up_mic(
1349 mic: &mut dyn iroh_live::media::traits::AudioSource,
1350 budget: std::time::Duration,
1351) -> f32 {
1352 let deadline = std::time::Instant::now() + budget;
1353 let mut buf = vec![0.0f32; 960];
1354 let mut peak = 0.0f32;
1355 let mut ready = 0u32;
1356 while std::time::Instant::now() < deadline {
1357 match mic.pop_samples(&mut buf) {
1358 Ok(Some(n)) if n > 0 => {
1359 ready = ready.saturating_add(1);
1360 for &s in &buf[..n] {
1361 peak = peak.max(s.abs());
1362 }
1363 if peak > 1e-3 {
1364 break;
1365 }
1366 }
1367 Ok(_) => {}
1368 Err(e) => {
1369 log::warn!("av-media: mic warm-up read error: {e:#}");
1370 break;
1371 }
1372 }
1373 std::thread::sleep(std::time::Duration::from_millis(10));
1374 }
1375 log::info!("av-media: mic warm-up ready_frames={ready} peak={peak:.4}");
1376 peak
1377}
1378
1379/// Always-ready mono 48 kHz silence — keeps an Opus track advertised when
1380/// no capture device is available (listen-only join).
1381struct SilenceSource {
1382 format: iroh_live::media::format::AudioFormat,
1383}
1384
1385impl Default for SilenceSource {
1386 fn default() -> Self {
1387 Self {
1388 format: iroh_live::media::format::AudioFormat::mono_48k(),
1389 }
1390 }
1391}
1392
1393impl iroh_live::media::traits::AudioSource for SilenceSource {
1394 fn format(&self) -> iroh_live::media::format::AudioFormat {
1395 self.format
1396 }
1397
1398 fn pop_samples(
1399 &mut self,
1400 buf: &mut [f32],
1401 ) -> anyhow::Result<Option<usize>> {
1402 for s in buf.iter_mut() {
1403 *s = 0.0;
1404 }
1405 Ok(Some(buf.len()))
1406 }
1407}
1408
1409#[cfg(not(target_os = "android"))]
1410fn camera_config() -> CameraConfig {
1411 CameraConfig {
1412 selector: CameraSelector::TargetResolution(640, 360),
1413 preferred_format: None,
1414 // CPU RGBA — safer for local preview tee + software H.264.
1415 zero_copy: false,
1416 }
1417}
1418
1419/// Join-time policy: only claim capture hardware when publish is on.
1420/// (Soft-gating an already-open device leaves STREAMON / Camera2 live and
1421/// lights the privacy LED while the UI shows camera off.)
1422fn should_claim_camera_hardware(publish_enabled: bool) -> bool {
1423 publish_enabled
1424}
1425
1426/// True when at least one camera enumerates (does not open / stream).
1427fn camera_devices_present() -> bool {
1428 !list_cameras().is_empty()
1429}
1430
1431/// Hold a local preview track so SharedVideoSource stays unparked for self-view
1432/// even with no remote H.264 subscriber.
1433fn hold_preview_keepalive(
1434 broadcast: &LocalBroadcast,
1435) -> Option<iroh_live::media::subscribe::VideoTrack> {
1436 let t = broadcast.preview();
1437 if t.is_some() {
1438 log::info!("av-media: local preview keepalive held (SharedVideoSource unparked)");
1439 } else {
1440 log::warn!("av-media: broadcast.preview() returned None after set_source");
1441 }
1442 t
1443}
1444
1445/// Release capture hardware: drop preview subscribers, clear the MoQ video
1446/// track (stops SharedVideoSource → streamoff / close), and stop Camera2.
1447fn release_local_camera(
1448 broadcast: &LocalBroadcast,
1449 preview_keepalive: &mut Option<iroh_live::media::subscribe::VideoTrack>,
1450 preview_pump: &mut Option<std::thread::JoinHandle<()>>,
1451 video_store: &VideoFrameStore,
1452 #[cfg(target_os = "android")] android_guard: &mut Option<
1453 crate::android_camera::CameraCaptureGuard,
1454 >,
1455) {
1456 *preview_keepalive = None;
1457 // Pump exits when publish is off (or the preview track closes). Join briefly
1458 // so we don't leave a detached thread holding a VideoTrack across clear().
1459 if let Some(h) = preview_pump.take() {
1460 let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500);
1461 loop {
1462 if h.is_finished() {
1463 let _ = h.join();
1464 break;
1465 }
1466 if std::time::Instant::now() >= deadline {
1467 log::warn!("av-media: local preview pump did not exit in 500ms; detaching");
1468 break;
1469 }
1470 std::thread::sleep(std::time::Duration::from_millis(10));
1471 }
1472 }
1473 broadcast.video().clear();
1474 #[cfg(target_os = "android")]
1475 {
1476 drop(android_guard.take());
1477 }
1478 video_store.remove(LOCAL_PREVIEW_KEY);
1479}
1480
1481/// Open a desktop camera and attach it as the broadcast H.264 source.
1482/// Returns the opened device id when known.
1483#[cfg(not(target_os = "android"))]
1484fn attach_desktop_camera(
1485 preferred: Option<&str>,
1486 broadcast: &LocalBroadcast,
1487 camera_enabled: Arc<AtomicBool>,
1488 video_store: VideoFrameStore,
1489 local_frame_count: Arc<AtomicU64>,
1490) -> Result<Option<String>> {
1491 let (cam, opened_id) = open_camera_with_fallback(preferred)?;
1492 let cam_name = cam.name().to_string();
1493 let gated = GatedCameraSource {
1494 inner: cam,
1495 enabled: camera_enabled.clone(),
1496 preview: video_store,
1497 frame_count: local_frame_count,
1498 streaming: false,
1499 };
1500 broadcast
1501 .video()
1502 .set_source(gated, VideoCodec::H264, [VideoPreset::P360])
1503 .context("video set_source")?;
1504 if is_virtual_camera(&cam_name, opened_id.as_deref().unwrap_or("")) {
1505 log::info!(
1506 "av-media: using virtual camera {cam_name} — \
1507 in OBS: Controls → Start Virtual Camera \
1508 (otherwise self-view is blank)"
1509 );
1510 }
1511 log::info!(
1512 "av-media: camera open id={opened_id:?} name={cam_name}, \
1513 publishing H.264 360p (publish={})",
1514 camera_enabled.load(Ordering::Relaxed)
1515 );
1516 Ok(opened_id)
1517}
1518
1519/// Open preferred camera, then fall back through the device list (hardware first).
1520///
1521/// Deliberately **does not** start/stop/probe frames here. V4L2 `dqbuf` is
1522/// blocking with no timeout — a probe that hangs leaves the device busy so
1523/// every later open fails (audio-only calls + blank self-view tile). Parent
1524/// behavior was `CameraCapturer::open` only; SharedVideoSource starts streaming
1525/// when the first preview/encoder subscriber arrives.
1526///
1527/// Open preferred camera, then fall back through hardware. Virtual (OBS) is
1528/// only used when the user **explicitly** preferred that id — never as a silent
1529/// fallback when the USB cam is busy (OBS often holds `/dev/video0` while
1530/// exposing `/dev/video10`).
1531#[cfg(not(target_os = "android"))]
1532fn open_camera_with_fallback(
1533 preferred: Option<&str>,
1534) -> Result<(Box<dyn VideoSource>, Option<String>)> {
1535 let config = camera_config();
1536 let preferred = preferred.filter(|s| !s.is_empty());
1537
1538 let listed = match CameraCapturer::list() {
1539 Ok(c) => c,
1540 Err(e) => {
1541 log::warn!("av-media: CameraCapturer::list failed: {e}");
1542 Vec::new()
1543 }
1544 };
1545 log::info!(
1546 "av-media: cameras available: {}",
1547 if listed.is_empty() {
1548 "(none)".into()
1549 } else {
1550 listed
1551 .iter()
1552 .map(|c| {
1553 let v = if is_virtual_camera(&c.name, &c.id) {
1554 " [virtual]"
1555 } else {
1556 ""
1557 };
1558 format!("{} ({}){v}", c.name, c.id)
1559 })
1560 .collect::<Vec<_>>()
1561 .join(", ")
1562 }
1563 );
1564
1565 let is_virt = |id: &str| {
1566 listed
1567 .iter()
1568 .find(|c| c.id == id || c.name == id)
1569 .map(|c| is_virtual_camera(&c.name, &c.id))
1570 .unwrap_or_else(|| is_virtual_camera(id, id))
1571 };
1572
1573 let mut candidates: Vec<Option<String>> = Vec::new();
1574
1575 // Explicit preference first (including virtual if the user picked OBS).
1576 if let Some(id) = preferred {
1577 candidates.push(Some(id.to_string()));
1578 if is_virt(id) {
1579 log::info!(
1580 "av-media: preferred {id} is virtual (OBS) — opening it first. \
1581 USB cam may be busy because OBS is using it as a source."
1582 );
1583 }
1584 }
1585
1586 // Then non-virtual hardware (skip virtuals for auto-pick).
1587 let mut cams = listed.clone();
1588 cams.sort_by(|a, b| {
1589 a.name
1590 .to_lowercase()
1591 .cmp(&b.name.to_lowercase())
1592 });
1593 for c in cams {
1594 if is_virtual_camera(&c.name, &c.id) {
1595 continue;
1596 }
1597 if c.supported_formats.is_empty() {
1598 continue;
1599 }
1600 if candidates.iter().any(|x| {
1601 x.as_deref() == Some(c.id.as_str()) || x.as_deref() == Some(c.name.as_str())
1602 }) {
1603 continue;
1604 }
1605 candidates.push(Some(c.id));
1606 }
1607 if !candidates.iter().any(|c| c.is_none()) {
1608 candidates.push(None);
1609 }
1610
1611 let mut errors: Vec<String> = Vec::new();
1612 for cand in &candidates {
1613 let label = cand.as_deref().unwrap_or("(default)");
1614 match open_camera_with_busy_retry(cand.as_deref(), &config) {
1615 Ok(cam) => {
1616 let name = cam.name().to_string();
1617 // Reject *accidental* virtual from default open when user did not
1618 // prefer virtual.
1619 let user_wants_virtual = preferred.is_some_and(|p| is_virt(p));
1620 if is_virtual_camera(&name, label) && !user_wants_virtual {
1621 log::warn!(
1622 "av-media: rejecting auto-opened virtual camera {name} ({label})"
1623 );
1624 errors.push(format!("{label}: rejected virtual {name}"));
1625 continue;
1626 }
1627 if is_virtual_camera(&name, label) {
1628 log::info!(
1629 "av-media: opened virtual camera {name} — ensure OBS has \
1630 'Start Virtual Camera' enabled or self-view will be blank"
1631 );
1632 }
1633 log::info!("av-media: opened camera id={label:?} name={name}");
1634 return Ok((cam, cand.clone()));
1635 }
1636 Err(e) => {
1637 let msg = format!("{e:#}");
1638 if msg.contains("busy") {
1639 log::warn!(
1640 "av-media: open camera {label}: {msg} \
1641 (often OBS or another app holds the USB cam)"
1642 );
1643 } else {
1644 log::warn!("av-media: open camera {label}: {msg}");
1645 }
1646 errors.push(format!("{label}: {e}"));
1647 }
1648 }
1649 }
1650 Err(anyhow::anyhow!(
1651 "no working camera ({})",
1652 errors.join(" | ")
1653 ))
1654}
1655
1656/// Open one device; retry briefly on "busy" (race with OBS / previous session).
1657///
1658/// v4l2loopback nodes (OBS Virtual Camera) are routed to
1659/// [`crate::v4l2cam::V4l2MmapCapture`]: rusty-capture's dqbuf leaves the
1660/// `memory` field 0, which loopback drivers reject with EINVAL on the first
1661/// frame (silent blank self-view). Hardware cams keep rusty-capture.
1662#[cfg(not(target_os = "android"))]
1663fn open_camera_with_busy_retry(
1664 id: Option<&str>,
1665 config: &CameraConfig,
1666) -> Result<Box<dyn VideoSource>> {
1667 // Loopback devices get the dqbuf-fixed capturer (no busy-retry needed —
1668 // loopback nodes are multi-reader, "busy" doesn't apply the same way).
1669 #[cfg(target_os = "linux")]
1670 {
1671 // Resolve `(default)` to the first listed device so a loopback first
1672 // entry (e.g. OBS-only setups) also lands on the fixed capture path.
1673 let resolved: Option<String> = match id {
1674 Some(p) if p.starts_with("/dev/video") => Some(p.to_string()),
1675 Some(_) => None, // name, not a device path — rusty-capture handles
1676 None => CameraCapturer::list()
1677 .ok()
1678 .and_then(|c| c.into_iter().next())
1679 .map(|c| c.id)
1680 .filter(|p| p.starts_with("/dev/video")),
1681 };
1682 if let Some(path) = resolved {
1683 if crate::v4l2cam::is_loopback_device(&path) {
1684 let (w, h) = match config.selector {
1685 CameraSelector::TargetResolution(w, h) => (w, h),
1686 _ => (640, 360),
1687 };
1688 let cam = crate::v4l2cam::V4l2MmapCapture::open(&path, w, h)
1689 .with_context(|| format!("loopback open {path}"))?;
1690 log::info!("av-media: using dqbuf-fixed capture for loopback {path}");
1691 return Ok(Box::new(cam));
1692 }
1693 }
1694 }
1695
1696 const ATTEMPTS: u32 = 4;
1697 let mut last = None;
1698 for attempt in 0..ATTEMPTS {
1699 match CameraCapturer::open(None, id, config) {
1700 Ok(cam) => return Ok(Box::new(cam)),
1701 Err(e) => {
1702 let busy = e.to_string().to_ascii_lowercase().contains("busy");
1703 last = Some(e);
1704 if busy && attempt + 1 < ATTEMPTS {
1705 std::thread::sleep(std::time::Duration::from_millis(150 * (attempt + 1) as u64));
1706 continue;
1707 }
1708 break;
1709 }
1710 }
1711 }
1712 Err(last.unwrap_or_else(|| anyhow::anyhow!("open failed")))
1713}
1714
1715/// Luma (Rec.601 approx) min/max across RGBA bytes — used by live diagnostics
1716/// and tests to detect a real camera picture vs a blank/uniform buffer.
1717/// Ships with the crate (not test-only) so pump/tee logs report real content.
1718pub(crate) fn rgba_luma_range(rgba: &[u8]) -> (u8, u8) {
1719 let mut min = 255u8;
1720 let mut max = 0u8;
1721 for px in rgba.chunks_exact(4) {
1722 let l = ((77u32 * px[0] as u32 + 150u32 * px[1] as u32 + 29u32 * px[2] as u32) >> 8) as u8;
1723 min = min.min(l);
1724 max = max.max(l);
1725 }
1726 (min, max)
1727}
1728
1729/// Pump `broadcast.preview()` frames into `__local__` for self-view.
1730fn spawn_local_preview_pump(
1731 broadcast: &LocalBroadcast,
1732 store: VideoFrameStore,
1733 enabled: Arc<AtomicBool>,
1734) -> Option<std::thread::JoinHandle<()>> {
1735 let mut track = match broadcast.preview() {
1736 Some(t) => t,
1737 None => {
1738 log::warn!("av-media: preview() for pump returned None");
1739 return None;
1740 }
1741 };
1742 match std::thread::Builder::new()
1743 .name("local-preview-pump".into())
1744 .spawn(move || {
1745 let mut logged = false;
1746 let mut pump_waited_ms = 0u64;
1747 let mut warned_no_frames = false;
1748 loop {
1749 // Exit on mute so we drop the preview VideoTrack and let
1750 // release_local_camera clear the SharedVideoSource promptly.
1751 // Spinning here used to keep a subscriber alive across "camera
1752 // off" and leave V4L2 STREAMON / Camera2 repeating (LED on).
1753 if !enabled.load(Ordering::Relaxed) {
1754 store.remove(LOCAL_PREVIEW_KEY);
1755 log::info!("av-media: local preview pump exit (publish off)");
1756 break;
1757 }
1758 if let Some(frame) = track.try_recv() {
1759 let (w, h) = (frame.width(), frame.height());
1760 match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1761 let rgba = frame.rgba_image();
1762 let bytes: Arc<[u8]> = rgba.as_raw().as_slice().into();
1763 (w, h, bytes)
1764 })) {
1765 Ok((w, h, bytes)) => {
1766 let (r0, g0, b0, a0) = bytes
1767 .get(0..4)
1768 .map(|p| (p[0], p[1], p[2], p[3]))
1769 .unwrap_or((0, 0, 0, 0));
1770 let (luma_min, luma_max) = rgba_luma_range(&bytes);
1771 store.set(LOCAL_PREVIEW_KEY, w, h, bytes);
1772 if !logged {
1773 logged = true;
1774 log::info!(
1775 "av-media: local preview pump first frame {w}x{h} \
1776 rgba0=({r0},{g0},{b0},{a0}) luma={luma_min}..{luma_max}"
1777 );
1778 if luma_max == luma_min {
1779 log::warn!(
1780 "av-media: local preview frame is uniform \
1781 (blank capture?) — OBS Virtual Camera not started?"
1782 );
1783 }
1784 }
1785 }
1786 Err(_) => {
1787 log::warn!(
1788 "av-media: local preview pump rgba_image panicked {w}x{h}"
1789 );
1790 }
1791 }
1792 } else if track.is_closed() {
1793 log::info!("av-media: local preview pump track closed");
1794 break;
1795 } else {
1796 // Warn once if capture is up but never yields a frame
1797 // (dead OBS virtual node, busy device, etc.).
1798 if !logged && pump_waited_ms >= 3_000 && !warned_no_frames {
1799 warned_no_frames = true;
1800 log::warn!(
1801 "av-media: no local preview frames after {}ms — camera capture \
1802 is not producing (OBS Virtual Camera not started, or device busy)",
1803 pump_waited_ms
1804 );
1805 }
1806 pump_waited_ms = pump_waited_ms.saturating_add(10);
1807 std::thread::sleep(std::time::Duration::from_millis(10));
1808 }
1809 }
1810 }) {
1811 Ok(h) => {
1812 log::info!("av-media: local preview pump thread started");
1813 Some(h)
1814 }
1815 Err(e) => {
1816 log::warn!("av-media: local preview pump spawn failed: {e}");
1817 None
1818 }
1819 }
1820}
1821
1822/// Mid-call device switch uses the same fallback path.
1823/// Start Android Camera2, register a gated NV12 [`VideoSource`], and return a
1824/// guard that stops the camera when dropped (call end or camera toggled off).
1825#[cfg(target_os = "android")]
1826fn open_android_camera(
1827 camera_id: Option<&str>,
1828 broadcast: &LocalBroadcast,
1829 camera_enabled: Arc<AtomicBool>,
1830 video_store: VideoFrameStore,
1831 local_frame_count: Arc<AtomicU64>,
1832 want_publish: bool,
1833) -> Result<crate::android_camera::CameraCaptureGuard> {
1834 crate::android_camera::start_capture(camera_id)
1835 .context("CameraCapture.start")?;
1836 // Camera2 open + session configure is async on a Java handler thread.
1837 if let Err(e) = crate::android_camera::wait_until_opened(std::time::Duration::from_secs(5)) {
1838 crate::android_camera::stop_capture();
1839 return Err(e).context("Camera2 session");
1840 }
1841
1842 let label = camera_id
1843 .filter(|s| !s.is_empty())
1844 .unwrap_or("front")
1845 .to_string();
1846 let cam = crate::android_camera::PushCameraSource::new(format!("android-camera:{label}"));
1847 let gated = GatedCameraSource {
1848 inner: Box::new(cam),
1849 enabled: camera_enabled.clone(),
1850 preview: video_store,
1851 frame_count: local_frame_count,
1852 streaming: false,
1853 };
1854 if let Err(e) = broadcast
1855 .video()
1856 .set_source(gated, VideoCodec::H264, [VideoPreset::P360])
1857 {
1858 // Camera2 is live but MoQ source failed — release hardware.
1859 crate::android_camera::stop_capture();
1860 return Err(e).context("video set_source");
1861 }
1862
1863 if want_publish {
1864 camera_enabled.store(true, Ordering::Relaxed);
1865 }
1866 log::info!(
1867 "av-media: Android camera open id={camera_id:?}, publishing H.264 360p (publish={})",
1868 camera_enabled.load(Ordering::Relaxed)
1869 );
1870 Ok(crate::android_camera::CameraCaptureGuard)
1871}
1872
1873/// Wraps camera capture while hardware is held. Publish/self-view are gated by
1874/// `enabled`. When publish flips off we **stop the inner capturer immediately**
1875/// (V4L2 `STREAMOFF` / close) so the privacy LED goes dark without waiting for
1876/// the media task's `release_local_camera` clear — the old gate still called
1877/// `inner.pop_frame()` first, which kept STREAMON alive while discarding frames.
1878/// `release_local_camera` remains the steady-state teardown (drop source +
1879/// Camera2 guard).
1880///
1881/// Always tees enabled frames into `__local__` for self-view (requires a
1882/// SharedVideoSource subscriber — we hold `broadcast.preview()` as keepalive).
1883///
1884/// `inner` is boxed so v4l2loopback devices (OBS Virtual Camera) can use the
1885/// loopback-safe [`crate::v4l2cam::V4l2MmapCapture`] instead of rusty-capture's
1886/// `CameraCapturer` (whose v4l2r dqbuf leaves `memory=0` → EINVAL on loopback).
1887/// On Android, `inner` is the Camera2 [`PushCameraSource`].
1888struct GatedCameraSource {
1889 inner: Box<dyn VideoSource>,
1890 enabled: Arc<AtomicBool>,
1891 preview: VideoFrameStore,
1892 frame_count: Arc<AtomicU64>,
1893 /// Whether `inner.start()` is live. Cleared on gate-off / `stop()` so mute
1894 /// can streamoff without waiting for SharedVideoSource teardown.
1895 streaming: bool,
1896}
1897
1898impl VideoSource for GatedCameraSource {
1899 fn name(&self) -> &str {
1900 self.inner.name()
1901 }
1902
1903 fn format(&self) -> iroh_live::media::format::VideoFormat {
1904 self.inner.format()
1905 }
1906
1907 fn start(&mut self) -> anyhow::Result<()> {
1908 let r = self.inner.start();
1909 if r.is_ok() {
1910 self.streaming = true;
1911 }
1912 r
1913 }
1914
1915 fn stop(&mut self) -> anyhow::Result<()> {
1916 self.streaming = false;
1917 self.inner.stop()
1918 }
1919
1920 fn pop_frame(
1921 &mut self,
1922 ) -> anyhow::Result<Option<iroh_live::media::format::VideoFrame>> {
1923 // Check publish flag *before* capturing. The previous order (dqbuf then
1924 // discard) left V4L2 STREAMON / the privacy LED on for the whole mute.
1925 if !self.enabled.load(Ordering::Relaxed) {
1926 if self.streaming {
1927 if let Err(e) = self.inner.stop() {
1928 log::warn!(
1929 "av-media: camera {} stop on mute failed: {e:#}",
1930 self.inner.name()
1931 );
1932 } else {
1933 log::info!(
1934 "av-media: camera {} hardware stopped (publish off; privacy light off)",
1935 self.inner.name()
1936 );
1937 }
1938 self.streaming = false;
1939 // Android PushCameraSource::stop only clears the frame cell —
1940 // Camera2 stays open until stop_capture / guard drop.
1941 #[cfg(target_os = "android")]
1942 crate::android_camera::stop_capture();
1943 }
1944 self.preview.remove(LOCAL_PREVIEW_KEY);
1945 // SharedVideoSource spins on Ok(None); back off while gated.
1946 std::thread::sleep(std::time::Duration::from_millis(20));
1947 return Ok(None);
1948 }
1949 if !self.streaming {
1950 self.inner.start().map_err(|e| {
1951 log::warn!(
1952 "av-media: camera {} restart after mute failed: {e:#}",
1953 self.inner.name()
1954 );
1955 e
1956 })?;
1957 self.streaming = true;
1958 // Android Camera2 is restarted by open_android_camera after
1959 // release_local_camera clears the guard — not from this gate.
1960 }
1961 let frame = match self.inner.pop_frame() {
1962 Ok(f) => f,
1963 Err(e) => {
1964 // Surface real capture failures (e.g. OBS Virtual Camera not
1965 // started → v4l2loopback dqbuf EINVAL; busy USB cam → EBUSY).
1966 // Upstream SharedVideoSource stops the capture thread silently
1967 // on Err — that left users staring at a blank tile with no log.
1968 let msg = format!("{e:#}");
1969 let already = self.frame_count.load(Ordering::Relaxed) > 0;
1970 let key = if msg.contains("EINVAL") {
1971 "virtual camera not producing (start OBS Virtual Camera output?)"
1972 } else if msg.contains("EBUSY") || msg.contains("busy") {
1973 "camera busy (held by OBS or another app?)"
1974 } else {
1975 "capture error"
1976 };
1977 log::warn!(
1978 "av-media: camera {} pop_frame failed: {msg} — {key} {}",
1979 self.inner.name(),
1980 if already {
1981 "(frames had been flowing)"
1982 } else {
1983 "(no frames ever captured)"
1984 }
1985 );
1986 self.preview.remove(LOCAL_PREVIEW_KEY);
1987 self.streaming = false;
1988 return Err(e);
1989 }
1990 };
1991 if let Some(ref f) = frame {
1992 let (w, h) = (f.width(), f.height());
1993 // Best-effort local preview — never fail the encode path.
1994 match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1995 let rgba = f.rgba_image();
1996 let bytes: Arc<[u8]> = rgba.as_raw().as_slice().into();
1997 (w, h, bytes)
1998 })) {
1999 Ok((w, h, bytes)) => {
2000 // Sample first pixel + luma range for blank/alpha diagnostics.
2001 let (r0, g0, b0, a0) = bytes
2002 .get(0..4)
2003 .map(|p| (p[0], p[1], p[2], p[3]))
2004 .unwrap_or((0, 0, 0, 0));
2005 let (luma_min, luma_max) = rgba_luma_range(&bytes);
2006 self.preview.set(LOCAL_PREVIEW_KEY, w, h, bytes);
2007 let n = self.frame_count.fetch_add(1, Ordering::Relaxed);
2008 if n == 0 {
2009 log::info!(
2010 "av-media: first published/preview frame {w}x{h} \
2011 rgba0=({r0},{g0},{b0},{a0}) luma={luma_min}..{luma_max} \
2012 (outbound video live)"
2013 );
2014 if luma_max == luma_min {
2015 log::warn!(
2016 "av-media: published frame is uniform (blank capture?) — \
2017 OBS Virtual Camera not started?"
2018 );
2019 }
2020 } else if n == 30 || n == 300 || n == 3000 {
2021 log::info!(
2022 "av-media: published frames={n} ({w}x{h}) \
2023 rgba0=({r0},{g0},{b0},{a0}) luma={luma_min}..{luma_max}"
2024 );
2025 }
2026 }
2027 Err(_) => {
2028 log::warn!("av-media: rgba_image panicked on local frame {w}x{h}");
2029 // Still count as a publishable frame even if preview tee failed.
2030 let n = self.frame_count.fetch_add(1, Ordering::Relaxed);
2031 if n == 0 {
2032 log::info!("av-media: first publish frame {w}x{h} (preview tee failed)");
2033 }
2034 }
2035 }
2036 }
2037 Ok(frame)
2038 }
2039}
2040
2041/// RMS of a PCM buffer — used by tests and as a simple energy metric.
2042pub fn pcm_rms(samples: &[f32]) -> f32 {
2043 if samples.is_empty() {
2044 return 0.0;
2045 }
2046 let sum_sq: f32 = samples.iter().map(|s| s * s).sum();
2047 (sum_sq / samples.len() as f32).sqrt()
2048}
2049
2050/// Continuous 48 kHz mono sine — freeq mesh rate (see freeq-av `SPEAK_RATE`).
2051///
2052/// Used by interop tests as a deterministic non-silent capture substitute so
2053/// we prove Opus encode/decode energy without requiring a real microphone.
2054#[cfg(test)]
2055struct ToneSource {
2056 format: iroh_live::media::format::AudioFormat,
2057 phase: f32,
2058 frequency: f32,
2059}
2060
2061#[cfg(test)]
2062impl ToneSource {
2063 fn hz440() -> Self {
2064 Self {
2065 format: iroh_live::media::format::AudioFormat::mono_48k(),
2066 phase: 0.0,
2067 frequency: 440.0,
2068 }
2069 }
2070}
2071
2072#[cfg(test)]
2073impl iroh_live::media::traits::AudioSource for ToneSource {
2074 fn format(&self) -> iroh_live::media::format::AudioFormat {
2075 self.format
2076 }
2077
2078 fn pop_samples(
2079 &mut self,
2080 buf: &mut [f32],
2081 ) -> anyhow::Result<Option<usize>> {
2082 let channels = self.format.channel_count.max(1) as usize;
2083 let frames = buf.len() / channels;
2084 let phase_inc = self.frequency / self.format.sample_rate as f32;
2085 for i in 0..frames {
2086 let sample = (2.0 * std::f32::consts::PI * self.phase).sin() * 0.5;
2087 for ch in 0..channels {
2088 buf[i * channels + ch] = sample;
2089 }
2090 self.phase += phase_inc;
2091 self.phase -= self.phase.floor();
2092 }
2093 // Always a full buffer — never `None` (iroh-live encoder skips `None`).
2094 Ok(Some(buf.len()))
2095 }
2096}
2097
2098/// Target peak after AGC (linear). Browsers apply getUserMedia AGC; cpal/PW
2099/// does not — EMEET often lands at peak ~0.005 which Opus/bots treat as silence.
2100const AGC_TARGET_PEAK: f32 = 0.28;
2101/// Cap boost so noise floor alone doesn't become roar (≈ +32 dB).
2102const AGC_MAX_GAIN: f32 = 40.0;
2103/// Don't boost pure digital silence / inactive rings.
2104const AGC_MIN_PEAK: f32 = 5e-5;
2105/// Speech-ish energy after gain (for diagnostics).
2106const VOICED_PEAK: f32 = 0.02;
2107
2108/// Wraps an AudioSource and emits silence while muted (keeps the Opus track live).
2109///
2110/// Mic level is measured on the **post-AGC** samples before mute zeros them.
2111///
2112/// When the inner source returns `None` (cpal ring not ready yet), we still
2113/// hand the encoder a full silence frame. The iroh-live encoder skips ticks
2114/// on `None`, which means **no Opus packets go out** — peers hear nothing
2115/// until the ring becomes ready, and intermittent `None`s produce dropouts.
2116/// Padding matches freeq-sdk-ffi `PushAudioSource` (always `Some(buf.len())`).
2117///
2118/// Short `Some(n)` reads are also padded to a full buffer so underruns never
2119/// starve the Opus encoder of continuous frames.
2120struct MuteableSource {
2121 inner: Box<dyn iroh_live::media::traits::AudioSource>,
2122 muted: Arc<AtomicBool>,
2123 level: MicLevel,
2124 /// Encode pulls (each ~20ms). Used for sparse diagnostics.
2125 pulls: u64,
2126 /// Pulls that had post-AGC energy above [`VOICED_PEAK`].
2127 voiced: u64,
2128 /// Adaptive linear gain (smoothed).
2129 gain: f32,
2130 /// Smoothed pre-gain peak for AGC.
2131 smooth_peak: f32,
2132}
2133
2134impl MuteableSource {
2135 fn silence(muted: Arc<AtomicBool>, level: MicLevel) -> Self {
2136 Self {
2137 inner: Box::new(SilenceSource::default()),
2138 muted,
2139 level,
2140 pulls: 0,
2141 voiced: 0,
2142 gain: 1.0,
2143 smooth_peak: 0.0,
2144 }
2145 }
2146
2147 fn apply_agc(&mut self, buf: &mut [f32], pre_peak: f32) -> f32 {
2148 // Slow envelope so gain doesn't pump on every syllable.
2149 self.smooth_peak = self.smooth_peak * 0.92 + pre_peak * 0.08;
2150 if self.smooth_peak >= AGC_MIN_PEAK {
2151 let desired = (AGC_TARGET_PEAK / self.smooth_peak).clamp(1.0, AGC_MAX_GAIN);
2152 self.gain = self.gain * 0.9 + desired * 0.1;
2153 } else {
2154 // Decay gain when capture is dead so we don't explode on first sample.
2155 self.gain = (self.gain * 0.95).max(1.0);
2156 }
2157 let g = self.gain;
2158 let mut post_peak = 0.0f32;
2159 for s in buf.iter_mut() {
2160 let v = (*s * g).clamp(-0.95, 0.95);
2161 *s = v;
2162 post_peak = post_peak.max(v.abs());
2163 }
2164 post_peak
2165 }
2166}
2167
2168impl iroh_live::media::traits::AudioSource for MuteableSource {
2169 fn format(&self) -> iroh_live::media::format::AudioFormat {
2170 self.inner.format()
2171 }
2172
2173 fn pop_samples(
2174 &mut self,
2175 buf: &mut [f32],
2176 ) -> anyhow::Result<Option<usize>> {
2177 let mut pre_peak = 0.0f32;
2178 let n = match self.inner.pop_samples(buf)? {
2179 Some(n) if n > 0 => {
2180 let take = n.min(buf.len());
2181 for &s in &buf[..take] {
2182 pre_peak = pre_peak.max(s.abs());
2183 }
2184 // Pad remainder before AGC so we gain a full encoder frame.
2185 for s in &mut buf[take..] {
2186 *s = 0.0;
2187 }
2188 let post = self.apply_agc(buf, pre_peak);
2189 self.level.observe(buf);
2190 let _ = post;
2191 buf.len()
2192 }
2193 Some(_) | None => {
2194 // Capture underrun / not-ready / empty: keep the track alive.
2195 for s in buf.iter_mut() {
2196 *s = 0.0;
2197 }
2198 self.level.observe(&[]);
2199 self.smooth_peak *= 0.9;
2200 buf.len()
2201 }
2202 };
2203 let mut post_peak = 0.0f32;
2204 for &s in buf.iter() {
2205 post_peak = post_peak.max(s.abs());
2206 }
2207 self.pulls = self.pulls.saturating_add(1);
2208 if post_peak > VOICED_PEAK {
2209 self.voiced = self.voiced.saturating_add(1);
2210 }
2211 let is_muted = self.muted.load(Ordering::Relaxed);
2212 if is_muted {
2213 for s in buf.iter_mut() {
2214 *s = 0.0;
2215 }
2216 }
2217 // Sparse log: first encode pull, then every ~5s (250 * 20ms).
2218 if self.pulls == 1 || self.pulls % 250 == 0 {
2219 log::info!(
2220 "av-media: outbound encode pulls={} voiced={} muted={} \
2221 pre_peak={pre_peak:.4} post_peak={post_peak:.4} gain={:.1}x",
2222 self.pulls,
2223 self.voiced,
2224 is_muted,
2225 self.gain
2226 );
2227 }
2228 // Always `Some(full)` — continuous Opus packets while the call is live.
2229 Ok(Some(n))
2230 }
2231}
2232
2233#[cfg(test)]
2234mod tests {
2235 use super::*;
2236 use iroh_live::media::codec::AudioCodec;
2237 use iroh_live::media::format::{AudioFormat, AudioPreset};
2238 use iroh_live::media::playout::PlaybackPolicy;
2239 use iroh_live::media::publish::LocalBroadcast;
2240 use iroh_live::media::subscribe::RemoteBroadcast;
2241 use iroh_live::media::traits::{
2242 AudioSink, AudioSinkHandle, AudioSource, AudioStreamFactory,
2243 };
2244 use n0_future::boxed::BoxFuture;
2245 use std::time::Duration;
2246
2247 /// Real-device check (Linux): open a camera with the same capture crate the
2248 /// app uses, pull frames, assert non-uniform content. Uses
2249 /// `SLEEK_TEST_CAMERA_ID` (e.g. `/dev/video10`) when set, else default.
2250 /// Skips with a printed reason when the device is busy or absent.
2251 #[cfg(all(test, not(target_os = "android")))]
2252 #[test]
2253 fn real_camera_capture_produces_nonuniform_frames() {
2254 let id = std::env::var("SLEEK_TEST_CAMERA_ID").ok();
2255 let Ok(mut cam) = CameraCapturer::open(None, id.as_deref(), &camera_config()) else {
2256 eprintln!("SKIP real_camera id={id:?}: open failed (busy/absent)");
2257 return;
2258 };
2259 if let Err(e) = cam.start() {
2260 eprintln!("SKIP real_camera id={id:?}: start failed: {e}");
2261 return;
2262 }
2263 let deadline = std::time::Instant::now() + Duration::from_millis(2_500);
2264 let mut frame = None;
2265 while std::time::Instant::now() < deadline {
2266 match cam.pop_frame() {
2267 Ok(Some(f)) => {
2268 frame = Some(f);
2269 break;
2270 }
2271 Ok(None) => std::thread::sleep(Duration::from_millis(20)),
2272 Err(e) => {
2273 let _ = cam.stop();
2274 eprintln!("SKIP real_camera id={id:?}: pop_frame error: {e}");
2275 return;
2276 }
2277 }
2278 }
2279 let _ = cam.stop();
2280 let Some(f) = frame else {
2281 eprintln!(
2282 "SKIP real_camera id={id:?}: no frames (OBS Virtual Camera not started?)"
2283 );
2284 return;
2285 };
2286 let (w, h) = (f.width(), f.height());
2287 let rgba = f.rgba_image();
2288 let bytes = rgba.as_raw().as_slice();
2289 let (min, max) = rgba_luma_range(bytes);
2290 let (r0, g0, b0, a0) = bytes
2291 .get(0..4)
2292 .map(|p| (p[0], p[1], p[2], p[3]))
2293 .unwrap_or((0, 0, 0, 0));
2294 eprintln!(
2295 "camera id={id:?} {w}x{h} rgba0=({r0},{g0},{b0},{a0}) luma range {min}..{max}"
2296 );
2297 assert!(
2298 max > min,
2299 "real camera frame must be non-uniform (luma range {min}..{max}); \
2300 got a blank/uniform buffer"
2301 );
2302 }
2303
2304 /// Source that always returns `None` — models cpal InputNotReady.
2305 struct AlwaysNone;
2306
2307 impl AudioSource for AlwaysNone {
2308 fn format(&self) -> AudioFormat {
2309 AudioFormat::mono_48k()
2310 }
2311 fn pop_samples(&mut self, _buf: &mut [f32]) -> anyhow::Result<Option<usize>> {
2312 Ok(None)
2313 }
2314 }
2315
2316 /// Source that returns short buffers (partial read underrun).
2317 struct ShortRead {
2318 n: usize,
2319 }
2320
2321 impl AudioSource for ShortRead {
2322 fn format(&self) -> AudioFormat {
2323 AudioFormat::mono_48k()
2324 }
2325 fn pop_samples(&mut self, buf: &mut [f32]) -> anyhow::Result<Option<usize>> {
2326 let n = self.n.min(buf.len());
2327 for s in &mut buf[..n] {
2328 *s = 0.25;
2329 }
2330 Ok(Some(n))
2331 }
2332 }
2333
2334 #[test]
2335 fn muteable_source_never_returns_none_on_underrun() {
2336 let muted = Arc::new(AtomicBool::new(false));
2337 let mut src = MuteableSource {
2338 inner: Box::new(AlwaysNone),
2339 muted,
2340 level: MicLevel::new(),
2341 pulls: 0,
2342 voiced: 0,
2343 gain: 1.0,
2344 smooth_peak: 0.0,
2345 };
2346 let mut buf = [1.0f32; 960]; // 20ms @ 48k
2347 for _ in 0..50 {
2348 let n = src.pop_samples(&mut buf).unwrap();
2349 assert_eq!(n, Some(buf.len()), "encoder must get continuous frames");
2350 assert!(
2351 buf.iter().all(|&s| s == 0.0),
2352 "underrun padding must be silence"
2353 );
2354 }
2355 }
2356
2357 #[test]
2358 fn muteable_source_pads_short_reads_to_full_buffer() {
2359 let muted = Arc::new(AtomicBool::new(false));
2360 let mut src = MuteableSource {
2361 inner: Box::new(ShortRead { n: 100 }),
2362 muted,
2363 level: MicLevel::new(),
2364 pulls: 0,
2365 voiced: 0,
2366 gain: 1.0,
2367 smooth_peak: 0.0,
2368 };
2369 let mut buf = [0.0f32; 960];
2370 let n = src.pop_samples(&mut buf).unwrap();
2371 assert_eq!(n, Some(960));
2372 // AGC may boost above 0.25; pad region stays 0.
2373 assert!(
2374 buf[..100].iter().all(|&s| s > 0.2),
2375 "short-read region should keep signal (with AGC)"
2376 );
2377 assert!(buf[100..].iter().all(|&s| s == 0.0));
2378 }
2379
2380 #[test]
2381 fn muteable_source_zeros_when_muted_but_keeps_frames() {
2382 let muted = Arc::new(AtomicBool::new(true));
2383 let mut src = MuteableSource {
2384 inner: Box::new(ToneSource::hz440()),
2385 muted,
2386 level: MicLevel::new(),
2387 pulls: 0,
2388 voiced: 0,
2389 gain: 1.0,
2390 smooth_peak: 0.0,
2391 };
2392 let mut buf = [0.0f32; 480];
2393 let n = src.pop_samples(&mut buf).unwrap();
2394 assert_eq!(n, Some(buf.len()));
2395 assert_eq!(pcm_rms(&buf), 0.0, "muted must be silence");
2396 // Level still saw pre-mute energy from the tone.
2397 assert!(
2398 src.level.get() > 0.01,
2399 "mic meter should move while muted"
2400 );
2401 }
2402
2403 #[test]
2404 fn muteable_source_tone_has_energy_when_unmuted() {
2405 let muted = Arc::new(AtomicBool::new(false));
2406 let mut src = MuteableSource {
2407 inner: Box::new(ToneSource::hz440()),
2408 muted,
2409 level: MicLevel::new(),
2410 pulls: 0,
2411 voiced: 0,
2412 gain: 1.0,
2413 smooth_peak: 0.0,
2414 };
2415 let mut buf = [0.0f32; 4800]; // 100ms
2416 let n = src.pop_samples(&mut buf).unwrap();
2417 assert_eq!(n, Some(buf.len()));
2418 let rms = pcm_rms(&buf);
2419 assert!(
2420 rms > 0.1,
2421 "unmuted tone RMS {rms} should be clearly non-silent"
2422 );
2423 }
2424
2425 #[test]
2426 fn silence_source_is_continuous_zeros() {
2427 let mut s = SilenceSource::default();
2428 let mut buf = [1.0f32; 320];
2429 assert_eq!(s.pop_samples(&mut buf).unwrap(), Some(320));
2430 assert!(buf.iter().all(|&x| x == 0.0));
2431 }
2432
2433 /// freeq-av-style tap: capture decoded PCM instead of playing it.
2434 struct TapBackend {
2435 tx: std::sync::mpsc::SyncSender<Vec<f32>>,
2436 }
2437
2438 impl AudioStreamFactory for TapBackend {
2439 fn create_input(
2440 &self,
2441 format: AudioFormat,
2442 ) -> BoxFuture<anyhow::Result<Box<dyn AudioSource>>> {
2443 let src = SilenceSource {
2444 format: if format.sample_rate == 0 {
2445 AudioFormat::mono_48k()
2446 } else {
2447 format
2448 },
2449 };
2450 Box::pin(async move { Ok(Box::new(src) as Box<dyn AudioSource>) })
2451 }
2452
2453 fn create_output(
2454 &self,
2455 format: AudioFormat,
2456 ) -> BoxFuture<anyhow::Result<Box<dyn AudioSink>>> {
2457 let tx = self.tx.clone();
2458 Box::pin(async move {
2459 Ok(Box::new(TapSink {
2460 format,
2461 paused: Arc::new(AtomicBool::new(false)),
2462 tx,
2463 }) as Box<dyn AudioSink>)
2464 })
2465 }
2466 }
2467
2468 struct TapSink {
2469 format: AudioFormat,
2470 paused: Arc<AtomicBool>,
2471 tx: std::sync::mpsc::SyncSender<Vec<f32>>,
2472 }
2473
2474 impl AudioSinkHandle for TapSink {
2475 fn cloned_boxed(&self) -> Box<dyn AudioSinkHandle> {
2476 Box::new(TapSinkHandle {
2477 paused: self.paused.clone(),
2478 })
2479 }
2480 fn pause(&self) {
2481 self.paused.store(true, Ordering::Relaxed);
2482 }
2483 fn resume(&self) {
2484 self.paused.store(false, Ordering::Relaxed);
2485 }
2486 fn is_paused(&self) -> bool {
2487 self.paused.load(Ordering::Relaxed)
2488 }
2489 fn toggle_pause(&self) {
2490 let _ = self
2491 .paused
2492 .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| Some(!v));
2493 }
2494 }
2495
2496 impl AudioSink for TapSink {
2497 fn format(&self) -> anyhow::Result<AudioFormat> {
2498 Ok(self.format)
2499 }
2500 fn push_samples(&mut self, buf: &[f32]) -> anyhow::Result<()> {
2501 let _ = self.tx.try_send(buf.to_vec());
2502 Ok(())
2503 }
2504 fn handle(&self) -> Box<dyn AudioSinkHandle> {
2505 Box::new(TapSinkHandle {
2506 paused: self.paused.clone(),
2507 })
2508 }
2509 }
2510
2511 struct TapSinkHandle {
2512 paused: Arc<AtomicBool>,
2513 }
2514
2515 impl AudioSinkHandle for TapSinkHandle {
2516 fn cloned_boxed(&self) -> Box<dyn AudioSinkHandle> {
2517 Box::new(TapSinkHandle {
2518 paused: self.paused.clone(),
2519 })
2520 }
2521 fn pause(&self) {
2522 self.paused.store(true, Ordering::Relaxed);
2523 }
2524 fn resume(&self) {
2525 self.paused.store(false, Ordering::Relaxed);
2526 }
2527 fn is_paused(&self) -> bool {
2528 self.paused.load(Ordering::Relaxed)
2529 }
2530 fn toggle_pause(&self) {
2531 let _ = self
2532 .paused
2533 .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| Some(!v));
2534 }
2535 }
2536
2537 /// Publish through the shipped MuteableSource → LocalBroadcast Opus path,
2538 /// subscribe like freeq-av/eve (RemoteBroadcast + tap sink), assert energy.
2539 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
2540 async fn outbound_muteable_tone_is_audible_to_subscriber() {
2541 let path = broadcast_path("01TESTSESS", "desktop", "a1b2c3d4");
2542 assert_eq!(path, "01TESTSESS/desktop~a1b2c3d4");
2543
2544 let broadcast = LocalBroadcast::new();
2545 let muted = Arc::new(AtomicBool::new(false));
2546 let muteable = MuteableSource {
2547 inner: Box::new(ToneSource::hz440()),
2548 muted,
2549 level: MicLevel::new(),
2550 pulls: 0,
2551 voiced: 0,
2552 gain: 1.0,
2553 smooth_peak: 0.0,
2554 };
2555 broadcast
2556 .audio()
2557 .set(muteable, AudioCodec::Opus, [AudioPreset::Hq])
2558 .expect("set Opus source (shipped publish path)");
2559
2560 let consumer = broadcast.consume();
2561 // Keep producer alive for the duration of the test.
2562 let _keepalive = broadcast;
2563
2564 let remote = RemoteBroadcast::with_playback_policy(
2565 &path,
2566 consumer,
2567 PlaybackPolicy::unmanaged(),
2568 )
2569 .await
2570 .expect("catalog from LocalBroadcast");
2571
2572 let (tx, rx) = std::sync::mpsc::sync_channel::<Vec<f32>>(64);
2573 let backend = TapBackend { tx };
2574 let _track = remote
2575 .audio_ready(&backend)
2576 .await
2577 .expect("audio_ready — catalog must advertise Opus");
2578
2579 // Collect decoded PCM for ~1.5s of wall time (Opus frame cadence).
2580 let deadline = std::time::Instant::now() + Duration::from_millis(1500);
2581 let mut samples: Vec<f32> = Vec::new();
2582 while std::time::Instant::now() < deadline {
2583 match rx.recv_timeout(Duration::from_millis(50)) {
2584 Ok(chunk) => samples.extend_from_slice(&chunk),
2585 Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
2586 Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
2587 }
2588 if samples.len() > 48_000 / 2 {
2589 break; // ~0.5s mono 48k is enough for RMS
2590 }
2591 }
2592
2593 assert!(
2594 !samples.is_empty(),
2595 "subscriber got no PCM — publish path not producing Opus frames"
2596 );
2597 let rms = pcm_rms(&samples);
2598 assert!(
2599 rms > 0.01,
2600 "decoded RMS {rms:.6} too low (samples={}) — bot would hear silence",
2601 samples.len()
2602 );
2603 }
2604
2605 /// Catalog-only silence when capture is missing still produces frames
2606 /// (listen-only), so agents stay attached rather than seeing no track.
2607 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
2608 async fn silence_source_still_publishes_continuous_opus_track() {
2609 let broadcast = LocalBroadcast::new();
2610 let muted = Arc::new(AtomicBool::new(true));
2611 let muteable = MuteableSource {
2612 inner: Box::new(SilenceSource::default()),
2613 muted,
2614 level: MicLevel::new(),
2615 pulls: 0,
2616 voiced: 0,
2617 gain: 1.0,
2618 smooth_peak: 0.0,
2619 };
2620 broadcast
2621 .audio()
2622 .set(muteable, AudioCodec::Opus, [AudioPreset::Hq])
2623 .expect("silence Opus set");
2624 let consumer = broadcast.consume();
2625 let _keepalive = broadcast;
2626
2627 let remote = RemoteBroadcast::with_playback_policy(
2628 "sess/listen-only",
2629 consumer,
2630 PlaybackPolicy::unmanaged(),
2631 )
2632 .await
2633 .expect("catalog");
2634
2635 let (tx, rx) = std::sync::mpsc::sync_channel::<Vec<f32>>(32);
2636 let backend = TapBackend { tx };
2637 let _track = remote.audio_ready(&backend).await.expect("audio track");
2638
2639 let deadline = std::time::Instant::now() + Duration::from_millis(800);
2640 let mut got = 0usize;
2641 while std::time::Instant::now() < deadline {
2642 if let Ok(chunk) = rx.recv_timeout(Duration::from_millis(40)) {
2643 got += chunk.len();
2644 if got > 2000 {
2645 break;
2646 }
2647 }
2648 }
2649 assert!(
2650 got > 0,
2651 "listen-only silence must still emit continuous frames (got 0 samples)"
2652 );
2653 }
2654
2655 #[test]
2656 fn camera_hardware_claimed_only_when_publish_enabled() {
2657 assert!(should_claim_camera_hardware(true));
2658 assert!(!should_claim_camera_hardware(false));
2659 }
2660
2661 /// Mute must stop the inner capturer (streamoff) — not merely discard
2662 /// frames after dqbuf, which left the privacy LED on.
2663 #[test]
2664 fn gated_camera_stops_inner_when_publish_off() {
2665 use iroh_live::media::format::{PixelFormat, VideoFormat, VideoFrame};
2666 use iroh_live::media::traits::VideoSource;
2667 use std::sync::atomic::AtomicUsize;
2668
2669 struct CountingCam {
2670 stops: Arc<AtomicUsize>,
2671 starts: Arc<AtomicUsize>,
2672 pops: Arc<AtomicUsize>,
2673 }
2674 impl VideoSource for CountingCam {
2675 fn name(&self) -> &str {
2676 "counting"
2677 }
2678 fn format(&self) -> VideoFormat {
2679 VideoFormat {
2680 pixel_format: PixelFormat::Rgba,
2681 dimensions: [2, 2],
2682 }
2683 }
2684 fn start(&mut self) -> anyhow::Result<()> {
2685 self.starts.fetch_add(1, Ordering::Relaxed);
2686 Ok(())
2687 }
2688 fn stop(&mut self) -> anyhow::Result<()> {
2689 self.stops.fetch_add(1, Ordering::Relaxed);
2690 Ok(())
2691 }
2692 fn pop_frame(&mut self) -> anyhow::Result<Option<VideoFrame>> {
2693 self.pops.fetch_add(1, Ordering::Relaxed);
2694 // Frame payload unused — we only assert stop/pop ordering.
2695 Ok(None)
2696 }
2697 }
2698
2699 let stops = Arc::new(AtomicUsize::new(0));
2700 let starts = Arc::new(AtomicUsize::new(0));
2701 let pops = Arc::new(AtomicUsize::new(0));
2702 let enabled = Arc::new(AtomicBool::new(true));
2703 let mut gated = GatedCameraSource {
2704 inner: Box::new(CountingCam {
2705 stops: stops.clone(),
2706 starts: starts.clone(),
2707 pops: pops.clone(),
2708 }),
2709 enabled: enabled.clone(),
2710 preview: VideoFrameStore::new(),
2711 frame_count: Arc::new(AtomicU64::new(0)),
2712 streaming: false,
2713 };
2714
2715 gated.start().expect("start");
2716 assert_eq!(starts.load(Ordering::Relaxed), 1);
2717 assert!(gated.pop_frame().expect("pop").is_none());
2718 assert_eq!(pops.load(Ordering::Relaxed), 1);
2719 assert_eq!(stops.load(Ordering::Relaxed), 0);
2720
2721 enabled.store(false, Ordering::Relaxed);
2722 assert!(gated.pop_frame().expect("gated pop").is_none());
2723 assert_eq!(
2724 stops.load(Ordering::Relaxed),
2725 1,
2726 "mute must call inner.stop() so STREAMON / privacy LED ends"
2727 );
2728 assert_eq!(
2729 pops.load(Ordering::Relaxed),
2730 1,
2731 "must not dqbuf/pop after publish off"
2732 );
2733
2734 // Further gated polls must not re-stop or capture.
2735 assert!(gated.pop_frame().expect("gated pop 2").is_none());
2736 assert_eq!(stops.load(Ordering::Relaxed), 1);
2737 assert_eq!(pops.load(Ordering::Relaxed), 1);
2738 }
2739
2740 #[tokio::test]
2741 async fn release_local_camera_clears_broadcast_video() {
2742 use iroh_live::media::format::{PixelFormat, VideoFormat, VideoFrame};
2743 use iroh_live::media::traits::VideoSource;
2744
2745 struct StubCam;
2746 impl VideoSource for StubCam {
2747 fn name(&self) -> &str {
2748 "stub"
2749 }
2750 fn format(&self) -> VideoFormat {
2751 VideoFormat {
2752 pixel_format: PixelFormat::Rgba,
2753 dimensions: [4, 4],
2754 }
2755 }
2756 fn start(&mut self) -> anyhow::Result<()> {
2757 Ok(())
2758 }
2759 fn stop(&mut self) -> anyhow::Result<()> {
2760 Ok(())
2761 }
2762 fn pop_frame(&mut self) -> anyhow::Result<Option<VideoFrame>> {
2763 Ok(None)
2764 }
2765 }
2766
2767 let broadcast = LocalBroadcast::new();
2768 broadcast
2769 .video()
2770 .set_source(StubCam, VideoCodec::H264, [VideoPreset::P360])
2771 .expect("set stub video");
2772 assert!(
2773 broadcast.preview().is_some(),
2774 "preview available while video source attached"
2775 );
2776
2777 let mut keepalive = broadcast.preview();
2778 let mut pump = None;
2779 let store = VideoFrameStore::new();
2780 store.set(LOCAL_PREVIEW_KEY, 4, 4, vec![0u8; 4 * 4 * 4].into());
2781 #[cfg(target_os = "android")]
2782 let mut guard = None;
2783
2784 release_local_camera(
2785 &broadcast,
2786 &mut keepalive,
2787 &mut pump,
2788 &store,
2789 #[cfg(target_os = "android")]
2790 &mut guard,
2791 );
2792
2793 assert!(keepalive.is_none());
2794 assert!(
2795 broadcast.preview().is_none(),
2796 "video.clear must drop the source so privacy LED can go dark"
2797 );
2798 assert!(
2799 !store
2800 .snapshot()
2801 .iter()
2802 .any(|(k, _)| k == LOCAL_PREVIEW_KEY),
2803 "local preview frame must be cleared on release"
2804 );
2805 }
2806
2807}