chore: 优化代码
This commit is contained in:
Del Wang
Del
+96
-107
@@ -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<String> = 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::<ControlPacket>(&buf[..n])
|
||||
{
|
||||
let args: Vec<String> = 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::<ControlPacket>(&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<AudioPacket> = 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::<AudioPacket>(&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<AudioPacket> = 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::<AudioPacket>(&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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String> = 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<String> = 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,
|
||||
|
||||
+42
-53
@@ -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<u8>)> = VecDeque::new();
|
||||
let delay_duration = Duration::from_secs(1);
|
||||
|
||||
// 使用队列存储编码后的 Opus 数据块及录制时间
|
||||
let mut delay_queue: VecDeque<(Instant, Vec<u8>)> = 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user