Hass/open-xiaoai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 优化推流播放与录音，支持立体声

Browse Source
This commit is contained in:
Del Wang
2026-01-06 14:06:15 +08:00
parent 0125a1478e
commit 487a5b8c14
6 changed files with 424 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/client-v2/src/app/client/mod.rs
+1
View File
@@ -279,6 +279,7 @@ impl Client {
println!("[Client] Starting playback...");
let handle = PlaybackPipeline::spawn(
config,
session.clock.clone(),
session.audio_socket.clone(),
session.cancel.clone(),
);
packages/client-v2/src/app/client/pipeline.rs
+67 -89
View File
@@ -22,13 +22,12 @@ use crate::audio::codec::OpusCodec;
use crate::audio::config::AudioConfig;
use crate::audio::player::AudioPlayer;
use crate::audio::recorder::AudioRecorder;
use crate::net::jitter_buffer::{JitterBuffer, JitterConfig};
use crate::net::network::AudioSocket;
use crate::net::protocol::AudioPacket;
use crate::net::sync::now_us;
use std::collections::VecDeque;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
@@ -36,18 +35,15 @@ use tokio_util::sync::CancellationToken;
/// 管道句柄 - 用于控制正在运行的音频管道
pub struct PipelineHandle {
cancel: CancellationToken,
/// 用于通知阻塞线程停止
stop_flag: Arc<AtomicBool>,
}
impl PipelineHandle {
fn new(cancel: CancellationToken, stop_flag: Arc<AtomicBool>) -> Self {
Self { cancel, stop_flag }
fn new(cancel: CancellationToken) -> Self {
Self { cancel }
}
/// 停止管道
pub fn stop(&self) {
self.stop_flag.store(true, Ordering::SeqCst);
self.cancel.cancel();
}
@@ -82,13 +78,12 @@ impl RecordPipeline {
parent_cancel: CancellationToken,
) -> PipelineHandle {
let cancel = parent_cancel.child_token();
let stop_flag = Arc::new(AtomicBool::new(false));
let handle = PipelineHandle::new(cancel.clone(), stop_flag.clone());
let handle = PipelineHandle::new(cancel.clone());
let token = cancel.clone();
tokio::spawn(async move {
if let Err(e) = Self::run(config, socket, target, token, stop_flag).await {
if let Err(e) = Self::run(config, socket, target, token).await {
eprintln!("[RecordPipeline] Error: {}", e);
}
});
@@ -101,15 +96,13 @@ impl RecordPipeline {
socket: Arc<AudioSocket>,
target: SocketAddr,
cancel: CancellationToken,
stop_flag: Arc<AtomicBool>,
) -> anyhow::Result<()> {
// 创建 PCM 数据通道
let (pcm_tx, mut pcm_rx) = mpsc::channel::<Vec<i16>>(32);
// 启动 ALSA 录音线程（阻塞 I/O）
let recorder_config = config.clone();
let recorder_stop = stop_flag.clone();
let channels = recorder_config.channels as usize;
std::thread::spawn(move || {
let recorder = match AudioRecorder::new(&recorder_config) {
Ok(r) => r,
@@ -119,13 +112,17 @@ impl RecordPipeline {
}
};
let mut buf = vec![0i16; recorder_config.frame_size];
let mut buf =
vec![0i16; recorder_config.frame_size * recorder_config.channels as usize];
// 使用 stop_flag 来优雅退出
while !recorder_stop.load(Ordering::SeqCst) {
loop {
match recorder.read(&mut buf) {
Ok(n) if n > 0 => {
if pcm_tx.blocking_send(buf[..n].to_vec()).is_err() {
let actual_samples = n * channels;
if pcm_tx
.blocking_send(buf[..actual_samples].to_vec())
.is_err()
{
break;
}
}
@@ -153,11 +150,11 @@ impl RecordPipeline {
pcm = pcm_rx.recv() => {
match pcm {
Some(samples) => {
if let Ok(len) = codec.encode(&samples, &mut opus_buf) {
if let Ok(encoded_len) = codec.encode(&samples, &mut opus_buf) {
let packet = AudioPacket {
seq: 0,
timestamp: 0,
data: opus_buf[..len].to_vec(),
data: opus_buf[..encoded_len].to_vec(),
};
let _ = socket.send(&packet, target).await;
}
@@ -189,17 +186,17 @@ impl PlaybackPipeline {
/// * `parent_cancel` - 父级取消令牌
pub fn spawn(
config: AudioConfig,
clock: Arc<parking_lot::Mutex<crate::net::sync::ClockSync>>,
socket: Arc<AudioSocket>,
parent_cancel: CancellationToken,
) -> PipelineHandle {
let cancel = parent_cancel.child_token();
let stop_flag = Arc::new(AtomicBool::new(false));
let handle = PipelineHandle::new(cancel.clone(), stop_flag.clone());
let handle = PipelineHandle::new(cancel.clone());
let token = cancel.clone();
tokio::spawn(async move {
if let Err(e) = Self::run(config, socket, token, stop_flag).await {
if let Err(e) = Self::run(config, clock, socket, token).await {
eprintln!("[PlaybackPipeline] Error: {}", e);
}
});
@@ -209,91 +206,72 @@ impl PlaybackPipeline {
async fn run(
config: AudioConfig,
clock: Arc<parking_lot::Mutex<crate::net::sync::ClockSync>>,
socket: Arc<AudioSocket>,
cancel: CancellationToken,
stop_flag: Arc<AtomicBool>,
) -> anyhow::Result<()> {
// 创建 PCM 数据通道
let (pcm_tx, pcm_rx) = mpsc::channel::<AudioPacket>(128);
// 1. 创建 PCM 通道
let (pcm_tx, mut pcm_rx) = mpsc::channel::<Vec<i16>>(64);
// 启动 ALSA 播放线程（阻塞 I/O）
// 2. 专用播放线程
let player_config = config.clone();
let player_stop = stop_flag.clone();
std::thread::spawn(move || {
let player = match AudioPlayer::new(&player_config) {
Ok(p) => p,
Err(e) => return eprintln!("[PlaybackPipeline] Player init error: {}", e),
};
println!("[PlaybackPipeline] Started");
// 主循环：从 UDP 接收，解码后发送给播放线程
let mut udp_buf = vec![0u8; 4096];
let mut last_time = now_us();
tokio::spawn(async move {
loop {
match socket.recv(&mut udp_buf).await {
Ok((packet, _src)) => {
let now = now_us();
let diff = now - last_time;
println!("Received packet now:{} diff:{}ms", now, diff / 1000);
last_time = now;
if let Err(e) = pcm_tx.send(packet).await {
eprintln!("[PlaybackPipeline] PCM channel send error: {}", e);
break;
}
}
Err(e) => {
eprintln!("[PlaybackPipeline] Recv error: {}", e);
}
// 当 pcm_tx 在异步任务中被 drop，这里会自动退出
while let Some(samples) = pcm_rx.blocking_recv() {
if let Err(e) = player.write(&samples) {
eprintln!("[PlaybackPipeline] Write error: {}", e);
break;
}
}
println!("[PlaybackPipeline] Player thread exited naturally");
});
let player = match AudioPlayer::new(&player_config) {
Ok(p) => p,
Err(e) => {
eprintln!("[PlaybackPipeline] Failed to create player: {}", e);
return Err(e);
}
};
// 3. Opus 解码器与 Jitter Buffer
let mut codec = OpusCodec::new(&config)?;
let mut jitter_buffer = JitterBuffer::new(JitterConfig::default());
let mut pcm_frame = vec![0i16; config.frame_size * config.channels as usize];
// 使用 blocking_recv 在线程中接收
let mut rx = pcm_rx;
// 创建 Opus 解码器
let mut codec = OpusCodec::new(&config).unwrap();
let mut pcm_buf = vec![0i16; config.frame_size * config.channels as usize];
// 提高定时精度
let mut ticker = tokio::time::interval(std::time::Duration::from_millis(5));
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
let mut jitter_buffer: VecDeque<AudioPacket> = VecDeque::new();
println!("[PlaybackPipeline] Running");
let mut start_time = 0u128;
let frame_duration_us =
(config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128;
while !player_stop.load(Ordering::SeqCst) {
while let Ok(p) = rx.try_recv() {
jitter_buffer.push_back(p);
}
if let Some(pck) = jitter_buffer.front() {
let now = now_us();
if start_time == 0 {
start_time = now;
// 4. 主逻辑循环
// 使用 loop + select，当 cancel 触发时直接 break
let mut udp_buf = vec![0u8; 4096];
loop {
tokio::select! {
// 优先级 1: 外部取消
_ = cancel.cancelled() => {
break;
}
let target_client_time = start_time + (pck.seq as u128) * frame_duration_us;
// 优先级 2: 网络接收
result = socket.recv(&mut udp_buf) => {
if let Ok((packet,_)) = result {
let arrival_time = clock.lock().to_server_time(now_us());
jitter_buffer.push(packet, arrival_time);
}
}
if now >= target_client_time {
let packet = jitter_buffer.pop_front().unwrap();
let samples = codec.decode(&packet.data, &mut pcm_buf)?;
player.write(&pcm_buf[..samples * config.channels as usize])?;
} 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;
// 优先级 3: 播放调度
_ = ticker.tick() => {
let current_time = clock.lock().to_server_time(now_us());
while let Some(packet) = jitter_buffer.pop(current_time) {
if let Ok(samples) = codec.decode(&packet.data, &mut pcm_frame) {
let samples = pcm_frame[..samples * config.channels as usize].to_vec();
if pcm_tx.try_send(samples).is_err() {
break;
}
}
}
}
} else {
tokio::time::sleep(Duration::from_millis(5)).await;
}
}
packages/client-v2/src/app/server/stream.rs
+6 -5
View File
@@ -78,7 +78,7 @@ pub struct StreamSender {
// 配置参数
input_len: usize,
frame_duration_us: u128,
max_lead_us: u128, // 允许的最大超前时间，例如 1_000_000 (1s)
max_lead_us: u128, // 允许的最大超前时间，例如 500_000 (1s)
}
impl StreamSender {
@@ -105,7 +105,7 @@ impl StreamSender {
stream_start_ts: None,
input_len,
frame_duration_us,
max_lead_us: 1_000_000,
max_lead_us: 500_000,
})
}
@@ -259,7 +259,7 @@ impl RecorderStream {
) -> anyhow::Result<()> {
let mut writer = WavWriter::create(&filename, config.sample_rate, config.channels)?;
let mut codec = OpusCodec::new(&config)?;
let mut pcm = vec![0i16; config.frame_size];
let mut pcm = vec![0i16; config.frame_size * config.channels as usize];
println!(
"[Recorder] Started -> {} (filter: {:?})",
@@ -280,8 +280,9 @@ impl RecorderStream {
}
// 解码并写入
if let Ok(n) = codec.decode(&frame.packet.data, &mut pcm) {
let _ = writer.write_samples(&pcm[..n]);
if let Ok(samples_per_channel) = codec.decode(&frame.packet.data, &mut pcm) {
let total_samples = samples_per_channel * config.channels as usize;
let _ = writer.write_samples(&pcm[..total_samples])?;
}
}
Err(broadcast::error::RecvError::Lagged(n)) => {
packages/client-v2/src/bin/server.rs
+13 -13
View File
@@ -103,13 +103,25 @@ async fn main() -> anyhow::Result<()> {
.await;
println!(" ✅ Event broadcasted");
// 5. 测试音频录制
println!("\n5️⃣ Testing Audio Recording (10 seconds)...");
match server.start_record(addr, AudioConfig::voice_16k()).await {
Ok(_) => {
println!(" ⏺️ Recording started...");
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
server.stop_record(addr).await?;
println!(" ⏹️ Recording stopped");
}
Err(e) => println!(" ❌ Recording failed: {}", e),
}
// 6. 测试音频播放（如果有测试文件）
println!("\n6️⃣ Testing Audio Playback...");
if std::path::Path::new("temp/test.wav").exists() {
match server.start_play(addr, "temp/test.wav").await {
Ok(_) => {
println!(" ▶️ Playback started...");
tokio::time::sleep(std::time::Duration::from_secs(60)).await;
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
server.stop_play(addr).await?;
println!(" ⏹️ Playback stopped");
}
@@ -119,18 +131,6 @@ async fn main() -> anyhow::Result<()> {
println!(" ⚠️ No test file found at temp/test.wav, skipping...");
}
// 5. 测试音频录制
println!("\n5️⃣ Testing Audio Recording (5 seconds)...");
match server.start_record(addr, AudioConfig::voice_16k()).await {
Ok(_) => {
println!(" ⏺️ Recording started...");
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
server.stop_record(addr).await?;
println!(" ⏹️ Recording stopped");
}
Err(e) => println!(" ❌ Recording failed: {}", e),
}
println!("\n═══════════════════════════════════════════════════════");
println!("✅ All tests completed!");
println!("\nServer status:");
packages/client-v2/src/net/jitter_buffer.rs
+334
View File
@@ -0,0 +1,334 @@
//! # Jitter Buffer - 抖动缓冲区
//!
//! 用于音频流的抖动缓冲和包重排序。
//!
//! ## 功能
//! - 自适应缓冲区大小
//! - 乱序包重排
//! - 丢包检测和统计
//! - 延迟统计
use crate::net::protocol::AudioPacket;
use std::cmp::Ordering;
use std::collections::{BTreeMap, VecDeque};
impl PartialEq for AudioPacket {
fn eq(&self, other: &Self) -> bool {
self.timestamp == other.timestamp
}
}
impl Eq for AudioPacket {}
impl Ord for AudioPacket {
fn cmp(&self, other: &Self) -> Ordering {
other.timestamp.cmp(&self.timestamp)
}
}
impl PartialOrd for AudioPacket {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
/// Jitter Buffer 统计信息
#[derive(Debug, Clone, Default)]
pub struct JitterStats {
/// 总接收包数
pub received: u64,
/// 总丢包数
pub lost: u64,
/// 总播放包数
pub played: u64,
/// 迟到的包数（到达时已过播放时间）
pub late: u64,
/// 重复包数
pub duplicate: u64,
/// 当前缓冲区大小（包数）
pub buffer_size: usize,
/// 最小延迟（微秒）
pub min_delay: u128,
/// 最大延迟（微秒）
pub max_delay: u128,
/// 平均延迟（微秒）
pub avg_delay: u128,
}
impl JitterStats {
/// 计算丢包率（百分比）
pub fn loss_rate(&self) -> f64 {
if self.received + self.lost == 0 {
return 0.0;
}
(self.lost as f64 / (self.received + self.lost) as f64) * 100.0
}
}
/// Jitter Buffer 配置
#[derive(Debug, Clone)]
pub struct JitterConfig {
/// 最小缓冲区大小（包数）
pub min_buffer_size: usize,
/// 最大缓冲区大小（包数）
pub max_buffer_size: usize,
/// 目标缓冲区大小（包数）
pub target_buffer_size: usize,
/// 自适应调整间隔（包数）
pub adapt_interval: usize,
/// 最大容忍延迟（微秒）
pub max_tolerable_delay: u128,
}
impl Default for JitterConfig {
fn default() -> Self {
Self {
min_buffer_size: 2,
max_buffer_size: 20,
target_buffer_size: 5,
adapt_interval: 50,
max_tolerable_delay: 100_000, // 100ms
}
}
}
/// Jitter Buffer - 用于音频流的抖动缓冲
pub struct JitterBuffer {
/// 配置
config: JitterConfig,
/// 缓冲区（按时间戳排序）
buffer: BTreeMap<u128, AudioPacket>,
/// 统计信息
stats: JitterStats,
/// 期望的下一个序列号
expected_seq: u32,
/// 是否已接收第一个包
first_packet_received: bool,
/// 延迟样本窗口（用于自适应）
delay_samples: VecDeque<u128>,
/// 自适应计数器
adapt_counter: usize,
/// 上次播放的时间戳
last_played_timestamp: u128,
}
impl JitterBuffer {
/// 创建新的 Jitter Buffer
pub fn new(config: JitterConfig) -> Self {
Self {
config,
buffer: BTreeMap::new(),
stats: JitterStats::default(),
expected_seq: 0,
first_packet_received: false,
delay_samples: VecDeque::with_capacity(100),
adapt_counter: 0,
last_played_timestamp: 0,
}
}
/// 使用默认配置创建
pub fn default() -> Self {
Self::new(JitterConfig::default())
}
/// 插入音频包
pub fn push(&mut self, packet: AudioPacket, arrival_time: u128) {
// 检测重复包
if self.buffer.contains_key(&packet.timestamp) {
self.stats.duplicate += 1;
return;
}
// 初始化序列号
if !self.first_packet_received {
self.expected_seq = packet.seq.wrapping_add(1);
self.first_packet_received = true;
} else {
// 检测丢包
let seq_diff = packet.seq.wrapping_sub(self.expected_seq);
if seq_diff > 0 && seq_diff < 1000 {
// 允许一定的序列号跳跃（处理回环）
self.stats.lost += seq_diff as u64;
}
self.expected_seq = packet.seq.wrapping_add(1);
}
// 检查是否迟到
if self.last_played_timestamp > 0 && packet.timestamp < self.last_played_timestamp {
self.stats.late += 1;
return;
}
// 超过最大容忍延迟，直接丢弃
if packet.timestamp < arrival_time
&& arrival_time - packet.timestamp > self.config.max_tolerable_delay
{
self.stats.late += 1;
return;
}
let delay = if packet.timestamp >= arrival_time {
packet.timestamp - arrival_time
} else {
0 // 包到达时已经过了播放时间
};
self.update_delay_stats(delay);
// 插入缓冲区
self.buffer.insert(packet.timestamp, packet);
self.stats.received += 1;
self.stats.buffer_size = self.buffer.len();
// 自适应调整
self.adapt_counter += 1;
if self.adapt_counter >= self.config.adapt_interval {
self.adapt_buffer_size();
self.adapt_counter = 0;
}
}
/// 获取下一个应该播放的包
///
/// # Arguments
/// * `current_time` - 当前服务器时间
///
/// # Returns
/// * `Some(packet)` - 如果有包应该播放
/// * `None` - 如果缓冲区为空或还没到播放时间
pub fn pop(&mut self, current_time: u128) -> Option<AudioPacket> {
// 如果缓冲区为空，直接返回
if self.buffer.is_empty() {
return None;
}
// 检查缓冲区是否达到最小大小
if self.buffer.len() < self.config.target_buffer_size && !self.should_drain() {
return None;
}
// 获取最早的包
if let Some((&timestamp, _)) = self.buffer.iter().next() {
// 检查是否到达播放时间
if current_time >= timestamp {
let packet = self.buffer.remove(&timestamp).unwrap();
self.last_played_timestamp = timestamp;
self.stats.played += 1;
self.stats.buffer_size = self.buffer.len();
return Some(packet);
}
}
None
}
/// 强制获取下一个包（无论时间）
pub fn pop_next(&mut self) -> Option<AudioPacket> {
if let Some((&timestamp, _)) = self.buffer.iter().next() {
let packet = self.buffer.remove(&timestamp).unwrap();
self.last_played_timestamp = timestamp;
self.stats.played += 1;
self.stats.buffer_size = self.buffer.len();
return Some(packet);
}
None
}
/// 查看下一个包的播放时间（不移除）
pub fn peek_next_timestamp(&self) -> Option<u128> {
self.buffer.keys().next().copied()
}
/// 获取缓冲区大小
pub fn len(&self) -> usize {
self.buffer.len()
}
/// 检查缓冲区是否为空
pub fn is_empty(&self) -> bool {
self.buffer.is_empty()
}
/// 获取统计信息
pub fn stats(&self) -> &JitterStats {
&self.stats
}
/// 重置统计信息
pub fn reset_stats(&mut self) {
self.stats = JitterStats::default();
self.stats.buffer_size = self.buffer.len();
}
/// 清空缓冲区
pub fn clear(&mut self) {
self.buffer.clear();
self.stats.buffer_size = 0;
self.first_packet_received = false;
self.last_played_timestamp = 0;
}
/// 检查是否应该排空缓冲区（处理长时间没有新包的情况）
fn should_drain(&self) -> bool {
// 如果缓冲区有包且已经等了很久，就开始播放
!self.buffer.is_empty() && self.buffer.len() >= self.config.min_buffer_size
}
/// 更新延迟统计
fn update_delay_stats(&mut self, delay: u128) {
// 更新最小/最大延迟
if self.stats.received == 0 {
self.stats.min_delay = delay;
self.stats.max_delay = delay;
self.stats.avg_delay = delay;
} else {
self.stats.min_delay = self.stats.min_delay.min(delay);
self.stats.max_delay = self.stats.max_delay.max(delay);
// 滑动平均
self.stats.avg_delay = (self.stats.avg_delay * 9 + delay) / 10;
}
// 保存延迟样本用于自适应
self.delay_samples.push_back(delay);
if self.delay_samples.len() > 100 {
self.delay_samples.pop_front();
}
}
/// 自适应调整缓冲区大小
fn adapt_buffer_size(&mut self) {
if self.delay_samples.len() < 10 {
return;
}
// 计算延迟方差（抖动）
let avg = self.stats.avg_delay;
let variance: f64 = self
.delay_samples
.iter()
.map(|&d| {
let diff = d as i128 - avg as i128;
(diff * diff) as f64
})
.sum::<f64>()
/ self.delay_samples.len() as f64;
let jitter = variance.sqrt();
// 根据抖动调整目标缓冲区大小
// 抖动大 -> 增加缓冲区
// 抖动小 -> 减少缓冲区
let target = if jitter > 50_000.0 {
// 高抖动（>50ms 标准差）
self.config.target_buffer_size + 2
} else if jitter < 10_000.0 {
// 低抖动（<10ms 标准差）
self.config.target_buffer_size.saturating_sub(1)
} else {
self.config.target_buffer_size
};
// 限制在最小/最大范围内
self.config.target_buffer_size = target
.max(self.config.min_buffer_size)
.min(self.config.max_buffer_size);
}
}
packages/client-v2/src/net/mod.rs
+3
View File
@@ -7,10 +7,13 @@
//! - `network` - 底层网络连接
//! - `protocol` - 通信协议定义
//! - `rpc` - RPC 调用管理
//! - `sync` - 时间同步工具
//! - `jitter_buffer` - 抖动缓冲区实现
pub mod command;
pub mod discovery;
pub mod event;
pub mod jitter_buffer;
pub mod network;
pub mod protocol;
pub mod rpc;