//! Independent playback engine for inline chat audio attachments. //! //! The rodio device sink stays on a dedicated OS thread and never enters iced //! state or the call-audio pipeline. The GUI sends small commands and reads a //! shared status snapshot at its redraw cadence. use crate::files::AttachmentId; use rodio::{Decoder, DeviceSinkBuilder, MixerDeviceSink, Player, Source, decoder::DecoderError}; use std::io::Cursor; use std::sync::{Arc, Mutex, mpsc}; use std::time::Duration; /// State published by the playback thread for the GUI. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct ClipStatus { pub playing_id: Option, pub position: Duration, pub total: Option, pub paused: bool, /// Set when output initialization or decoding rejects the requested clip. /// The app consumes this as a signal to fall back to the normal file chip. pub failure: Option, } #[derive(Debug, Clone, PartialEq, Eq)] pub struct ClipFailure { pub id: AttachmentId, pub error: String, /// Decoder rejection means the filename hint should fall back to a file /// chip. Output-device failures remain retryable as audio. pub invalid_audio: bool, } pub type SharedClipStatus = Arc>; #[derive(Debug)] enum ClipCommand { Play(AttachmentId, Vec), Pause, Resume, Seek(Duration), Stop, SetVolume(f32), } /// Cheap, `Send` command handle for the dedicated playback thread. pub struct ClipPlayer { command_tx: mpsc::Sender, status: SharedClipStatus, } impl ClipPlayer { /// Start the playback worker. The system output device is opened lazily on /// first Play, so merely launching PeerSpeak never claims another stream. /// /// `initial_volume` is the universal gain (`1.0` = unity) applied to every /// clip, restored from config so the level persists across sessions. pub fn new(initial_volume: f32) -> (Self, SharedClipStatus) { let (command_tx, command_rx) = mpsc::channel(); let status = Arc::new(Mutex::new(ClipStatus::default())); let worker_status = Arc::clone(&status); std::thread::Builder::new() .name("peerspeak-clip-player".to_string()) .spawn(move || playback_worker(command_rx, worker_status, initial_volume)) .expect("failed to spawn clip playback thread"); ( Self { command_tx, status: Arc::clone(&status), }, status, ) } pub fn play(&self, id: AttachmentId, bytes: Vec) { update_status(&self.status, |status| { status.playing_id = Some(id); status.position = Duration::ZERO; status.total = None; status.paused = false; status.failure = None; }); let _ = self.command_tx.send(ClipCommand::Play(id, bytes)); } pub fn pause(&self) { let _ = self.command_tx.send(ClipCommand::Pause); } pub fn resume(&self) { let _ = self.command_tx.send(ClipCommand::Resume); } pub fn seek(&self, position: Duration) { let _ = self.command_tx.send(ClipCommand::Seek(position)); } pub fn stop(&self) { let _ = self.command_tx.send(ClipCommand::Stop); } /// Set the universal playback gain (`1.0` = unity). Applies to the current /// clip immediately and to every clip played afterwards. pub fn set_volume(&self, volume: f32) { let _ = self.command_tx.send(ClipCommand::SetVolume(volume)); } } fn playback_worker( command_rx: mpsc::Receiver, status: SharedClipStatus, initial_volume: f32, ) { let mut output: Option = None; let mut player: Option = None; // Universal gain remembered across clips so a level set on one upload // carries to the next; reapplied to each freshly connected player. let mut volume = initial_volume.max(0.0); loop { match command_rx.recv_timeout(Duration::from_millis(100)) { Ok(ClipCommand::Play(id, bytes)) => { // In-memory readers do not expose file metadata to rodio. Pass // the known attachment length explicitly so formats without a // duration in their headers (notably MP3 and Vorbis) can derive // a total duration and support reliable seeking. let source = match decode_clip(bytes) { Ok(source) => source, Err(error) => { fail( &status, id, format!("unsupported or invalid audio: {error}"), true, ); continue; } }; let total = source.total_duration(); if output.is_none() { match DeviceSinkBuilder::open_default_sink() { Ok(sink) => { let new_player = Player::connect_new(sink.mixer()); new_player.set_volume(volume); player = Some(new_player); output = Some(sink); } Err(error) => { fail( &status, id, format!("audio output unavailable: {error}"), false, ); continue; } } } if let Some(player) = player.as_ref() { player.clear(); player.append(source); player.play(); update_status(&status, |s| { s.playing_id = Some(id); s.position = Duration::ZERO; s.total = total; s.paused = false; s.failure = None; }); } } Ok(ClipCommand::Pause) => { if let Some(player) = player.as_ref() { player.pause(); update_status(&status, |s| s.paused = true); } } Ok(ClipCommand::Resume) => { if let Some(player) = player.as_ref() { player.play(); update_status(&status, |s| s.paused = false); } } Ok(ClipCommand::Seek(position)) => { if let Some(player) = player.as_ref() && player.try_seek(position).is_ok() { update_status(&status, |s| s.position = position); } } Ok(ClipCommand::Stop) => { if let Some(player) = player.as_ref() { player.clear(); } reset(&status); } Ok(ClipCommand::SetVolume(level)) => { volume = level.max(0.0); if let Some(player) = player.as_ref() { player.set_volume(volume); } } Err(mpsc::RecvTimeoutError::Disconnected) => break, Err(mpsc::RecvTimeoutError::Timeout) => {} } if let Some(player) = player.as_ref() { let (active, failed) = status .lock() .map(|s| (s.playing_id.is_some(), s.failure.is_some())) .unwrap_or_default(); if active && !failed && player.empty() { reset(&status); } else if active && !failed { update_status(&status, |s| { s.position = player.get_pos(); s.paused = player.is_paused(); }); } } } } fn decode_clip(bytes: Vec) -> Result>>, DecoderError> { let byte_len = bytes.len() as u64; Decoder::builder() .with_data(Cursor::new(bytes)) .with_byte_len(byte_len) .build() } fn fail(status: &SharedClipStatus, id: AttachmentId, error: String, invalid_audio: bool) { crate::log_msg(&format!("Inline audio playback failed: {error}")); update_status(status, |s| { // Keep the id active until the GUI observes the failure on its next // tick. This guarantees the active-only timer cannot disappear in the // small window between sending Play and decoder/output failure. s.playing_id = Some(id); s.position = Duration::ZERO; s.total = None; s.paused = false; s.failure = Some(ClipFailure { id, error, invalid_audio, }); }); } fn reset(status: &SharedClipStatus) { update_status(status, |s| *s = ClipStatus::default()); } fn update_status(status: &SharedClipStatus, update: impl FnOnce(&mut ClipStatus)) { if let Ok(mut status) = status.lock() { update(&mut status); } } pub fn status_snapshot(status: &SharedClipStatus) -> ClipStatus { status.lock().map(|s| s.clone()).unwrap_or_default() } /// Format clip time as `mm:ss` (hours are folded into minutes). pub fn format_time(duration: Duration) -> String { let seconds = duration.as_secs(); format!("{}:{:02}", seconds / 60, seconds % 60) } /// Playback progress in `0.0..=1.0`; unknown and zero durations report zero. pub fn progress(position: Duration, total: Option) -> f32 { let Some(total) = total.filter(|duration| !duration.is_zero()) else { return 0.0; }; (position.as_secs_f64() / total.as_secs_f64()).clamp(0.0, 1.0) as f32 } /// Convert a slider fraction into a clamped position within a clip. pub fn seek_target(fraction: f32, total: Duration) -> Duration { total.mul_f64(f64::from(fraction.clamp(0.0, 1.0))) } #[cfg(test)] mod tests { use super::*; use base64::Engine; #[test] fn in_memory_mp3_reports_duration() { // One headerless constant-bitrate MP3 frame repeated to model files // that do not carry an Xing/VBR duration header. let frame = base64::engine::general_purpose::STANDARD .decode("//sQxAAABIQVWVRggDCqCKiDNlAAAAGgS4BgAmTT2AQAABCxOD5d7gQOfqBAEHS4Ph/EAIRI7//0A0KBNpABgMRIDCSI04PcIFdF0PJKFgzlUf5eAoF8BRIPfh4FTvUDQl+dUi5pc0w=") .expect("valid test fixture"); let bytes = frame.repeat(20); let decoder = decode_clip(bytes).expect("CBR MP3 should decode"); assert!(decoder.total_duration().is_some()); } #[test] fn formats_clip_time() { assert_eq!(format_time(Duration::ZERO), "0:00"); assert_eq!(format_time(Duration::from_secs(65)), "1:05"); assert_eq!(format_time(Duration::from_secs(3_661)), "61:01"); } #[test] fn progress_handles_unknown_zero_and_clamps() { assert_eq!(progress(Duration::from_secs(1), None), 0.0); assert_eq!(progress(Duration::from_secs(1), Some(Duration::ZERO)), 0.0); assert_eq!( progress(Duration::from_secs(5), Some(Duration::from_secs(10))), 0.5 ); assert_eq!( progress(Duration::from_secs(20), Some(Duration::from_secs(10))), 1.0 ); } #[test] fn seek_target_clamps_fraction() { let total = Duration::from_secs(100); assert_eq!(seek_target(0.25, total), Duration::from_secs(25)); assert_eq!(seek_target(-1.0, total), Duration::ZERO); assert_eq!(seek_target(2.0, total), total); } }