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