fix: 修复服务端推流播放异常的问题
This commit is contained in:
Del Wang
Generated
+188
-2
@@ -9,7 +9,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "812947049edcd670a82cd5c73c3661d2e58468577ba8489de58e1a73c04cbd5d"
|
||||
dependencies = [
|
||||
"alsa-sys",
|
||||
"bitflags",
|
||||
"bitflags 2.10.0",
|
||||
"cfg-if",
|
||||
"libc",
|
||||
]
|
||||
@@ -30,6 +30,12 @@ version = "1.0.100"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61"
|
||||
|
||||
[[package]]
|
||||
name = "arrayvec"
|
||||
version = "0.7.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
|
||||
|
||||
[[package]]
|
||||
name = "atomic-polyfill"
|
||||
version = "1.0.3"
|
||||
@@ -50,12 +56,24 @@ dependencies = [
|
||||
"pkg-config",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bitflags"
|
||||
version = "1.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
|
||||
|
||||
[[package]]
|
||||
name = "bitflags"
|
||||
version = "2.10.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3"
|
||||
|
||||
[[package]]
|
||||
name = "bytemuck"
|
||||
version = "1.24.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1fbdf580320f38b612e485521afda1ee26d10cc9884efaaa750d383e13e3c5f4"
|
||||
|
||||
[[package]]
|
||||
name = "byteorder"
|
||||
version = "1.5.0"
|
||||
@@ -140,6 +158,15 @@ version = "0.6.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d"
|
||||
|
||||
[[package]]
|
||||
name = "encoding_rs"
|
||||
version = "0.8.35"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "errno"
|
||||
version = "0.3.14"
|
||||
@@ -150,6 +177,12 @@ dependencies = [
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "extended"
|
||||
version = "0.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "af9673d8203fcb076b19dfd17e38b3d4ae9f44959416ea532ce72415a6020365"
|
||||
|
||||
[[package]]
|
||||
name = "find-msvc-tools"
|
||||
version = "0.1.6"
|
||||
@@ -240,6 +273,12 @@ dependencies = [
|
||||
"stable_deref_trait",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "lazy_static"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
|
||||
|
||||
[[package]]
|
||||
name = "libc"
|
||||
version = "0.2.178"
|
||||
@@ -366,7 +405,7 @@ version = "0.5.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
|
||||
dependencies = [
|
||||
"bitflags",
|
||||
"bitflags 2.10.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -473,6 +512,152 @@ version = "1.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
|
||||
|
||||
[[package]]
|
||||
name = "symphonia"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5773a4c030a19d9bfaa090f49746ff35c75dfddfa700df7a5939d5e076a57039"
|
||||
dependencies = [
|
||||
"lazy_static",
|
||||
"symphonia-bundle-flac",
|
||||
"symphonia-bundle-mp3",
|
||||
"symphonia-codec-adpcm",
|
||||
"symphonia-codec-pcm",
|
||||
"symphonia-codec-vorbis",
|
||||
"symphonia-core",
|
||||
"symphonia-format-mkv",
|
||||
"symphonia-format-ogg",
|
||||
"symphonia-format-riff",
|
||||
"symphonia-metadata",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-bundle-flac"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c91565e180aea25d9b80a910c546802526ffd0072d0b8974e3ebe59b686c9976"
|
||||
dependencies = [
|
||||
"log",
|
||||
"symphonia-core",
|
||||
"symphonia-metadata",
|
||||
"symphonia-utils-xiph",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-bundle-mp3"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4872dd6bb56bf5eac799e3e957aa1981086c3e613b27e0ac23b176054f7c57ed"
|
||||
dependencies = [
|
||||
"lazy_static",
|
||||
"log",
|
||||
"symphonia-core",
|
||||
"symphonia-metadata",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-codec-adpcm"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2dddc50e2bbea4cfe027441eece77c46b9f319748605ab8f3443350129ddd07f"
|
||||
dependencies = [
|
||||
"log",
|
||||
"symphonia-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-codec-pcm"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4e89d716c01541ad3ebe7c91ce4c8d38a7cf266a3f7b2f090b108fb0cb031d95"
|
||||
dependencies = [
|
||||
"log",
|
||||
"symphonia-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-codec-vorbis"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f025837c309cd69ffef572750b4a2257b59552c5399a5e49707cc5b1b85d1c73"
|
||||
dependencies = [
|
||||
"log",
|
||||
"symphonia-core",
|
||||
"symphonia-utils-xiph",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-core"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ea00cc4f79b7f6bb7ff87eddc065a1066f3a43fe1875979056672c9ef948c2af"
|
||||
dependencies = [
|
||||
"arrayvec",
|
||||
"bitflags 1.3.2",
|
||||
"bytemuck",
|
||||
"lazy_static",
|
||||
"log",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-format-mkv"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "122d786d2c43a49beb6f397551b4a050d8229eaa54c7ddf9ee4b98899b8742d0"
|
||||
dependencies = [
|
||||
"lazy_static",
|
||||
"log",
|
||||
"symphonia-core",
|
||||
"symphonia-metadata",
|
||||
"symphonia-utils-xiph",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-format-ogg"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2b4955c67c1ed3aa8ae8428d04ca8397fbef6a19b2b051e73b5da8b1435639cb"
|
||||
dependencies = [
|
||||
"log",
|
||||
"symphonia-core",
|
||||
"symphonia-metadata",
|
||||
"symphonia-utils-xiph",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-format-riff"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c2d7c3df0e7d94efb68401d81906eae73c02b40d5ec1a141962c592d0f11a96f"
|
||||
dependencies = [
|
||||
"extended",
|
||||
"log",
|
||||
"symphonia-core",
|
||||
"symphonia-metadata",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-metadata"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "36306ff42b9ffe6e5afc99d49e121e0bd62fe79b9db7b9681d48e29fa19e6b16"
|
||||
dependencies = [
|
||||
"encoding_rs",
|
||||
"lazy_static",
|
||||
"log",
|
||||
"symphonia-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "symphonia-utils-xiph"
|
||||
version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ee27c85ab799a338446b68eec77abf42e1a6f1bb490656e121c6e27bfbab9f16"
|
||||
dependencies = [
|
||||
"symphonia-core",
|
||||
"symphonia-metadata",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "syn"
|
||||
version = "2.0.111"
|
||||
@@ -661,6 +846,7 @@ dependencies = [
|
||||
"parking_lot",
|
||||
"postcard",
|
||||
"serde",
|
||||
"symphonia",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
]
|
||||
|
||||
@@ -11,10 +11,6 @@ panic = "abort"
|
||||
strip = true
|
||||
debug = false
|
||||
|
||||
[features]
|
||||
default = []
|
||||
app = []
|
||||
|
||||
[dependencies]
|
||||
opus = "0.3"
|
||||
anyhow = "1.0"
|
||||
@@ -27,3 +23,6 @@ dashmap = "6.1.0"
|
||||
|
||||
[target.'cfg(target_os = "linux")'.dependencies]
|
||||
alsa = "0.11"
|
||||
|
||||
[target.'cfg(not(target_os = "linux"))'.dependencies]
|
||||
symphonia = { version = "0.5", features = ["mp3", "wav", "pcm"] }
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
build-server:
|
||||
cargo build --release --bin server --features app
|
||||
cargo build --release --bin server
|
||||
|
||||
build-client:
|
||||
docker run --rm -v $(shell pwd):/app idootop/open-xiaoai-runtime:oh2p \
|
||||
cargo build --target armv7-unknown-linux-gnueabihf --release --bin client --features app
|
||||
cargo build --target armv7-unknown-linux-gnueabihf --release --bin client
|
||||
|
||||
run-server:
|
||||
target/release/server
|
||||
|
||||
@@ -25,8 +25,6 @@
|
||||
//! └─────────────────────────────────────┘
|
||||
//! ```
|
||||
|
||||
#![cfg(target_os = "linux")]
|
||||
|
||||
mod pipeline;
|
||||
mod session;
|
||||
|
||||
@@ -41,6 +39,7 @@ use crate::net::discovery::Discovery;
|
||||
use crate::net::event::{ClientEvent, NotificationLevel, ServerEvent};
|
||||
use crate::net::network::{AudioSocket, Connection};
|
||||
use crate::net::protocol::ControlPacket;
|
||||
use crate::net::sync::now_us;
|
||||
use anyhow::{Result, anyhow};
|
||||
use session::handshake;
|
||||
use std::net::SocketAddr;
|
||||
@@ -57,8 +56,8 @@ pub struct ClientConfig {
|
||||
pub client_auth: String,
|
||||
/// 服务端认证
|
||||
pub server_auth: String,
|
||||
/// 心跳间隔（秒）
|
||||
pub heartbeat_interval: u64,
|
||||
/// 心跳间隔（毫秒）
|
||||
pub heartbeat_ms: u64,
|
||||
/// 连接超时（秒）
|
||||
pub timeout: u64,
|
||||
/// 客户端型号
|
||||
@@ -75,7 +74,7 @@ impl Default for ClientConfig {
|
||||
.unwrap_or_else(|_| "xiao-server".to_string()),
|
||||
client_auth: std::env::var("XIAO_CLIENT_AUTH")
|
||||
.unwrap_or_else(|_| "xiao-client".to_string()),
|
||||
heartbeat_interval: 10,
|
||||
heartbeat_ms: 200,
|
||||
timeout: 60,
|
||||
// todo 获取设备信息
|
||||
model: "Open-XiaoAi-V2".to_string(),
|
||||
@@ -192,17 +191,21 @@ impl Client {
|
||||
|
||||
/// 启动心跳任务
|
||||
fn spawn_heartbeat(&self, session: Arc<Session>) {
|
||||
let interval = std::time::Duration::from_secs(self.config.heartbeat_interval);
|
||||
let interval = std::time::Duration::from_millis(self.config.heartbeat_ms);
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut seq = 0;
|
||||
let mut ticker = tokio::time::interval(interval);
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = session.cancel.cancelled() => break,
|
||||
_ = ticker.tick() => {
|
||||
if session.send(&ControlPacket::Ping).await.is_err() {
|
||||
let t1 = now_us();
|
||||
let msg = ControlPacket::Ping { client_ts: t1, seq };
|
||||
if session.send(&msg).await.is_err() {
|
||||
break;
|
||||
}
|
||||
seq += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -238,10 +241,14 @@ impl Client {
|
||||
/// 处理控制包
|
||||
async fn handle_packet(&self, session: &Arc<Session>, packet: ControlPacket) -> Result<()> {
|
||||
match packet {
|
||||
ControlPacket::Ping => {
|
||||
session.send(&ControlPacket::Pong).await?;
|
||||
ControlPacket::Pong {
|
||||
client_ts,
|
||||
server_ts,
|
||||
..
|
||||
} => {
|
||||
let t4 = now_us();
|
||||
session.update_clock(client_ts, server_ts, t4);
|
||||
}
|
||||
ControlPacket::Pong => {}
|
||||
ControlPacket::RpcResponse { id, result } => {
|
||||
session.resolve_rpc(id, result);
|
||||
}
|
||||
|
||||
@@ -24,9 +24,12 @@ use crate::audio::player::AudioPlayer;
|
||||
use crate::audio::recorder::AudioRecorder;
|
||||
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::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::Duration;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
@@ -152,6 +155,8 @@ impl RecordPipeline {
|
||||
Some(samples) => {
|
||||
if let Ok(len) = codec.encode(&samples, &mut opus_buf) {
|
||||
let packet = AudioPacket {
|
||||
seq: 0,
|
||||
timestamp: 0,
|
||||
data: opus_buf[..len].to_vec(),
|
||||
};
|
||||
let _ = socket.send(&packet, target).await;
|
||||
@@ -209,63 +214,86 @@ impl PlaybackPipeline {
|
||||
stop_flag: Arc<AtomicBool>,
|
||||
) -> anyhow::Result<()> {
|
||||
// 创建 PCM 数据通道
|
||||
let (pcm_tx, pcm_rx) = mpsc::channel::<Vec<i16>>(32);
|
||||
let (pcm_tx, pcm_rx) = mpsc::channel::<AudioPacket>(128);
|
||||
|
||||
// 启动 ALSA 播放线程（阻塞 I/O）
|
||||
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) => {
|
||||
eprintln!("[PlaybackPipeline] Failed to create player: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
println!("[PlaybackPipeline] Started");
|
||||
|
||||
// 使用 blocking_recv 在线程中接收
|
||||
let mut rx = pcm_rx;
|
||||
while !player_stop.load(Ordering::SeqCst) {
|
||||
match rx.blocking_recv() {
|
||||
Some(samples) => {
|
||||
if let Err(e) = player.write(&samples) {
|
||||
eprintln!("[PlaybackPipeline] Write error: {}", e);
|
||||
// 主循环：从 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;
|
||||
}
|
||||
}
|
||||
None => {
|
||||
// 通道关闭
|
||||
break;
|
||||
Err(e) => {
|
||||
eprintln!("[PlaybackPipeline] Recv error: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let player = match AudioPlayer::new(&player_config) {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
eprintln!("[PlaybackPipeline] Failed to create player: {}", e);
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
|
||||
// 使用 blocking_recv 在线程中接收
|
||||
let mut rx = pcm_rx;
|
||||
// 创建 Opus 解码器
|
||||
let mut codec = OpusCodec::new(&config)?;
|
||||
let mut pcm_buf = vec![0i16; config.frame_size];
|
||||
let mut udp_buf = vec![0u8; 4096];
|
||||
let mut codec = OpusCodec::new(&config).unwrap();
|
||||
let mut pcm_buf = vec![0i16; config.frame_size * config.channels as usize];
|
||||
|
||||
println!("[PlaybackPipeline] Started");
|
||||
let mut jitter_buffer: VecDeque<AudioPacket> = VecDeque::new();
|
||||
|
||||
// 主循环：从 UDP 接收，解码后发送给播放线程
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = cancel.cancelled() => break,
|
||||
result = socket.recv(&mut udp_buf) => {
|
||||
match result {
|
||||
Ok((packet, _src)) => {
|
||||
if let Ok(n) = codec.decode(&packet.data, &mut pcm_buf) {
|
||||
// 使用 try_send 避免阻塞
|
||||
let _ = pcm_tx.try_send(pcm_buf[..n].to_vec());
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("[PlaybackPipeline] Recv error: {}", e);
|
||||
}
|
||||
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;
|
||||
}
|
||||
|
||||
let target_client_time = start_time + (pck.seq as u128) * frame_duration_us;
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
tokio::time::sleep(Duration::from_millis(5)).await;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -273,4 +301,3 @@ impl PlaybackPipeline {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@ use crate::net::command::{Command, CommandResult};
|
||||
use crate::net::network::{AudioSocket, Connection};
|
||||
use crate::net::protocol::{ClientInfo, ControlPacket};
|
||||
use crate::net::rpc::RpcManager;
|
||||
use crate::net::sync::ClockSync;
|
||||
use anyhow::{Context, Result, anyhow};
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
@@ -94,6 +95,9 @@ pub struct Session {
|
||||
/// 会话取消令牌
|
||||
pub cancel: CancellationToken,
|
||||
|
||||
/// 时间同步
|
||||
pub clock: Arc<parking_lot::Mutex<ClockSync>>,
|
||||
|
||||
/// 活动管道
|
||||
pipelines: parking_lot::Mutex<ActivePipelines>,
|
||||
|
||||
@@ -118,6 +122,7 @@ impl Session {
|
||||
server_audio_addr,
|
||||
rpc: Arc::new(RpcManager::new()),
|
||||
cancel,
|
||||
clock: Arc::new(parking_lot::Mutex::new(ClockSync::new(100))),
|
||||
pipelines: parking_lot::Mutex::new(ActivePipelines::default()),
|
||||
created_at: std::time::Instant::now(),
|
||||
volume: parking_lot::Mutex::new(100),
|
||||
@@ -144,6 +149,13 @@ impl Session {
|
||||
self.conn.recv().await
|
||||
}
|
||||
|
||||
/// 更新时间
|
||||
pub fn update_clock(&self, client_send_ts: u128, server_ts: u128, client_recv_ts: u128) {
|
||||
self.clock
|
||||
.lock()
|
||||
.update(client_send_ts, server_ts, client_recv_ts);
|
||||
}
|
||||
|
||||
/// 发起 RPC 调用（新版）
|
||||
pub async fn execute(&self, command: Command) -> Result<CommandResult> {
|
||||
let (id, rx) = self.rpc.register();
|
||||
|
||||
@@ -1,3 +1,2 @@
|
||||
#[cfg(target_os = "linux")]
|
||||
pub mod client;
|
||||
pub mod server;
|
||||
|
||||
@@ -51,6 +51,7 @@ use crate::net::discovery::Discovery;
|
||||
use crate::net::event::{ClientEvent, ServerEvent, ServerEventBus};
|
||||
use crate::net::network::Connection;
|
||||
use crate::net::protocol::ControlPacket;
|
||||
use crate::net::sync::now_us;
|
||||
use anyhow::{Context, Result, anyhow};
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
@@ -305,10 +306,14 @@ impl Server {
|
||||
/// 处理控制包
|
||||
async fn handle_packet(&self, session: &Arc<Session>, packet: ControlPacket) -> Result<()> {
|
||||
match packet {
|
||||
ControlPacket::Ping => {
|
||||
session.send(&ControlPacket::Pong).await?;
|
||||
ControlPacket::Ping { client_ts, seq } => {
|
||||
let pong = ControlPacket::Pong {
|
||||
client_ts,
|
||||
server_ts: now_us(),
|
||||
seq,
|
||||
};
|
||||
session.send(&pong).await?;
|
||||
}
|
||||
ControlPacket::Pong => {}
|
||||
ControlPacket::RpcResponse { id, result } => {
|
||||
session.resolve_rpc(id, result);
|
||||
}
|
||||
|
||||
@@ -25,6 +25,7 @@ use crate::audio::config::AudioConfig;
|
||||
use crate::audio::wav::{WavReader, WavWriter};
|
||||
use crate::net::network::AudioSocket;
|
||||
use crate::net::protocol::AudioPacket;
|
||||
use crate::net::sync::now_us;
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::broadcast;
|
||||
@@ -98,33 +99,78 @@ impl FilePlaybackStream {
|
||||
target: SocketAddr,
|
||||
cancel: CancellationToken,
|
||||
) -> anyhow::Result<()> {
|
||||
let mut codec = OpusCodec::new(&config)?;
|
||||
let mut pcm = vec![0i16; config.frame_size];
|
||||
let mut opus_buf = vec![0u8; 4096];
|
||||
let frame_duration = std::time::Duration::from_millis(20);
|
||||
let mut interval = tokio::time::interval(frame_duration);
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
{
|
||||
use crate::audio::reader::AudioReader;
|
||||
let mut codec = OpusCodec::new(&config)?;
|
||||
let mut pcm = vec![0i16; config.frame_size * config.channels as usize];
|
||||
let mut opus_buf = vec![0u8; 4096];
|
||||
|
||||
println!("[FilePlayback] Started -> {}", target);
|
||||
println!("[FilePlayback] Started -> {}", target);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = cancel.cancelled() => break,
|
||||
_ = interval.tick() => {
|
||||
match reader.read_samples(&mut pcm) {
|
||||
Ok(0) => {
|
||||
println!("[FilePlayback] EOF reached");
|
||||
break;
|
||||
}
|
||||
Ok(n) => {
|
||||
if let Ok(len) = codec.encode(&pcm[..n], &mut opus_buf) {
|
||||
let mut seq = 0u32;
|
||||
let delay_us = 0_000; // 100ms 基础延迟
|
||||
|
||||
let mut stream_start_ts = 0;
|
||||
let frame_duration_us =
|
||||
(config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128;
|
||||
|
||||
let mut reader = AudioReader::new("temp/test.wav")?;
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = cancel.cancelled() => break,
|
||||
result = async{
|
||||
if let Some((left_pcm, right_pcm)) = reader.read_chunk(config.frame_size)?{
|
||||
let actual_len = left_pcm.len();
|
||||
|
||||
// 1. 先将 pcm 缓冲区清零（处理尾帧时的静音填充）
|
||||
pcm.fill(0);
|
||||
|
||||
// 2. 只循环实际读取到的长度
|
||||
if config.channels == 2 {
|
||||
for i in 0..actual_len {
|
||||
pcm[i * 2] = left_pcm[i];
|
||||
pcm[i * 2 + 1] = right_pcm[i];
|
||||
}
|
||||
} else {
|
||||
pcm[..actual_len].copy_from_slice(&left_pcm);
|
||||
}
|
||||
|
||||
// 3. 编码时依然使用固定的 frame_size
|
||||
let input_len = config.frame_size * config.channels as usize;
|
||||
if let Ok(len) = codec.encode(&pcm[..input_len], &mut opus_buf) {
|
||||
let now = now_us();
|
||||
if stream_start_ts == 0 {
|
||||
// 初始化流开始时间戳
|
||||
stream_start_ts = now;
|
||||
}
|
||||
let target_ts = stream_start_ts + ((seq) as u128 * frame_duration_us) + delay_us;
|
||||
let packet = AudioPacket {
|
||||
seq,
|
||||
timestamp: target_ts,
|
||||
data: opus_buf[..len].to_vec(),
|
||||
};
|
||||
|
||||
let _ = socket.send(&packet, target).await;
|
||||
seq += 1;
|
||||
|
||||
// 音频发送时长超过音频播放 1s 时进行等待（控制数据超前缓冲 1s）
|
||||
if target_ts > now + 1_000_000 {
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
|
||||
}
|
||||
|
||||
return Ok(())
|
||||
}
|
||||
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("[FilePlayback] Read error: {}", e);
|
||||
Err(anyhow::anyhow!("Unexpected EOF"))
|
||||
} => {
|
||||
if let Err(e) = result {
|
||||
if e.to_string() == "EOF" {
|
||||
break;
|
||||
}
|
||||
eprintln!("[FilePlayback] Error: {}", e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
@@ -299,4 +345,3 @@ impl ForwardStream {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -23,7 +23,7 @@ impl AudioConfig {
|
||||
pub fn voice_16k() -> Self {
|
||||
Self {
|
||||
capture_device: "plug:Capture".to_string(),
|
||||
playback_device: "plug:default".to_string(),
|
||||
playback_device: "default".to_string(),
|
||||
sample_rate: 16_000,
|
||||
channels: 1,
|
||||
frame_size: 320, // 20ms
|
||||
@@ -37,7 +37,7 @@ impl AudioConfig {
|
||||
pub fn music_48k() -> Self {
|
||||
Self {
|
||||
capture_device: "plug:Capture".to_string(),
|
||||
playback_device: "plug:default".to_string(),
|
||||
playback_device: "default".to_string(),
|
||||
sample_rate: 48_000,
|
||||
channels: 2,
|
||||
frame_size: 960, // 20ms
|
||||
|
||||
@@ -3,3 +3,4 @@ pub mod config;
|
||||
pub mod player;
|
||||
pub mod recorder;
|
||||
pub mod wav;
|
||||
pub mod reader;
|
||||
@@ -1,44 +1,58 @@
|
||||
#![cfg(target_os = "linux")]
|
||||
|
||||
use crate::audio::config::AudioConfig;
|
||||
use alsa::Direction;
|
||||
use alsa::pcm::{Access, Format, HwParams, PCM};
|
||||
use anyhow::{Context, Result};
|
||||
use anyhow::{Context, Result, anyhow};
|
||||
|
||||
pub struct AudioPlayer {
|
||||
pcm: PCM,
|
||||
#[cfg(target_os = "linux")]
|
||||
pcm: alsa::pcm::PCM,
|
||||
}
|
||||
|
||||
impl AudioPlayer {
|
||||
pub fn new(config: &AudioConfig) -> Result<Self> {
|
||||
let pcm = PCM::new(&config.playback_device, Direction::Playback, false)
|
||||
.context("Failed to open playback PCM device")?;
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
let hwp = HwParams::any(&pcm).context("Failed to get HwParams")?;
|
||||
hwp.set_access(Access::RWInterleaved)?;
|
||||
hwp.set_format(Format::s16())?;
|
||||
hwp.set_rate_near(config.sample_rate, alsa::ValueOr::Nearest)?;
|
||||
hwp.set_channels(config.channels as u32)?;
|
||||
pcm.hw_params(&hwp)?;
|
||||
use alsa::Direction;
|
||||
use alsa::pcm::{Access, Format, HwParams, PCM};
|
||||
|
||||
let pcm = PCM::new(&config.playback_device, Direction::Playback, false)
|
||||
.context("Failed to open playback PCM device")?;
|
||||
{
|
||||
let hwp = HwParams::any(&pcm).context("Failed to get HwParams")?;
|
||||
hwp.set_access(Access::RWInterleaved)?;
|
||||
hwp.set_format(Format::s16())?;
|
||||
hwp.set_rate(config.sample_rate, alsa::ValueOr::Nearest)?;
|
||||
hwp.set_channels(config.channels as u32)?;
|
||||
pcm.hw_params(&hwp)?;
|
||||
pcm.prepare().context("Failed to prepare PCM")?;
|
||||
}
|
||||
Ok(Self { pcm })
|
||||
}
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
{
|
||||
Err(anyhow!("Linux Only"))
|
||||
}
|
||||
pcm.prepare().context("Failed to prepare PCM")?;
|
||||
Ok(Self { pcm })
|
||||
}
|
||||
|
||||
pub fn write(&self, buf: &[i16]) -> Result<usize> {
|
||||
let res = self.pcm.io_i16()?.writei(buf);
|
||||
match res {
|
||||
Ok(n) => Ok(n),
|
||||
Err(e) if e.errno() == 32 => {
|
||||
println!("ALSA write underrun, preparing PCM");
|
||||
// Broken pipe (underrun)
|
||||
self.pcm.prepare()?;
|
||||
self.pcm
|
||||
.io_i16()?
|
||||
.writei(buf)
|
||||
.context("ALSA write retry failed")
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
let res = self.pcm.io_i16()?.writei(buf);
|
||||
match res {
|
||||
Ok(n) => Ok(n),
|
||||
Err(e) if e.errno() == 32 => {
|
||||
println!("ALSA write underrun, preparing PCM");
|
||||
// Broken pipe (underrun)
|
||||
self.pcm.prepare()?;
|
||||
self.pcm
|
||||
.io_i16()?
|
||||
.writei(buf)
|
||||
.context("ALSA write retry failed")
|
||||
}
|
||||
Err(e) => Err(e.into()),
|
||||
}
|
||||
Err(e) => Err(e.into()),
|
||||
}
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
{
|
||||
Err(anyhow!("Linux Only"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
#![cfg(not(target_os = "linux"))]
|
||||
use anyhow::{Context, Result};
|
||||
use std::fs::File;
|
||||
use std::path::Path;
|
||||
use symphonia::core::audio::SampleBuffer;
|
||||
use symphonia::core::codecs::{Decoder, DecoderOptions};
|
||||
use symphonia::core::formats::{FormatOptions, FormatReader};
|
||||
use symphonia::core::io::MediaSourceStream;
|
||||
use symphonia::core::meta::MetadataOptions;
|
||||
use symphonia::core::probe::Hint;
|
||||
|
||||
pub struct AudioReader {
|
||||
format: Box<dyn FormatReader>,
|
||||
decoder: Box<dyn Decoder>,
|
||||
track_id: u32,
|
||||
sample_buf: Option<SampleBuffer<i16>>,
|
||||
channels: usize,
|
||||
left_buffer: Vec<i16>,
|
||||
right_buffer: Vec<i16>,
|
||||
}
|
||||
|
||||
impl AudioReader {
|
||||
pub fn new(path: &str) -> Result<Self> {
|
||||
let src =
|
||||
File::open(Path::new(path)).context(format!("Failed to open audio file: {}", path))?;
|
||||
let mss = MediaSourceStream::new(Box::new(src), Default::default());
|
||||
|
||||
let mut hint = Hint::new();
|
||||
if path.ends_with(".wav") {
|
||||
hint.with_extension("wav");
|
||||
} else if path.ends_with(".mp3") {
|
||||
hint.with_extension("mp3");
|
||||
}
|
||||
|
||||
let probed = symphonia::default::get_probe()
|
||||
.format(
|
||||
&hint,
|
||||
mss,
|
||||
&FormatOptions::default(),
|
||||
&MetadataOptions::default(),
|
||||
)
|
||||
.context("Failed to probe audio format")?;
|
||||
|
||||
let format = probed.format;
|
||||
let track = format
|
||||
.tracks()
|
||||
.iter()
|
||||
.find(|t| t.codec_params.codec != symphonia::core::codecs::CODEC_TYPE_NULL)
|
||||
.context("No supported audio track found")?;
|
||||
|
||||
let track_id = track.id;
|
||||
let decoder = symphonia::default::get_codecs()
|
||||
.make(&track.codec_params, &DecoderOptions::default())
|
||||
.context("Failed to create decoder")?;
|
||||
|
||||
let channels = track.codec_params.channels.map(|c| c.count()).unwrap_or(1);
|
||||
|
||||
Ok(Self {
|
||||
format,
|
||||
decoder,
|
||||
track_id,
|
||||
sample_buf: None,
|
||||
channels,
|
||||
left_buffer: Vec::new(),
|
||||
right_buffer: Vec::new(),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn read_chunk(&mut self, chunk_size: usize) -> Result<Option<(Vec<i16>, Vec<i16>)>> {
|
||||
while self.left_buffer.len() < chunk_size {
|
||||
if let Some((l, r)) = self.read_frame_internal()? {
|
||||
self.left_buffer.extend(l);
|
||||
self.right_buffer.extend(r);
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if self.left_buffer.is_empty() {
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
let actual_size = std::cmp::min(chunk_size, self.left_buffer.len());
|
||||
let left = self.left_buffer.drain(0..actual_size).collect();
|
||||
let right = self.right_buffer.drain(0..actual_size).collect();
|
||||
|
||||
Ok(Some((left, right)))
|
||||
}
|
||||
|
||||
fn read_frame_internal(&mut self) -> Result<Option<(Vec<i16>, Vec<i16>)>> {
|
||||
loop {
|
||||
let packet = match self.format.next_packet() {
|
||||
Ok(packet) => packet,
|
||||
Err(symphonia::core::errors::Error::IoError(ref e))
|
||||
if e.kind() == std::io::ErrorKind::UnexpectedEof =>
|
||||
{
|
||||
return Ok(None);
|
||||
}
|
||||
Err(e) => return Err(e.into()),
|
||||
};
|
||||
|
||||
if packet.track_id() != self.track_id {
|
||||
continue;
|
||||
}
|
||||
|
||||
let decoded = self
|
||||
.decoder
|
||||
.decode(&packet)
|
||||
.context("Failed to decode packet")?;
|
||||
|
||||
if self.sample_buf.is_none() {
|
||||
self.sample_buf = Some(SampleBuffer::<i16>::new(
|
||||
decoded.capacity() as u64,
|
||||
*decoded.spec(),
|
||||
));
|
||||
}
|
||||
|
||||
if let Some(buf) = self.sample_buf.as_mut() {
|
||||
buf.copy_interleaved_ref(decoded);
|
||||
let samples = buf.samples();
|
||||
|
||||
let mut left = Vec::with_capacity(samples.len() / self.channels);
|
||||
let mut right = Vec::with_capacity(samples.len() / self.channels);
|
||||
|
||||
if self.channels == 2 {
|
||||
for i in (0..samples.len()).step_by(2) {
|
||||
left.push(samples[i]);
|
||||
right.push(samples[i + 1]);
|
||||
}
|
||||
} else {
|
||||
for &s in samples {
|
||||
left.push(s);
|
||||
right.push(s);
|
||||
}
|
||||
}
|
||||
return Ok(Some((left, right)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,42 +1,56 @@
|
||||
#![cfg(target_os = "linux")]
|
||||
|
||||
use crate::audio::config::AudioConfig;
|
||||
use alsa::Direction;
|
||||
use alsa::pcm::{Access, Format, HwParams, PCM};
|
||||
use anyhow::{Context, Result};
|
||||
use anyhow::{Context, Result, anyhow};
|
||||
|
||||
pub struct AudioRecorder {
|
||||
pcm: PCM,
|
||||
#[cfg(target_os = "linux")]
|
||||
pcm: alsa::pcm::PCM,
|
||||
}
|
||||
|
||||
impl AudioRecorder {
|
||||
pub fn new(config: &AudioConfig) -> Result<Self> {
|
||||
let pcm = PCM::new(&config.capture_device, Direction::Capture, false)?;
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
let hwp = HwParams::any(&pcm)?;
|
||||
hwp.set_access(Access::RWInterleaved)?;
|
||||
hwp.set_format(Format::s16())?;
|
||||
hwp.set_rate_near(config.sample_rate, alsa::ValueOr::Nearest)?;
|
||||
hwp.set_channels_near(config.channels as u32)?;
|
||||
pcm.hw_params(&hwp)?;
|
||||
use alsa::Direction;
|
||||
use alsa::pcm::{Access, Format, HwParams, PCM};
|
||||
|
||||
let pcm = PCM::new(&config.capture_device, Direction::Capture, false)?;
|
||||
{
|
||||
let hwp = HwParams::any(&pcm)?;
|
||||
hwp.set_access(Access::RWInterleaved)?;
|
||||
hwp.set_format(Format::s16())?;
|
||||
hwp.set_rate_near(config.sample_rate, alsa::ValueOr::Nearest)?;
|
||||
hwp.set_channels_near(config.channels as u32)?;
|
||||
pcm.hw_params(&hwp)?;
|
||||
}
|
||||
pcm.prepare()?;
|
||||
Ok(Self { pcm })
|
||||
}
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
{
|
||||
Err(anyhow!("Linux Only"))
|
||||
}
|
||||
pcm.prepare()?;
|
||||
Ok(Self { pcm })
|
||||
}
|
||||
|
||||
pub fn read(&self, buf: &mut [i16]) -> Result<usize> {
|
||||
match self.pcm.io_i16()?.readi(buf) {
|
||||
Ok(n) => Ok(n),
|
||||
Err(e) if e.errno() == 32 => {
|
||||
// 32 = Broken pipe (Overrun)
|
||||
println!("ALSA recording overrun, recovering...");
|
||||
self.pcm.prepare()?;
|
||||
self.pcm
|
||||
.io_i16()?
|
||||
.readi(buf)
|
||||
.context("ALSA read retry failed")
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
match self.pcm.io_i16()?.readi(buf) {
|
||||
Ok(n) => Ok(n),
|
||||
Err(e) if e.errno() == 32 => {
|
||||
// 32 = Broken pipe (Overrun)
|
||||
println!("ALSA recording overrun, recovering...");
|
||||
self.pcm.prepare()?;
|
||||
self.pcm
|
||||
.io_i16()?
|
||||
.readi(buf)
|
||||
.context("ALSA read retry failed")
|
||||
}
|
||||
Err(e) => Err(e.into()),
|
||||
}
|
||||
Err(e) => Err(e.into()),
|
||||
}
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
{
|
||||
Err(anyhow!("Linux Only"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -36,7 +36,7 @@ impl WavWriter {
|
||||
pub fn finalize(mut self) -> Result<()> {
|
||||
self.writer.flush()?;
|
||||
let mut file = self.writer.into_inner()?;
|
||||
|
||||
|
||||
file.seek(SeekFrom::Start(0))?;
|
||||
|
||||
let file_size = 36 + self.data_size;
|
||||
@@ -93,14 +93,43 @@ impl WavReader {
|
||||
})
|
||||
}
|
||||
|
||||
pub fn read_samples(&mut self, samples: &mut [i16]) -> Result<usize> {
|
||||
let mut bytes = vec![0u8; samples.len() * 2];
|
||||
let n = self.reader.read(&mut bytes)?;
|
||||
let sample_count = n / 2;
|
||||
for i in 0..sample_count {
|
||||
samples[i] = i16::from_le_bytes([bytes[i * 2], bytes[i * 2 + 1]]);
|
||||
/// 读取指定的采样数。如果是立体声，则返回左右声道独立的 Vec。
|
||||
/// chunk_size: 每个声道需要读取的采样点数。
|
||||
pub fn read_chunk(&mut self, chunk_size: usize) -> Result<Option<(Vec<i16>, Vec<i16>)>> {
|
||||
let channels = self.channels as usize;
|
||||
let total_samples_to_read = chunk_size * channels;
|
||||
let mut bytes = vec![0u8; total_samples_to_read * 2]; // 预分配完整大小的字节数组
|
||||
|
||||
let bytes_read = self.reader.read(&mut bytes)?;
|
||||
|
||||
// 如果连 1 字节都没读到，说明到文件结尾了
|
||||
if bytes_read == 0 {
|
||||
return Ok(None);
|
||||
}
|
||||
Ok(sample_count)
|
||||
|
||||
let mut left = vec![0i16; chunk_size]; // 预填 0
|
||||
let mut right = vec![0i16; chunk_size]; // 预填 0
|
||||
|
||||
// 实际读到了多少个采样点（总数）
|
||||
let total_samples_read = bytes_read / 2;
|
||||
// 计算每个声道实际读到了多少个点
|
||||
let samples_per_channel = total_samples_read / channels;
|
||||
|
||||
for i in 0..samples_per_channel {
|
||||
let base = i * channels * 2;
|
||||
|
||||
// 左声道 (或单声道)
|
||||
left[i] = i16::from_le_bytes([bytes[base], bytes[base + 1]]);
|
||||
|
||||
if channels == 2 {
|
||||
// 右声道
|
||||
right[i] = i16::from_le_bytes([bytes[base + 2], bytes[base + 3]]);
|
||||
} else {
|
||||
// 单声道填充双声道时，复制左声道
|
||||
right[i] = left[i];
|
||||
}
|
||||
}
|
||||
|
||||
Ok(Some((left, right)))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -6,21 +6,11 @@
|
||||
//! - 音频录制和播放
|
||||
//! - 事件处理
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
use std::sync::Arc;
|
||||
#[cfg(target_os = "linux")]
|
||||
use xiao::app::client::{Client, ClientConfig};
|
||||
#[cfg(target_os = "linux")]
|
||||
use xiao::net::command::Command;
|
||||
#[cfg(target_os = "linux")]
|
||||
use xiao::net::event::NotificationLevel;
|
||||
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
fn main() {
|
||||
println!("This client only works on Linux due to ALSA dependencies.");
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
println!("╔═══════════════════════════════════════════════════════╗");
|
||||
|
||||
@@ -109,7 +109,7 @@ async fn main() -> anyhow::Result<()> {
|
||||
match server.start_play(addr, "temp/test.wav").await {
|
||||
Ok(_) => {
|
||||
println!(" ▶️ Playback started...");
|
||||
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
|
||||
tokio::time::sleep(std::time::Duration::from_secs(60)).await;
|
||||
server.stop_play(addr).await?;
|
||||
println!(" ⏹️ Playback stopped");
|
||||
}
|
||||
|
||||
@@ -321,4 +321,3 @@ impl CommandResult {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -45,16 +45,10 @@ pub enum ServerEvent {
|
||||
},
|
||||
|
||||
/// 客户端加入（广播给其他客户端）
|
||||
ClientJoined {
|
||||
addr: String,
|
||||
model: String,
|
||||
},
|
||||
ClientJoined { addr: String, model: String },
|
||||
|
||||
/// 客户端离开
|
||||
ClientLeft {
|
||||
addr: String,
|
||||
model: String,
|
||||
},
|
||||
ClientLeft { addr: String, model: String },
|
||||
|
||||
/// 服务器消息/通知
|
||||
Notification {
|
||||
@@ -71,9 +65,7 @@ pub enum ServerEvent {
|
||||
},
|
||||
|
||||
/// 播放完成
|
||||
PlaybackComplete {
|
||||
filename: String,
|
||||
},
|
||||
PlaybackComplete { filename: String },
|
||||
|
||||
/// 服务器状态更新
|
||||
ServerStatus {
|
||||
@@ -82,10 +74,7 @@ pub enum ServerEvent {
|
||||
},
|
||||
|
||||
/// 自定义事件
|
||||
Custom {
|
||||
name: String,
|
||||
payload: Vec<u8>,
|
||||
},
|
||||
Custom { name: String, payload: Vec<u8> },
|
||||
}
|
||||
|
||||
/// 客户端事件（Client → Server）
|
||||
@@ -99,16 +88,10 @@ pub enum ClientEvent {
|
||||
},
|
||||
|
||||
/// 音频电平
|
||||
AudioLevel {
|
||||
level_db: f32,
|
||||
is_silent: bool,
|
||||
},
|
||||
AudioLevel { level_db: f32, is_silent: bool },
|
||||
|
||||
/// 按键事件
|
||||
KeyPress {
|
||||
key: String,
|
||||
action: KeyAction,
|
||||
},
|
||||
KeyPress { key: String, action: KeyAction },
|
||||
|
||||
/// 警告/错误
|
||||
Alert {
|
||||
@@ -117,10 +100,7 @@ pub enum ClientEvent {
|
||||
},
|
||||
|
||||
/// 自定义事件
|
||||
Custom {
|
||||
name: String,
|
||||
payload: Vec<u8>,
|
||||
},
|
||||
Custom { name: String, payload: Vec<u8> },
|
||||
}
|
||||
|
||||
/// 通知级别
|
||||
@@ -283,4 +263,3 @@ where
|
||||
handler.handle(event).await;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -14,3 +14,4 @@ pub mod event;
|
||||
pub mod network;
|
||||
pub mod protocol;
|
||||
pub mod rpc;
|
||||
pub mod sync;
|
||||
|
||||
@@ -19,8 +19,9 @@ pub struct ClientInfo {
|
||||
/// 音频数据包 - UDP 通道传输
|
||||
#[derive(Serialize, Deserialize, Debug, Clone)]
|
||||
pub struct AudioPacket {
|
||||
/// Opus 编码的音频数据
|
||||
pub data: Vec<u8>,
|
||||
pub seq: u32, // 序列号，用于丢包检测
|
||||
pub timestamp: u128, // 目标播放时间 (主节点时间)
|
||||
pub data: Vec<u8>, // Opus 编码数据
|
||||
}
|
||||
|
||||
// ==================== 控制包 ====================
|
||||
@@ -51,8 +52,15 @@ pub enum ControlPacket {
|
||||
},
|
||||
|
||||
// ========== 心跳 ==========
|
||||
Ping,
|
||||
Pong,
|
||||
Ping {
|
||||
client_ts: u128,
|
||||
seq: u32,
|
||||
},
|
||||
Pong {
|
||||
client_ts: u128,
|
||||
server_ts: u128,
|
||||
seq: u32,
|
||||
},
|
||||
|
||||
// ========== RPC ==========
|
||||
/// RPC 请求（新版，使用 Command 类型）
|
||||
|
||||
@@ -0,0 +1,224 @@
|
||||
use std::collections::VecDeque;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
/// 获取当前微秒级时间戳
|
||||
pub fn now_us() -> u128 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.expect("时间倒流")
|
||||
.as_micros()
|
||||
}
|
||||
|
||||
/// 时钟同步管理器，用于计算主从节点间的时钟偏移
|
||||
/// 采用改进的 NTP 算法 + Kalman 滤波思想
|
||||
pub struct ClockSync {
|
||||
/// 偏移量样本窗口
|
||||
offsets: VecDeque<OffsetSample>,
|
||||
/// 当前估计的时钟偏移 (server_time - client_time)
|
||||
pub current_offset: i128,
|
||||
/// RTT 样本窗口
|
||||
rtts: VecDeque<i128>,
|
||||
/// 当前估计的最小 RTT
|
||||
min_rtt: i128,
|
||||
/// 窗口大小
|
||||
window_size: usize,
|
||||
/// 时钟漂移率 (ppm: parts per million)
|
||||
/// 正值表示从节点时钟比主节点快
|
||||
drift_rate: f64,
|
||||
/// 上次更新时间
|
||||
last_update_time: u128,
|
||||
/// 漂移率估计窗口
|
||||
drift_samples: VecDeque<DriftSample>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
struct OffsetSample {
|
||||
offset: i128,
|
||||
rtt: i128,
|
||||
timestamp: u128,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
struct DriftSample {
|
||||
offset: i128,
|
||||
timestamp: u128,
|
||||
}
|
||||
|
||||
impl ClockSync {
|
||||
pub fn new(window_size: usize) -> Self {
|
||||
Self {
|
||||
offsets: VecDeque::with_capacity(window_size),
|
||||
current_offset: 0,
|
||||
rtts: VecDeque::with_capacity(window_size),
|
||||
min_rtt: i128::MAX,
|
||||
window_size,
|
||||
drift_rate: 0.0,
|
||||
last_update_time: now_us(),
|
||||
drift_samples: VecDeque::with_capacity(60), // 保留 60 秒的样本
|
||||
}
|
||||
}
|
||||
|
||||
/// 更新时钟偏移估计 (NTP 算法)
|
||||
///
|
||||
/// NTP 时间戳标记:
|
||||
/// t1 = client_send_ts : 客户端发送 Ping 的时间
|
||||
/// t2 = server_ts : 服务器接收 Ping 的时间
|
||||
/// t3 = server_ts : 服务器发送 Pong 的时间 (假设处理时间忽略不计)
|
||||
/// t4 = client_recv_ts : 客户端接收 Pong 的时间
|
||||
///
|
||||
/// RTT = (t4 - t1) - (t3 - t2) = (t4 - t1) (因为 t3 = t2)
|
||||
/// Offset = ((t2 - t1) + (t3 - t4)) / 2 = ((t2 - t1) + (t2 - t4)) / 2
|
||||
/// = t2 - (t1 + t4) / 2
|
||||
pub fn update(&mut self, client_send_ts: u128, server_ts: u128, client_recv_ts: u128) {
|
||||
let t1 = client_send_ts as i128;
|
||||
let t2 = server_ts as i128;
|
||||
let t4 = client_recv_ts as i128;
|
||||
|
||||
let rtt = t4 - t1;
|
||||
|
||||
// 过滤异常 RTT (局域网内 > 100ms 视为异常)
|
||||
if rtt < 0 || rtt > 100_000 {
|
||||
return;
|
||||
}
|
||||
|
||||
// 计算时钟偏移: offset = server_time - client_time
|
||||
// offset = t2 - (t1 + t4) / 2
|
||||
let offset = t2 - (t1 + t4) / 2;
|
||||
|
||||
// 更新 RTT 窗口
|
||||
self.rtts.push_back(rtt);
|
||||
if self.rtts.len() > self.window_size {
|
||||
self.rtts.pop_front();
|
||||
}
|
||||
self.min_rtt = *self.rtts.iter().min().unwrap_or(&rtt);
|
||||
|
||||
// 更新偏移量窗口
|
||||
let sample = OffsetSample {
|
||||
offset,
|
||||
rtt,
|
||||
timestamp: client_recv_ts,
|
||||
};
|
||||
self.offsets.push_back(sample);
|
||||
if self.offsets.len() > self.window_size {
|
||||
self.offsets.pop_front();
|
||||
}
|
||||
|
||||
// 偏移量估计: 使用低 RTT 样本的中位数
|
||||
// 原理: RTT 较小的样本受网络抖动影响小，时间测量更准确
|
||||
let mut low_rtt_offsets: Vec<i128> = self
|
||||
.offsets
|
||||
.iter()
|
||||
.filter(|s| s.rtt <= self.min_rtt + 5000) // 5ms 容差
|
||||
.map(|s| s.offset)
|
||||
.collect();
|
||||
|
||||
if !low_rtt_offsets.is_empty() {
|
||||
low_rtt_offsets.sort_unstable();
|
||||
let new_offset = low_rtt_offsets[low_rtt_offsets.len() / 2];
|
||||
|
||||
// 漂移率估计
|
||||
self.estimate_drift(new_offset, client_recv_ts);
|
||||
|
||||
// 平滑更新偏移量 (避免突变)
|
||||
let alpha = 0.3; // 低通滤波系数
|
||||
self.current_offset =
|
||||
(alpha * new_offset as f64 + (1.0 - alpha) * self.current_offset as f64) as i128;
|
||||
}
|
||||
|
||||
self.last_update_time = client_recv_ts;
|
||||
}
|
||||
|
||||
/// 估计时钟漂移率
|
||||
/// 时钟漂移率 = d(offset) / dt
|
||||
fn estimate_drift(&mut self, offset: i128, timestamp: u128) {
|
||||
self.drift_samples
|
||||
.push_back(DriftSample { offset, timestamp });
|
||||
if self.drift_samples.len() > 60 {
|
||||
self.drift_samples.pop_front();
|
||||
}
|
||||
|
||||
// 至少需要 10 秒的数据才能估计漂移
|
||||
if self.drift_samples.len() < 10 {
|
||||
return;
|
||||
}
|
||||
|
||||
// 使用线性回归估计漂移率
|
||||
let first = self.drift_samples.front().unwrap();
|
||||
let last = self.drift_samples.back().unwrap();
|
||||
|
||||
let dt = (last.timestamp - first.timestamp) as f64;
|
||||
let d_offset = (last.offset - first.offset) as f64;
|
||||
|
||||
if dt > 10_000_000.0 {
|
||||
// 超过 10 秒
|
||||
// drift_rate 单位: 微秒/秒 = ppm
|
||||
let new_drift = d_offset / (dt / 1_000_000.0);
|
||||
|
||||
// 平滑更新漂移率
|
||||
let beta = 0.1;
|
||||
self.drift_rate = beta * new_drift + (1.0 - beta) * self.drift_rate;
|
||||
}
|
||||
}
|
||||
|
||||
/// 将本地时间转换为服务器(主节点)时间
|
||||
/// 考虑时钟漂移补偿
|
||||
pub fn to_server_time(&self, client_time: u128) -> u128 {
|
||||
let base_server_time = (client_time as i128 + self.current_offset) as u128;
|
||||
|
||||
// 漂移补偿: 根据距离上次同步的时间，补偿时钟漂移
|
||||
let elapsed_since_update = client_time.saturating_sub(self.last_update_time) as f64;
|
||||
let drift_correction = (self.drift_rate * elapsed_since_update / 1_000_000.0) as i128;
|
||||
|
||||
(base_server_time as i128 + drift_correction) as u128
|
||||
}
|
||||
|
||||
/// 将服务器(主节点)时间转换为本地时间
|
||||
pub fn to_client_time(&self, server_time: u128) -> u128 {
|
||||
// 简化版本，不考虑漂移补偿 (播放时主要用 to_server_time)
|
||||
(server_time as i128 - self.current_offset) as u128
|
||||
}
|
||||
|
||||
/// 获取当前估计的 RTT (微秒)
|
||||
pub fn get_rtt(&self) -> i128 {
|
||||
self.min_rtt
|
||||
}
|
||||
|
||||
/// 获取当前时钟漂移率 (ppm)
|
||||
pub fn get_drift_rate(&self) -> f64 {
|
||||
self.drift_rate
|
||||
}
|
||||
|
||||
/// 获取同步质量评估 (0-100, 越高越好)
|
||||
pub fn get_sync_quality(&self) -> u8 {
|
||||
if self.offsets.is_empty() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
// 基于 RTT 稳定性和偏移量方差评估
|
||||
let rtt_variance = self.calculate_variance(&self.rtts.iter().copied().collect::<Vec<_>>());
|
||||
let offset_variance =
|
||||
self.calculate_variance(&self.offsets.iter().map(|s| s.offset).collect::<Vec<_>>());
|
||||
|
||||
// RTT 越稳定，方差越小，质量越高
|
||||
let rtt_score = ((100_000.0 - rtt_variance.min(100_000.0)) / 100_000.0 * 50.0) as u8;
|
||||
let offset_score = ((50_000.0 - offset_variance.min(50_000.0)) / 50_000.0 * 50.0) as u8;
|
||||
|
||||
rtt_score + offset_score
|
||||
}
|
||||
|
||||
fn calculate_variance(&self, samples: &[i128]) -> f64 {
|
||||
if samples.is_empty() {
|
||||
return 0.0;
|
||||
}
|
||||
let mean = samples.iter().sum::<i128>() as f64 / samples.len() as f64;
|
||||
let variance = samples
|
||||
.iter()
|
||||
.map(|&x| {
|
||||
let diff = x as f64 - mean;
|
||||
diff * diff
|
||||
})
|
||||
.sum::<f64>()
|
||||
/ samples.len() as f64;
|
||||
variance.sqrt()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user