From fb9a9dd4392c78efdf1ce076c0b15f90d6d96cc2 Mon Sep 17 00:00:00 2001 From: Del Wang Date: Mon, 29 Dec 2025 19:14:42 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E4=BC=98=E5=8C=96=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/hello/src/bin/client.rs | 203 +++++++++++++++++------------------ apps/hello/src/bin/stereo.rs | 79 ++++++-------- apps/hello/src/bin/test.rs | 95 ++++++++-------- 3 files changed, 168 insertions(+), 209 deletions(-) diff --git a/apps/hello/src/bin/client.rs b/apps/hello/src/bin/client.rs index 75ba8f0..1fee262 100644 --- a/apps/hello/src/bin/client.rs +++ b/apps/hello/src/bin/client.rs @@ -1,9 +1,7 @@ -use anyhow::Result; -#[cfg(not(target_os = "linux"))] -use hello::audio::OpusCodec; -#[cfg(target_os = "linux")] -use hello::audio::{AudioPlayer, OpusCodec}; +#![cfg(target_os = "linux")] +use anyhow::Result; +use hello::audio::{AudioPlayer, OpusCodec}; use hello::config::AudioConfig; use hello::net::{AudioPacket, ChannelRole, ControlPacket, Discovery}; use hello::sync::{ClockSync, now_us}; @@ -15,120 +13,111 @@ use tokio::net::TcpStream; #[tokio::main] async fn main() -> Result<()> { - #[cfg(not(target_os = "linux"))] - { - println!("Client is only supported on Linux (ALSA required)"); + let args: Vec = env::args().collect(); + if args.len() < 2 { + eprintln!("Usage: {} [left|right]", args[0]); return Ok(()); } + let role = if args[1] == "left" { + ChannelRole::Left + } else { + ChannelRole::Right + }; - #[cfg(target_os = "linux")] + let config = AudioConfig { + sample_rate: 48000, + channels: 1, + frame_size: 960, + bitrate: 32000, + ..AudioConfig::default() + }; + + println!("Searching for server..."); + let server_addr = Discovery::client_discover_server().await?; + let mut stream = TcpStream::connect(server_addr).await?; + + // 1. Identify + stream + .write_all(&postcard::to_allocvec(&ControlPacket::ClientIdentify { + role: role.clone(), + })?) + .await?; + + // 2. Sync + let mut clock = ClockSync::new(); + let t1 = now_us(); + stream + .write_all(&postcard::to_allocvec(&ControlPacket::Ping { + client_ts: t1, + })?) + .await?; + let mut buf = [0u8; 1024]; + let n = stream.read(&mut buf).await?; + if let Ok(ControlPacket::Pong { + client_ts, + server_ts, + .. + }) = postcard::from_bytes::(&buf[..n]) { - let args: Vec = env::args().collect(); - if args.len() < 2 { - eprintln!("Usage: {} [left|right]", args[0]); - return Ok(()); - } - let role = if args[1] == "left" { - ChannelRole::Left - } else { - ChannelRole::Right - }; + let t4 = now_us(); + clock.update(client_ts, server_ts, t4); + println!( + "Clock synced. Offset: {}us, RTT: {}us", + clock.offset, + t4 - t1 + ); + } - let config = AudioConfig { - sample_rate: 48000, - channels: 1, - frame_size: 960, - bitrate: 32000, - ..AudioConfig::default() - }; - - println!("Searching for server..."); - let server_addr = Discovery::client_discover_server().await?; - let mut stream = TcpStream::connect(server_addr).await?; - - // 1. Identify - stream - .write_all(&postcard::to_allocvec(&ControlPacket::ClientIdentify { - role: role.clone(), - })?) - .await?; - - // 2. Sync - let mut clock = ClockSync::new(); - let t1 = now_us(); - stream - .write_all(&postcard::to_allocvec(&ControlPacket::Ping { - client_ts: t1, - })?) - .await?; - let mut buf = [0u8; 1024]; - let n = stream.read(&mut buf).await?; - if let Ok(ControlPacket::Pong { - client_ts, - server_ts, - .. - }) = postcard::from_bytes::(&buf[..n]) - { - let t4 = now_us(); - clock.update(client_ts, server_ts, t4); - println!( - "Clock synced. Offset: {}us, RTT: {}us", - clock.offset, - t4 - t1 - ); - } - - let player = AudioPlayer::new(&config)?; - let mut codec = OpusCodec::new(&config)?; - let mut jitter_buffer: VecDeque = VecDeque::new(); - let mut pcm_buf = vec![0i16; config.frame_size]; - - let (tx, mut rx) = tokio::sync::mpsc::channel(100); - tokio::spawn(async move { - loop { - let mut s_buf = [0u8; 4]; - if stream.read_exact(&mut s_buf).await.is_err() { - break; - } - let size = u32::from_be_bytes(s_buf) as usize; - let mut data = vec![0u8; size]; - if stream.read_exact(&mut data).await.is_err() { - break; - } - if let Ok(p) = postcard::from_bytes::(&data) { - let _ = tx.send(p).await; - } - } - }); - - println!("Streaming started. Playing channel {:?}", role); + let player = AudioPlayer::new(&config)?; + let mut codec = OpusCodec::new(&config)?; + let mut jitter_buffer: VecDeque = VecDeque::new(); + let mut pcm_buf = vec![0i16; config.frame_size]; + let (tx, mut rx) = tokio::sync::mpsc::channel(100); + tokio::spawn(async move { loop { - while let Ok(p) = rx.try_recv() { - jitter_buffer.push_back(p); + let mut s_buf = [0u8; 4]; + if stream.read_exact(&mut s_buf).await.is_err() { + break; } + let size = u32::from_be_bytes(s_buf) as usize; + let mut data = vec![0u8; size]; + if stream.read_exact(&mut data).await.is_err() { + break; + } + if let Ok(p) = postcard::from_bytes::(&data) { + let _ = tx.send(p).await; + } + } + }); - if let Some(p) = jitter_buffer.front() { - let target_client_time = clock.to_client_time(p.timestamp); - let now = now_us(); + println!("Streaming started. Playing channel {:?}", role); - if now >= target_client_time { - let packet = jitter_buffer.pop_front().unwrap(); - let len = codec.decode(&packet.data, &mut pcm_buf)?; - player.write(&pcm_buf[..len])?; - } else if target_client_time - now > 500_000 { - // Too far in the future, maybe clock jumped? - jitter_buffer.pop_front(); - } else { - // Wait until it's time - let wait = (target_client_time - now) as u64; - if wait > 1000 { - tokio::time::sleep(Duration::from_micros(wait)).await; - } - } + loop { + while let Ok(p) = rx.try_recv() { + jitter_buffer.push_back(p); + } + + if let Some(p) = jitter_buffer.front() { + let target_client_time = clock.to_client_time(p.timestamp); + let now = now_us(); + + if now >= target_client_time { + let packet = jitter_buffer.pop_front().unwrap(); + let len = codec.decode(&packet.data, &mut pcm_buf)?; + player.write(&pcm_buf[..len])?; + } else if target_client_time - now > 500_000 { + // Too far in the future, maybe clock jumped? + jitter_buffer.pop_front(); } else { - tokio::time::sleep(Duration::from_millis(5)).await; + // Wait until it's time + let wait = (target_client_time - now) as u64; + if wait > 1000 { + tokio::time::sleep(Duration::from_micros(wait)).await; + } } + } else { + tokio::time::sleep(Duration::from_millis(5)).await; } } } diff --git a/apps/hello/src/bin/stereo.rs b/apps/hello/src/bin/stereo.rs index fafe4ca..8f3c8cb 100644 --- a/apps/hello/src/bin/stereo.rs +++ b/apps/hello/src/bin/stereo.rs @@ -1,5 +1,6 @@ +#![cfg(target_os = "linux")] + use anyhow::{Context, Result}; -#[cfg(target_os = "linux")] use hello::audio::{AudioPlayer, OpusCodec}; use hello::config::AudioConfig; use hello::net::{AudioPacket, ChannelRole, ControlPacket, DISCOVERY_PORT, SERVER_PORT}; @@ -14,14 +15,10 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream, UdpSocket}; use tokio::signal; -#[cfg(target_os = "linux")] const FIFO_PATH: &str = "/tmp/stereo_out.fifo"; -#[cfg(target_os = "linux")] -const TEMP_ASOUND_CONF: &str = "/tmp/asound.stereo.conf"; -#[cfg(target_os = "linux")] const REAL_ASOUND_CONF: &str = "/etc/asound.conf"; +const TEMP_ASOUND_CONF: &str = "/tmp/asound.stereo.conf"; -#[cfg(target_os = "linux")] #[derive(Debug, Clone, Copy, PartialEq)] enum DeviceModel { Lx06, @@ -29,7 +26,6 @@ enum DeviceModel { Unknown, } -#[cfg(target_os = "linux")] fn detect_model() -> DeviceModel { let model = Command::new("sh") .args(["-c", "micocfg_model"]) @@ -45,7 +41,6 @@ fn detect_model() -> DeviceModel { DeviceModel::Unknown } -#[cfg(target_os = "linux")] fn setup_alsa_config(model: DeviceModel) -> Result<()> { cleanup_alsa_config(); @@ -89,14 +84,13 @@ pcm.stereo_interceptor {{ .context("Failed to execute mount command")?; if !status.success() { - return Err(anyhow::anyhow!("Failed to mount asound.conf, need root?")); + return Err(anyhow::anyhow!("Failed to mount asound.conf")); } println!("Successfully redirected ALSA output to {}", FIFO_PATH); Ok(()) } -#[cfg(target_os = "linux")] fn cleanup_alsa_config() { println!("Cleaning up ALSA configurations..."); let _ = Command::new("umount") @@ -109,50 +103,40 @@ fn cleanup_alsa_config() { #[tokio::main] async fn main() -> Result<()> { - #[cfg(not(target_os = "linux"))] - { - println!("Stereo 模式仅支持 Linux (需要 ALSA)"); + let args: Vec = env::args().collect(); + if args.len() < 3 { + eprintln!("用法: {} [master|slave] [left|right]", args[0]); + eprintln!("示例:"); + eprintln!(" 主设备: {} master left", args[0]); + eprintln!(" 从设备: {} slave right", args[0]); return Ok(()); } - #[cfg(target_os = "linux")] - { - let args: Vec = env::args().collect(); - if args.len() < 3 { - eprintln!("用法: {} [master|slave] [left|right]", args[0]); - eprintln!("示例:"); - eprintln!(" 主设备: {} master left", args[0]); - eprintln!(" 从设备: {} slave right", args[0]); - return Ok(()); - } + let mode = &args[1]; + let role = if args[2].to_lowercase() == "left" { + ChannelRole::Left + } else { + ChannelRole::Right + }; - let mode = &args[1]; - let role = if args[2].to_lowercase() == "left" { - ChannelRole::Left - } else { - ChannelRole::Right - }; - - let result = tokio::select! { - res = async { - if mode == "master" { - run_master(role).await - } else { - run_slave(role).await - } - } => res, - _ = signal::ctrl_c() => { - println!("\n收到退出信号"); - Ok(()) + let result = tokio::select! { + res = async { + if mode == "master" { + run_master(role).await + } else { + run_slave(role).await } - }; + } => res, + _ = signal::ctrl_c() => { + println!("\n收到退出信号"); + Ok(()) + } + }; - cleanup_alsa_config(); - return result; - } + cleanup_alsa_config(); + return result; } -#[cfg(target_os = "linux")] async fn run_master(role: ChannelRole) -> Result<()> { let config = AudioConfig { sample_rate: 48000, @@ -241,7 +225,6 @@ async fn run_master(role: ChannelRole) -> Result<()> { } } -#[cfg(target_os = "linux")] async fn run_master_audio_loop( slave_stream: &mut TcpStream, role: ChannelRole, @@ -347,7 +330,6 @@ async fn run_master_audio_loop( } } -#[cfg(target_os = "linux")] async fn run_slave(role: ChannelRole) -> Result<()> { let config = AudioConfig { sample_rate: 48000, @@ -453,7 +435,6 @@ async fn run_slave(role: ChannelRole) -> Result<()> { } } -#[cfg(target_os = "linux")] async fn run_slave_audio_loop( stream: TcpStream, config: &AudioConfig, diff --git a/apps/hello/src/bin/test.rs b/apps/hello/src/bin/test.rs index 0c3b7ba..0ee8a4d 100644 --- a/apps/hello/src/bin/test.rs +++ b/apps/hello/src/bin/test.rs @@ -1,72 +1,61 @@ -#[cfg(not(target_os = "linux"))] -use hello::audio::OpusCodec; -#[cfg(target_os = "linux")] -use hello::audio::{AudioPlayer, AudioRecorder, OpusCodec}; +#![cfg(target_os = "linux")] use anyhow::Result; +use hello::audio::{AudioPlayer, AudioRecorder, OpusCodec}; use hello::config::AudioConfig; use std::collections::VecDeque; use std::time::{Duration, Instant}; fn main() -> Result<()> { - #[cfg(not(target_os = "linux"))] - { - println!("Test utility is only supported on Linux (ALSA required)"); - return Ok(()); - } + let config = AudioConfig::default(); + println!("Starting Audio Modular Demo with 1s delay"); + println!( + "Config: capture={}, playback={}, rate={}, channels={}, frame_size={}", + config.capture_device, + config.playback_device, + config.sample_rate, + config.channels, + config.frame_size + ); - #[cfg(target_os = "linux")] - { - let config = AudioConfig::default(); - println!("Starting Audio Modular Demo with 1s delay"); - println!( - "Config: capture={}, playback={}, rate={}, channels={}, frame_size={}", - config.capture_device, - config.playback_device, - config.sample_rate, - config.channels, - config.frame_size - ); + let recorder = AudioRecorder::new(&config)?; + let player = AudioPlayer::new(&config)?; + let mut codec = OpusCodec::new(&config)?; - let recorder = AudioRecorder::new(&config)?; - let player = AudioPlayer::new(&config)?; - let mut codec = OpusCodec::new(&config)?; + // 使用队列存储编码后的 Opus 数据块及录制时间 + let mut delay_queue: VecDeque<(Instant, Vec)> = VecDeque::new(); + let delay_duration = Duration::from_secs(1); - // 使用队列存储编码后的 Opus 数据块及录制时间 - let mut delay_queue: VecDeque<(Instant, Vec)> = VecDeque::new(); - let delay_duration = Duration::from_secs(1); + let mut pcm_in = vec![0i16; config.frame_size * config.channels as usize]; + let mut opus_buf = vec![0u8; 1024]; - let mut pcm_in = vec![0i16; config.frame_size * config.channels as usize]; - let mut opus_buf = vec![0u8; 1024]; + println!("Recording... Delaying playback by 1s. Press Ctrl+C to stop."); - println!("Recording... Delaying playback by 1s. Press Ctrl+C to stop."); + loop { + // 1. 录制原始 PCM + let read = recorder.read(&mut pcm_in)?; + if read == config.frame_size { + // 2. 编码并存入队列 + let opus_len = codec.encode(&pcm_in, &mut opus_buf)?; + let opus_data = opus_buf[..opus_len].to_vec(); + delay_queue.push_back((Instant::now(), opus_data)); + } - loop { - // 1. 录制原始 PCM - let read = recorder.read(&mut pcm_in)?; - if read == config.frame_size { - // 2. 编码并存入队列 - let opus_len = codec.encode(&pcm_in, &mut opus_buf)?; - let opus_data = opus_buf[..opus_len].to_vec(); - delay_queue.push_back((Instant::now(), opus_data)); - } + // 3. 处理延迟播放：检查队列头部数据是否已等待超过 1s + while let Some((timestamp, _)) = delay_queue.front() { + if timestamp.elapsed() >= delay_duration { + let (_, opus_data) = delay_queue.pop_front().unwrap(); - // 3. 处理延迟播放：检查队列头部数据是否已等待超过 1s - while let Some((timestamp, _)) = delay_queue.front() { - if timestamp.elapsed() >= delay_duration { - let (_, opus_data) = delay_queue.pop_front().unwrap(); + let mut pcm_out = vec![0i16; config.frame_size * config.channels as usize]; + let decoded_len = codec.decode(&opus_data, &mut pcm_out)?; - let mut pcm_out = vec![0i16; config.frame_size * config.channels as usize]; - let decoded_len = codec.decode(&opus_data, &mut pcm_out)?; - - if decoded_len == config.frame_size { - // 4. 播放还原后的 PCM - player.write(&pcm_out)?; - } - } else { - // 头部数据还未到 1s，由于队列是按时间排序的，后面的肯定也没到 - break; + if decoded_len == config.frame_size { + // 4. 播放还原后的 PCM + player.write(&pcm_out)?; } + } else { + // 头部数据还未到 1s，由于队列是按时间排序的，后面的肯定也没到 + break; } } }