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