From 4ed8d3dca9b66e2036eece84cf019bfa4f08eae4 Mon Sep 17 00:00:00 2001 From: Del Wang Date: Fri, 2 Jan 2026 13:04:19 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=88=9D=E5=A7=8B=E5=8C=96=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E5=9C=BA=E6=99=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/client-v2/Cargo.lock | 1 + packages/client-v2/Cargo.toml | 1 + packages/client-v2/Makefile | 8 +- packages/client-v2/src/app/client/core.rs | 224 ++++++++++++ packages/client-v2/src/app/client/entry.rs | 10 +- packages/client-v2/src/app/client/mod.rs | 1 + packages/client-v2/src/app/mod.rs | 1 - packages/client-v2/src/app/server/core.rs | 305 ++++++++++++++++ packages/client-v2/src/app/server/entry.rs | 55 ++- packages/client-v2/src/app/server/mod.rs | 1 + packages/client-v2/src/app/stereo/entry.rs | 35 -- packages/client-v2/src/app/stereo/master.rs | 365 -------------------- packages/client-v2/src/app/stereo/mod.rs | 3 - packages/client-v2/src/app/stereo/slave.rs | 199 ----------- packages/client-v2/src/audio/config.rs | 6 +- packages/client-v2/src/audio/mod.rs | 1 + packages/client-v2/src/audio/wav.rs | 106 ++++++ packages/client-v2/src/bin/client.rs | 4 - packages/client-v2/src/bin/stereo.rs | 14 - packages/client-v2/src/net/discovery.rs | 12 +- packages/client-v2/src/net/network.rs | 58 ++-- packages/client-v2/src/net/protocol.rs | 70 ++-- 22 files changed, 785 insertions(+), 695 deletions(-) create mode 100644 packages/client-v2/src/app/client/core.rs create mode 100644 packages/client-v2/src/app/server/core.rs delete mode 100644 packages/client-v2/src/app/stereo/entry.rs delete mode 100644 packages/client-v2/src/app/stereo/master.rs delete mode 100644 packages/client-v2/src/app/stereo/mod.rs delete mode 100644 packages/client-v2/src/app/stereo/slave.rs create mode 100644 packages/client-v2/src/audio/wav.rs delete mode 100644 packages/client-v2/src/bin/stereo.rs diff --git a/packages/client-v2/Cargo.lock b/packages/client-v2/Cargo.lock index 57a53c0..3f88ac8 100644 --- a/packages/client-v2/Cargo.lock +++ b/packages/client-v2/Cargo.lock @@ -541,6 +541,7 @@ dependencies = [ "alsa", "anyhow", "opus", + "parking_lot", "postcard", "serde", "tokio", diff --git a/packages/client-v2/Cargo.toml b/packages/client-v2/Cargo.toml index 79a2541..ab96699 100644 --- a/packages/client-v2/Cargo.toml +++ b/packages/client-v2/Cargo.toml @@ -21,6 +21,7 @@ anyhow = "1.0" tokio = { version = "1.48", features = ["full"] } serde = { version = "1.0", features = ["derive"] } postcard = { version = "1.0", features = ["alloc", "use-std"] } +parking_lot = "0.12.5" [target.'cfg(target_os = "linux")'.dependencies] alsa = "0.11" diff --git a/packages/client-v2/Makefile b/packages/client-v2/Makefile index ba65170..9c898b8 100644 --- a/packages/client-v2/Makefile +++ b/packages/client-v2/Makefile @@ -1,16 +1,14 @@ build-server: - cargo build --release --bin server --features app + clear && cargo build --release --bin server --features app build-client: - docker run --rm -v $(shell pwd):/app idootop/open-xiaoai-runtime:oh2p \ + clear && docker run --rm -v $(shell pwd):/app idootop/open-xiaoai-runtime:oh2p \ cargo build --target armv7-unknown-linux-gnueabihf --release --bin client --features app run-server: - cargo run --bin server --features app + target/release/server # 部署到小爱音箱（调试自用） deploy: - dd if=target/armv7-unknown-linux-gnueabihf/release/client \ - | sshpass -p open-xiaoai ssh -o HostKeyAlgorithms=+ssh-rsa root@192.168.31.153 "dd of=/data/client" dd if=target/armv7-unknown-linux-gnueabihf/release/client \ | sshpass -p open-xiaoai ssh -o HostKeyAlgorithms=+ssh-rsa root@192.168.31.235 "dd of=/data/client" \ No newline at end of file diff --git a/packages/client-v2/src/app/client/core.rs b/packages/client-v2/src/app/client/core.rs new file mode 100644 index 0000000..2c8ef5d --- /dev/null +++ b/packages/client-v2/src/app/client/core.rs @@ -0,0 +1,224 @@ +#![cfg(target_os = "linux")] + +use crate::audio::codec::OpusCodec; +use crate::audio::config::AudioConfig; +use crate::audio::player::AudioPlayer; +use crate::audio::recorder::AudioRecorder; +use crate::net::discovery::Discovery; +use crate::net::network::{AudioSocket, ClientNetwork, ControlConnection}; +use crate::net::protocol::{AudioPacket, ControlPacket, DeviceInfo, RpcResult}; +use anyhow::{Context, Result}; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::sync::broadcast; + +pub struct Client { + info: DeviceInfo, +} + +impl Client { + pub fn new(model: &str, mac: &str, version: u32) -> Self { + Self { + info: DeviceInfo { + model: model.to_string(), + mac: mac.to_string(), + version, + }, + } + } + + pub async fn run(self) -> Result<()> { + println!("正在寻找服务端..."); + let (server_ip, server_port) = Discovery::discover_server().await?; + let server_addr = SocketAddr::new(server_ip, server_port); + println!("发现服务端: {}", server_addr); + + let network = ClientNetwork::connect(server_addr).await?; + let mut control = network.into_control(); + + // 握手 + control + .send_packet(&ControlPacket::ClientIdentify { + info: self.info.clone(), + }) + .await?; + + match control.recv_packet().await? { + ControlPacket::IdentifyOk => println!("认证成功"), + p => return Err(anyhow::anyhow!("认证失败: {:?}", p)), + } + + let (stop_tx, _) = broadcast::channel::<()>(1); + let audio_socket = Arc::new(AudioSocket::bind().await?); + + loop { + tokio::select! { + packet = control.recv_packet() => { + let packet = packet?; + match packet { + ControlPacket::StartRecording { config } => { + println!("开始录音: {:?}", config); + let socket = audio_socket.clone(); + let mut stop_rx = stop_tx.subscribe(); + let server_addr = control.peer_addr()?; + tokio::spawn(async move { + if let Err(e) = handle_recording(config, socket, server_addr, stop_rx).await { + eprintln!("录音出错: {}", e); + } + }); + } + ControlPacket::StopRecording => { + println!("停止录音"); + let _ = stop_tx.send(()); + } + ControlPacket::StartPlayback { config } => { + println!("开始播放: {:?}", config); + let socket = audio_socket.clone(); + let mut stop_rx = stop_tx.subscribe(); + tokio::spawn(async move { + if let Err(e) = handle_playback(config, socket, stop_rx).await { + eprintln!("播放出错: {}", e); + } + }); + } + ControlPacket::StopPlayback => { + println!("停止播放"); + let _ = stop_tx.send(()); + } + ControlPacket::RpcRequest { id, method, args } => { + println!("收到 RPC 请求: {} {:?}", method, args); + let result = handle_rpc(&method, args).await; + control.send_packet(&ControlPacket::RpcResponse { id, result }).await?; + } + ControlPacket::Ping => { + control.send_packet(&ControlPacket::Pong).await?; + } + _ => {} + } + } + } + } + } +} + +async fn handle_recording( + config: AudioConfig, + socket: Arc, + server_addr: SocketAddr, + mut stop_rx: broadcast::Receiver<()>, +) -> Result<()> { + #[cfg(target_os = "linux")] + { + let recorder = AudioRecorder::new(&config)?; + let mut codec = OpusCodec::new(&config)?; + let mut pcm_buf = vec![0i16; config.frame_size]; + let mut opus_buf = vec![0u8; 4096]; + + loop { + if stop_rx.try_recv().is_ok() { + break; + } + let n = recorder.read(&mut pcm_buf)?; + if n > 0 { + let opus_len = codec.encode(&pcm_buf[..n], &mut opus_buf)?; + let packet = AudioPacket { + data: opus_buf[..opus_len].to_vec(), + }; + socket.send_packet(&packet, server_addr).await?; + } + } + } + #[cfg(not(target_os = "linux"))] + { + println!("当前系统不支持 ALSA 录音，模拟发送音频数据..."); + loop { + if stop_rx.try_recv().is_ok() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + let packet = AudioPacket { + data: vec![0u8; 10], + }; + socket.send_packet(&packet, server_addr).await?; + } + } + Ok(()) +} + +async fn handle_playback( + config: AudioConfig, + socket: Arc, + mut stop_rx: broadcast::Receiver<()>, +) -> Result<()> { + #[cfg(target_os = "linux")] + { + let player = AudioPlayer::new(&config)?; + let mut codec = OpusCodec::new(&config)?; + let mut pcm_buf = vec![0i16; config.frame_size]; + let mut udp_buf = vec![0u8; 4096]; + + loop { + if stop_rx.try_recv().is_ok() { + break; + } + // 这里简单处理，UDP 接收可能阻塞。实际建议加超时或 select + let (packet, _) = socket.recv_packet(&mut udp_buf).await?; + let pcm_len = codec.decode(&packet.data, &mut pcm_buf)?; + player.write(&pcm_buf[..pcm_len])?; + } + } + #[cfg(not(target_os = "linux"))] + { + println!("当前系统不支持 ALSA 播放，模拟接收音频数据..."); + let mut udp_buf = vec![0u8; 4096]; + loop { + tokio::select! { + _ = stop_rx.recv() => break, + res = socket.recv_packet(&mut udp_buf) => { + let _ = res?; + } + } + } + } + Ok(()) +} + +async fn handle_rpc(method: &str, args: Vec) -> RpcResult { + if method == "shell" && !args.is_empty() { + let cmd = &args[0]; + println!("执行 Shell: {}", cmd); + + // 模拟执行 + #[cfg(target_os = "linux")] + { + use std::process::Command; + let output = Command::new("sh").arg("-c").arg(cmd).output(); + match output { + Ok(out) => RpcResult { + stdout: String::from_utf8_lossy(&out.stdout).to_string(), + stderr: String::from_utf8_lossy(&out.stderr).to_string(), + code: out.status.code().unwrap_or(0), + }, + Err(e) => RpcResult { + stdout: "".to_string(), + stderr: e.to_string(), + code: -1, + }, + } + } + #[cfg(not(target_os = "linux"))] + { + RpcResult { + stdout: format!("模拟执行 {} 成功", cmd), + stderr: "".to_string(), + code: 0, + } + } + } else { + RpcResult { + stdout: "".to_string(), + stderr: "未知方法或参数错误".to_string(), + code: -1, + } + } +} diff --git a/packages/client-v2/src/app/client/entry.rs b/packages/client-v2/src/app/client/entry.rs index 639ae06..79ee323 100644 --- a/packages/client-v2/src/app/client/entry.rs +++ b/packages/client-v2/src/app/client/entry.rs @@ -1,8 +1,14 @@ #![cfg(target_os = "linux")] +use crate::app::client::core::Client; use anyhow::Result; pub async fn run_client() -> Result<()> { - println!("Hello, Client!"); - Ok(()) + // 模拟从系统获取信息 + let model = "XiaoAi-V2-Simulated"; + let mac = "00:11:22:33:44:55"; + let version = 1; + + let client = Client::new(model, mac, version); + client.run().await } diff --git a/packages/client-v2/src/app/client/mod.rs b/packages/client-v2/src/app/client/mod.rs index eb5becf..1ca5324 100644 --- a/packages/client-v2/src/app/client/mod.rs +++ b/packages/client-v2/src/app/client/mod.rs @@ -1 +1,2 @@ +pub mod core; pub mod entry; diff --git a/packages/client-v2/src/app/mod.rs b/packages/client-v2/src/app/mod.rs index ae350e2..c07f47e 100644 --- a/packages/client-v2/src/app/mod.rs +++ b/packages/client-v2/src/app/mod.rs @@ -1,3 +1,2 @@ pub mod client; pub mod server; -pub mod stereo; diff --git a/packages/client-v2/src/app/server/core.rs b/packages/client-v2/src/app/server/core.rs new file mode 100644 index 0000000..04674b3 --- /dev/null +++ b/packages/client-v2/src/app/server/core.rs @@ -0,0 +1,305 @@ +use crate::audio::codec::OpusCodec; +use crate::audio::config::AudioConfig; +use crate::audio::wav::{WavReader, WavWriter}; +use crate::net::discovery::Discovery; +use crate::net::network::{AudioSocket, ControlConnection, ServerNetwork}; +use crate::net::protocol::{AudioPacket, ControlPacket, DeviceInfo, RpcResult}; +use anyhow::{Context, Result}; +use parking_lot::Mutex; +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::atomic::{AtomicU32, Ordering}; +use tokio::sync::{broadcast, oneshot}; + +pub struct RpcManager { + next_id: AtomicU32, + pending: Mutex>>, +} + +impl RpcManager { + pub fn new() -> Self { + Self { + next_id: AtomicU32::new(1), + pending: Mutex::new(HashMap::new()), + } + } + + pub fn alloc_id(&self) -> (u32, oneshot::Receiver) { + let id = self.next_id.fetch_add(1, Ordering::SeqCst); + let (tx, rx) = oneshot::channel(); + self.pending.lock().insert(id, tx); + (id, rx) + } + + pub fn fulfill(&self, id: u32, result: RpcResult) { + if let Some(tx) = self.pending.lock().remove(&id) { + let _ = tx.send(result); + } + } +} + +pub struct Session { + pub info: DeviceInfo, + pub control: Arc>, + pub addr: SocketAddr, + pub rpc: Arc, +} + +pub struct Server { + sessions: Arc>>>, + audio_socket: Arc, +} + +impl Server { + pub async fn new() -> Result { + Ok(Self { + sessions: Arc::new(Mutex::new(HashMap::new())), + audio_socket: Arc::new(AudioSocket::bind().await?), + }) + } + + pub async fn run(self: Arc, port: u16) -> Result<()> { + let network = ServerNetwork::setup(port).await?; + println!("服务端启动在: {}", network.local_addr()?); + + // 启动广播 + Discovery::start_broadcast(port).await?; + + loop { + let (control, addr) = network.accept().await?; + let server = self.clone(); + tokio::spawn(async move { + if let Err(e) = server.handle_connection(control, addr).await { + eprintln!("连接 {} 出错: {}", addr, e); + } + server.sessions.lock().remove(&addr); + println!("客户端 {} 断开连接", addr); + }); + } + } + + async fn handle_connection( + &self, + mut control: ControlConnection, + addr: SocketAddr, + ) -> Result<()> { + println!("新客户端连接: {}", addr); + + // 握手认证 + let info = match control.recv_packet().await? { + ControlPacket::ClientIdentify { info } => info, + p => return Err(anyhow::anyhow!("预期的握手包，收到: {:?}", p)), + }; + + println!( + "客户端识别: {} ({}) v{}", + info.model, info.mac, info.version + ); + control.send_packet(&ControlPacket::IdentifyOk).await?; + + let rpc = Arc::new(RpcManager::new()); + let control = Arc::new(tokio::sync::Mutex::new(control)); + let session = Arc::new(Session { + info, + control: control.clone(), + addr, + rpc: rpc.clone(), + }); + + self.sessions.lock().insert(addr, session.clone()); + + // 处理控制消息循环 + loop { + let mut ctrl = control.lock().await; + let packet = ctrl.recv_packet().await?; + drop(ctrl); // 释放锁以允许发送 + + match packet { + ControlPacket::RpcResponse { id, result } => { + rpc.fulfill(id, result); + } + ControlPacket::Ping => { + control + .lock() + .await + .send_packet(&ControlPacket::Pong) + .await?; + } + _ => {} + } + } + } + + // 暴露给外部调用的方法 + pub async fn call_shell(&self, addr: SocketAddr, cmd: &str) -> Result { + let session = self + .sessions + .lock() + .get(&addr) + .cloned() + .context("未找到 Session")?; + let (id, rx) = session.rpc.alloc_id(); + session + .control + .lock() + .await + .send_packet(&ControlPacket::RpcRequest { + id, + method: "shell".to_string(), + args: vec![cmd.to_string()], + }) + .await?; + Ok(rx.await?) + } + + pub async fn start_recording(&self, addr: SocketAddr, config: AudioConfig) -> Result<()> { + let session = self + .sessions + .lock() + .get(&addr) + .cloned() + .context("未找到 Session")?; + + // 发送开始录音指令 + session + .control + .lock() + .await + .send_packet(&ControlPacket::StartRecording { + config: config.clone(), + }) + .await?; + + let audio_socket = self.audio_socket.clone(); + tokio::spawn(async move { + if let Err(e) = save_audio_to_wav(audio_socket, config, "temp/recorded.wav").await { + eprintln!("保存录音失败: {}", e); + } + }); + + Ok(()) + } + + pub async fn stop_recording(&self, addr: SocketAddr) -> Result<()> { + let session = self + .sessions + .lock() + .get(&addr) + .cloned() + .context("未找到 Session")?; + session + .control + .lock() + .await + .send_packet(&ControlPacket::StopRecording) + .await?; + Ok(()) + } + + pub async fn start_playback(&self, addr: SocketAddr, config: AudioConfig) -> Result<()> { + let session = self + .sessions + .lock() + .get(&addr) + .cloned() + .context("未找到 Session")?; + + session + .control + .lock() + .await + .send_packet(&ControlPacket::StartPlayback { + config: config.clone(), + }) + .await?; + + let audio_socket = self.audio_socket.clone(); + tokio::spawn(async move { + if let Err(e) = stream_wav_to_client(audio_socket, addr, config, "temp/test.wav").await + { + eprintln!("推流失败: {}", e); + } + }); + + Ok(()) + } + + pub async fn stop_playback(&self, addr: SocketAddr) -> Result<()> { + let session = self + .sessions + .lock() + .get(&addr) + .cloned() + .context("未找到 Session")?; + session + .control + .lock() + .await + .send_packet(&ControlPacket::StopPlayback) + .await?; + Ok(()) + } + + pub fn get_sessions(&self) -> Vec { + self.sessions.lock().keys().cloned().collect() + } +} + +async fn save_audio_to_wav( + socket: Arc, + config: AudioConfig, + path: &str, +) -> Result<()> { + std::fs::create_dir_all("temp")?; + let mut writer = WavWriter::create(path, config.sample_rate, config.channels)?; + let mut codec = OpusCodec::new(&config)?; + let mut pcm_buf = vec![0i16; config.frame_size]; + let mut udp_buf = vec![0u8; 4096]; + + println!("正在录制到 {}...", path); + + // 这里需要一个停止机制，目前简单起见，如果一段时间没收到包就停止，或者通过全局状态 + // 为简单演示，我们录制 100 个包 + for _ in 0..100 { + let (packet, _) = socket.recv_packet(&mut udp_buf).await?; + let pcm_len = codec.decode(&packet.data, &mut pcm_buf)?; + writer.write_samples(&pcm_buf[..pcm_len])?; + } + + writer.finalize()?; + println!("录制完成: {}", path); + Ok(()) +} + +async fn stream_wav_to_client( + socket: Arc, + target: SocketAddr, + config: AudioConfig, + path: &str, +) -> Result<()> { + let mut reader = WavReader::open(path)?; + let mut codec = OpusCodec::new(&config)?; + let mut pcm_buf = vec![0i16; config.frame_size]; + let mut opus_buf = vec![0u8; 4096]; + + println!("正在从 {} 推流...", path); + + loop { + let n = reader.read_samples(&mut pcm_buf)?; + if n == 0 { + break; + } + let opus_len = codec.encode(&pcm_buf[..n], &mut opus_buf)?; + let packet = AudioPacket { + data: opus_buf[..opus_len].to_vec(), + }; + socket.send_packet(&packet, target).await?; + + // 控制发送频率，约 20ms 一帧 + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + + println!("推流结束"); + Ok(()) +} diff --git a/packages/client-v2/src/app/server/entry.rs b/packages/client-v2/src/app/server/entry.rs index df15716..3378f5a 100644 --- a/packages/client-v2/src/app/server/entry.rs +++ b/packages/client-v2/src/app/server/entry.rs @@ -1,6 +1,57 @@ +use crate::app::server::core::Server; +use crate::audio::config::AudioConfig; use anyhow::Result; +use std::sync::Arc; pub async fn run_server() -> Result<()> { - println!("Hello, Server!"); - Ok(()) + let server = Arc::new(Server::new().await?); + let s = server.clone(); + + // 运行服务器 + tokio::spawn(async move { + if let Err(e) = s.run(53531).await { + eprintln!("Server error: {}", e); + } + }); + + // 等待一个客户端连接并进行演示 + println!("等待客户端连接以进行功能演示..."); + loop { + let sessions = server.get_sessions(); + if !sessions.is_empty() { + let addr = sessions[0]; + println!("开始对 {} 进行功能测试...", addr); + + // 1. 测试 RPC + println!("测试 RPC: echo hello"); + let res = server.call_shell(addr, "echo hello").await?; + println!("RPC 结果: {:?}", res); + + // 2. 测试录音 + println!("测试录制 2 秒音频..."); + server.start_recording(addr, AudioConfig::voice()).await?; + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + server.stop_recording(addr).await?; + println!("录制结束，请检查 temp/recorded.wav"); + + // 3. 测试播放 (如果 temp/test.wav 存在) + if std::path::Path::new("temp/test.wav").exists() { + println!("测试播放 temp/test.wav..."); + server.start_playback(addr, AudioConfig::voice()).await?; + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + server.stop_playback(addr).await?; + println!("播放结束"); + } else { + println!("跳过播放测试 (temp/test.wav 不存在)"); + } + + break; + } + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + + // 保持运行 + loop { + tokio::time::sleep(std::time::Duration::from_secs(3600)).await; + } } diff --git a/packages/client-v2/src/app/server/mod.rs b/packages/client-v2/src/app/server/mod.rs index eb5becf..1ca5324 100644 --- a/packages/client-v2/src/app/server/mod.rs +++ b/packages/client-v2/src/app/server/mod.rs @@ -1 +1,2 @@ +pub mod core; pub mod entry; diff --git a/packages/client-v2/src/app/stereo/entry.rs b/packages/client-v2/src/app/stereo/entry.rs deleted file mode 100644 index 5aee647..0000000 --- a/packages/client-v2/src/app/stereo/entry.rs +++ /dev/null @@ -1,35 +0,0 @@ -#![cfg(target_os = "linux")] - -use crate::app::master::run_master; -use crate::app::slave::run_slave; -use crate::net::protocol::ChannelRole; -use anyhow::Result; -use std::env; - -pub async fn run_stereo() -> Result<()> { - let args: Vec = env::args().collect(); - if args.len() < 3 { - eprintln!("用法: {} [master|slave] [left|right]", args[0]); - return Ok(()); - } - - let mode = if args[1].to_lowercase() == "master" { - "主节点" - } else { - "从节点" - }; - - let role = if args[2].to_lowercase() == "left" { - ChannelRole::Left - } else { - ChannelRole::Right - }; - - println!("🚗 当前为: {} {}", mode, role.to_string()); - - if mode == "主节点" { - run_master(role).await - } else { - run_slave(role).await - } -} diff --git a/packages/client-v2/src/app/stereo/master.rs b/packages/client-v2/src/app/stereo/master.rs deleted file mode 100644 index 4d29be4..0000000 --- a/packages/client-v2/src/app/stereo/master.rs +++ /dev/null @@ -1,365 +0,0 @@ -#![cfg(target_os = "linux")] - -use crate::audio::codec::OpusCodec; -use crate::audio::config::AudioConfig; -use crate::audio::player::AudioPlayer; -use crate::net::discovery::Discovery; -use crate::net::network::{ControlConnection, MasterNetwork}; -use crate::net::protocol::{AudioPacket, ChannelRole, ControlPacket}; -use crate::utils::alsa::AlsaRedirector; -use crate::utils::sync::now_us; -use anyhow::{Result, anyhow}; -use std::net::SocketAddr; -use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::time::Duration; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::signal::unix::{SignalKind, signal}; -use tokio::sync::Mutex; - -pub const SERVER_TCP_PORT: u16 = 53531; - -#[derive(Clone)] -struct SlaveSession { - udp_addr: SocketAddr, - role: ChannelRole, -} - -pub async fn run_master(master_role: ChannelRole) -> Result<()> { - // 0. 设置 ALSA 重定向 - println!("🔥 启动中，请稍等..."); - let _alsa_guard = AlsaRedirector::new()?; - - // 1. 设置网络 (UDP + TCP) - let network = MasterNetwork::setup(SERVER_TCP_PORT).await?; - let audio_socket = network.audio_socket().clone_inner(); - - // 2. 启动服务发现广播 - Discovery::start_broadcast(SERVER_TCP_PORT).await?; - - println!("✅ 服务已启动，等待连接..."); - - let shutdown_flag = Arc::new(AtomicBool::new(false)); - let slaves = Arc::new(Mutex::new(Vec::::new())); - - // 3. 启动连接监听任务 - let slaves_clone = slaves.clone(); - let audio_socket_clone = audio_socket.clone(); - tokio::spawn(async move { - loop { - match network.accept().await { - Ok((control_conn, client_addr)) => { - let slaves_for_session = slaves_clone.clone(); - let audio_socket_for_session = audio_socket_clone.clone(); - tokio::spawn(async move { - if let Err(e) = handle_master_session( - control_conn, - audio_socket_for_session, - slaves_for_session, - client_addr.to_string(), - ) - .await - { - eprintln!("❌ 会话错误: {:?}", e); - } - }); - } - Err(e) => { - eprintln!("❌ Accept 错误: {:?}", e); - } - } - } - }); - - // 4. 音频处理主循环 - let config = AudioConfig::music(); - let encode_config = AudioConfig { - channels: 1, - vbr: true, - ..AudioConfig::music() - }; - let player = AudioPlayer::new(&AudioConfig { - channels: 2, - playback_device: "plug:original_default".into(), - ..config.clone() - })?; - - let mut raw_buf = vec![0u8; config.frame_size * 2 * 2]; - let mut pcm_out = vec![0i16; config.frame_size * 2]; - let mut left_pcm = vec![0i16; config.frame_size]; - let mut right_pcm = vec![0i16; config.frame_size]; - let mut opus_out = vec![0u8; 1500]; - let mut seq = 0u32; - - // 播放延迟: 只需覆盖网络延迟 + 时钟偏移 - let delay_us = 100_000; // 100ms 基础延迟 - let frame_duration_us = - (config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128; - - let mut stream_start_ts = 0; - let mut stream_start_seq = 0; - - let shutdown_flag_clone = shutdown_flag.clone(); - let audio_loop = async move { - loop { - if shutdown_flag_clone.load(Ordering::Relaxed) { - break; - } - - // 打开 FIFO - let mut fifo = match tokio::fs::File::open(AlsaRedirector::fifo_path()).await { - Ok(f) => f, - Err(_) => { - if shutdown_flag_clone.load(Ordering::Relaxed) { - break; - } - tokio::time::sleep(Duration::from_secs(1)).await; - continue; - } - }; - - // 每个新流开始时，重置编码器状态以避免残留音频导致爆音 - let mut left_encoder = OpusCodec::new(&encode_config)?; - let mut right_encoder = OpusCodec::new(&encode_config)?; - - loop { - if shutdown_flag_clone.load(Ordering::Relaxed) { - break; - } - - // 从 FIFO 读取 - if let Err(_) = fifo.read_exact(&mut raw_buf).await { - break; // FIFO 关闭，重新打开 - } - - let active_slaves = { - let s = slaves.lock().await; - if s.is_empty() { None } else { Some(s.clone()) } - }; - - let now = now_us(); - if stream_start_ts == 0 { - stream_start_ts = now; - stream_start_seq = seq; - } - - // 计算该帧应当播放的目标时间 - // target_ts = 数据包发送时间 + 播放延迟 - let target_ts = stream_start_ts - + ((seq - stream_start_seq) as u128 * frame_duration_us) - + delay_us; - - // 提取 PCM 数据 - for i in 0..config.frame_size { - left_pcm[i] = i16::from_le_bytes([raw_buf[i * 4], raw_buf[i * 4 + 1]]); - right_pcm[i] = i16::from_le_bytes([raw_buf[i * 4 + 2], raw_buf[i * 4 + 3]]); - } - - if let Some(slaves_list) = active_slaves { - // 情况 1: 有从节点，进行网络传输，并本地构造静音声道回放 - - // 1. 检查各声道是否有从节点需要 - let needs_left = slaves_list.iter().any(|s| s.role == ChannelRole::Left); - let needs_right = slaves_list.iter().any(|s| s.role == ChannelRole::Right); - - // 2. 编码需要的声道 - let mut left_bytes = None; - let mut right_bytes = None; - - if needs_left { - let len = left_encoder.encode(&left_pcm, &mut opus_out)?; - let packet = AudioPacket { - seq, - timestamp: target_ts, - data: opus_out[..len].to_vec(), - }; - left_bytes = Some(postcard::to_allocvec(&packet)?); - } - - if needs_right { - let len = right_encoder.encode(&right_pcm, &mut opus_out)?; - let packet = AudioPacket { - seq, - timestamp: target_ts, - data: opus_out[..len].to_vec(), - }; - right_bytes = Some(postcard::to_allocvec(&packet)?); - } - - // 3. 发送给对应的从节点 - for slave in &slaves_list { - let bytes = match slave.role { - ChannelRole::Left => left_bytes.as_ref(), - ChannelRole::Right => right_bytes.as_ref(), - }; - if let Some(b) = bytes { - let _ = audio_socket.send_to(b, slave.udp_addr).await; - } - } - - // 4. 将非本节点的声道置为静音 - for i in 0..config.frame_size { - match master_role { - ChannelRole::Left => { - pcm_out[i * 2] = left_pcm[i]; - pcm_out[i * 2 + 1] = 0; - } - ChannelRole::Right => { - pcm_out[i * 2] = 0; - pcm_out[i * 2 + 1] = right_pcm[i]; - } - } - } - - // 5. 等待播放 - let now = now_us(); - if now < target_ts { - let wait = target_ts - now; - if wait > 1000 { - tokio::time::sleep(Duration::from_micros(wait as u64)).await; - } else { - // 小于 1ms，直接播放，让播放时机稍微早一点点 - } - } - } else { - // 情况 2: 没有从节点，本地立体声播放 - for i in 0..config.frame_size { - pcm_out[i * 2] = left_pcm[i]; - pcm_out[i * 2 + 1] = right_pcm[i]; - } - } - - // 统一写入播放器 (始终是立体声) - if let Err(_) = player.write(&pcm_out) { - if shutdown_flag_clone.load(Ordering::Relaxed) { - break; - } - } - - seq += 1; - } - - // 重置流计时 - stream_start_ts = 0; - } - - Ok::<(), anyhow::Error>(()) - }; - - tokio::select! { - res = audio_loop => { - if let Err(e) = res { - eprintln!("❌ 音频循环错误: {:?}", e); - } - }, - _ = shutdown_signal() => { - // 设置退出标志，通知音频循环停止 - shutdown_flag.store(true, Ordering::Relaxed); - }, - } - - // 显式清理 - println!("👋 正在退出..."); - AlsaRedirector::cleanup(); - - // 强制退出 - std::process::exit(0); -} - -/// 监听系统退出信号 (SIGINT, SIGTERM, SIGQUIT) -async fn shutdown_signal() { - let mut sigint = signal(SignalKind::interrupt()).expect("无法注册 SIGINT 处理器"); - let mut sigterm = signal(SignalKind::terminate()).expect("无法注册 SIGTERM 处理器"); - let mut sigquit = signal(SignalKind::quit()).expect("无法注册 SIGQUIT 处理器"); - - tokio::select! { - _ = sigint.recv() => {}, - _ = sigterm.recv() => {}, - _ = sigquit.recv() => {}, - } -} - -/// 处理主节点与从节点的会话 -async fn handle_master_session( - mut control: ControlConnection, - audio_socket: Arc, - slaves: Arc>>, - client_tcp_addr: String, -) -> Result<()> { - let mut buf = [0u8; 1024]; - - // 握手 - let pkt = control.recv_packet(&mut buf).await?; - let slave_role = match pkt { - ControlPacket::ClientIdentify { role } => role, - _ => return Err(anyhow!("无效的握手协议")), - }; - - let xiao = ControlPacket::ServerHello { - udp_port: audio_socket.local_addr()?.port(), - }; - control.send_packet(&xiao).await?; - - // 等待 UDP 打洞/确认 - let mut buf = [0u8; 128]; - let (_, client_udp_addr) = audio_socket.recv_from(&mut buf).await?; - - println!( - "✅ 从节点已连接: {} {}", - client_tcp_addr, - slave_role.to_string(), - ); - - // 添加到从节点列表 - let session = SlaveSession { - udp_addr: client_udp_addr, - role: slave_role, - }; - { - let mut s = slaves.lock().await; - s.push(session.clone()); - } - - // 分离 TCP 读写，处理控制消息和心跳 - let (mut tcp_rx, mut tcp_tx) = control.split(); - - let mut buf = [0u8; 1024]; - loop { - match tcp_rx.read(&mut buf).await { - Ok(0) | Err(_) => { - break; - } - Ok(n) => { - if let Ok(ControlPacket::Ping { client_ts, seq }) = postcard::from_bytes(&buf[..n]) - { - let pong = ControlPacket::Pong { - client_ts, - server_ts: now_us(), - seq, - }; - if tcp_tx - .write_all(&postcard::to_allocvec(&pong).unwrap()) - .await - .is_err() - { - break; - } - } - } - } - } - - println!( - "❌ 从节点已断开: {} {}", - client_tcp_addr, - slave_role.to_string(), - ); - - // 从列表中移除 - { - let mut s = slaves.lock().await; - s.retain(|x| x.udp_addr != client_udp_addr); - } - - Ok(()) -} diff --git a/packages/client-v2/src/app/stereo/mod.rs b/packages/client-v2/src/app/stereo/mod.rs deleted file mode 100644 index 8af1333..0000000 --- a/packages/client-v2/src/app/stereo/mod.rs +++ /dev/null @@ -1,3 +0,0 @@ -pub mod entry; -pub mod master; -pub mod slave; diff --git a/packages/client-v2/src/app/stereo/slave.rs b/packages/client-v2/src/app/stereo/slave.rs deleted file mode 100644 index 7ab3d6e..0000000 --- a/packages/client-v2/src/app/stereo/slave.rs +++ /dev/null @@ -1,199 +0,0 @@ -#![cfg(target_os = "linux")] - -use crate::audio::codec::OpusCodec; -use crate::audio::config::AudioConfig; -use crate::audio::player::AudioPlayer; -use crate::net::discovery::Discovery; -use crate::net::network::SlaveNetwork; -use crate::net::protocol::{AudioPacket, ChannelRole, ControlPacket}; -use crate::utils::sync::{ClockSync, now_us}; -use anyhow::{Result, anyhow}; -use std::sync::Arc; -use std::time::Duration; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::sync::{Mutex, mpsc}; - -/// 运行从节点模式 -pub async fn run_slave(role: ChannelRole) -> Result<()> { - loop { - match handle_connection(role.clone()).await { - Err(e) => { - eprintln!("❌ {:?}", e); - tokio::time::sleep(Duration::from_secs(3)).await; - } - Ok(_) => {} - } - } -} - -async fn handle_connection(role: ChannelRole) -> Result<()> { - // 1. 发现主节点 - println!("🔍 正在扫描主节点..."); - let (master_ip, master_tcp_port) = Discovery::discover_master().await?; - let master_tcp_addr = format!("{}:{}", master_ip, master_tcp_port); - - // 2. 建立 TCP 连接 - println!("🔥 发现主节点: {}", master_tcp_addr); - let network = SlaveNetwork::connect(master_tcp_addr.parse()?).await?; - let (mut control, audio) = network.split(); - - // 3. 身份认证 - control - .send_packet(&ControlPacket::ClientIdentify { role: role.clone() }) - .await?; - - let mut buf = [0u8; 1024]; - let pkt = control.recv_packet(&mut buf).await?; - let server_udp_port = match pkt { - ControlPacket::ServerHello { udp_port } => udp_port, - _ => return Err(anyhow!("身份认证应答异常")), - }; - - // 4. UDP 打洞 - audio - .punch(format!("{}:{}", master_ip, server_udp_port).parse()?) - .await?; - - // 5. 初始化音频与同步组件 - let config = AudioConfig { - channels: 1, - ..AudioConfig::music() - }; - let player = AudioPlayer::new(&config)?; - let mut codec = OpusCodec::new(&config)?; - let clock = Arc::new(Mutex::new(ClockSync::new(100))); - - // 用于通知主循环 TCP 已断开的消息通道 - let (disconnect_tx, mut disconnect_rx) = mpsc::channel::<()>(1); - - // 6. 分离 TCP 读写 - let (mut tcp_rx, mut tcp_tx) = control.split(); - let clock_updater = clock.clone(); - let d_tx_ping = disconnect_tx.clone(); - let d_tx_pong = disconnect_tx.clone(); - - // 定时发送 Ping (心跳 & 时间同步) - let _sync_handle = tokio::spawn(async move { - let mut seq = 0; - loop { - let t1 = now_us(); - let msg = ControlPacket::Ping { client_ts: t1, seq }; - let data = postcard::to_allocvec(&msg).unwrap(); - if tcp_tx.write_all(&data).await.is_err() { - let _ = d_tx_ping.send(()).await; // 通知主线程 TCP 失败 - break; - } - tokio::time::sleep(Duration::from_millis(200)).await; - seq += 1; - } - }); - - // 接收 Pong - tokio::spawn(async move { - let mut buf = [0u8; 1024]; - loop { - match tcp_rx.read(&mut buf).await { - Ok(n) if n > 0 => { - if let Ok(ControlPacket::Pong { - client_ts, - server_ts, - .. - }) = postcard::from_bytes(&buf[..n]) - { - let t4 = now_us(); - clock_updater.lock().await.update(client_ts, server_ts, t4); - } - } - _ => { - let _ = d_tx_pong.send(()).await; // TCP 断开 - break; - } - } - } - }); - - // 7. 接收音频数据包 (UDP) - let (audio_tx, mut audio_rx) = mpsc::channel(100); - let audio_socket = audio.clone_inner(); - tokio::spawn(async move { - let mut buf = [0u8; 2048]; - loop { - if let Ok((len, _)) = audio_socket.recv_from(&mut buf).await { - if let Ok(packet) = postcard::from_bytes::(&buf[..len]) { - if audio_tx.send(packet).await.is_err() { - break; - } - } - } - } - }); - - // 8. 播放提示 - println!("✅ 主节点已连接，音频串流中..."); - let role_str = role.to_string(); - tokio::spawn(async move { - let _ = tokio::process::Command::new("sh") - .arg("-c") - .arg(format!( - "/usr/sbin/tts_play.sh \"主节点已连接，{}\" >/dev/null 2>&1", - role_str - )) - .status() - .await; - }); - - // 9. 播放主循环 - let mut pcm_buf = vec![0i16; config.frame_size]; - let mut last_seq: Option = None; - - loop { - // 检查 TCP 是否已断开 - if let Ok(_) = disconnect_rx.try_recv() { - return Err(anyhow!("主节点已断开: {}", master_tcp_addr)); - } - - // 接收数据包 - if let Ok(pkt) = audio_rx.try_recv() { - let now = now_us(); - let current_server_time = clock.lock().await.to_server_time(now); - - // 检查包是否迟到（目标时间已过） - if current_server_time > pkt.timestamp { - let late_ms = (current_server_time - pkt.timestamp) / 1000; - if late_ms > 50 { - // 迟到超过 50ms，直接丢弃 - continue; - } - // 轻微迟到（<50ms），尝试播放 - } - - last_seq = Some(pkt.seq); - - // 精确等待到播放时间 - loop { - let now = now_us(); - let current_server_time_precise = clock.lock().await.to_server_time(now); - - if current_server_time_precise >= pkt.timestamp { - break; - } - - let wait_us = (pkt.timestamp - current_server_time_precise) as u64; - - if wait_us > 1000 { - tokio::time::sleep(Duration::from_micros(wait_us as u64)).await; - } else { - // 小于 1ms，直接播放，让播放时机稍微早一点点 - break; - } - } - - // 解码并播放 - if let Ok(len) = codec.decode(&pkt.data, &mut pcm_buf) { - let _ = player.write(&pcm_buf[..len]); - } - } else { - tokio::time::sleep(Duration::from_micros(100)).await; - } - } -} diff --git a/packages/client-v2/src/audio/config.rs b/packages/client-v2/src/audio/config.rs index 1c982ef..4d06228 100644 --- a/packages/client-v2/src/audio/config.rs +++ b/packages/client-v2/src/audio/config.rs @@ -1,10 +1,12 @@ -#[derive(Debug, Clone, PartialEq, Eq, Hash)] +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum AudioScene { Music, Voice, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct AudioConfig { // ALSA 设备参数，用于录音和播放 pub capture_device: String, diff --git a/packages/client-v2/src/audio/mod.rs b/packages/client-v2/src/audio/mod.rs index 25d9907..2caf4ee 100644 --- a/packages/client-v2/src/audio/mod.rs +++ b/packages/client-v2/src/audio/mod.rs @@ -2,3 +2,4 @@ pub mod codec; pub mod config; pub mod player; pub mod recorder; +pub mod wav; diff --git a/packages/client-v2/src/audio/wav.rs b/packages/client-v2/src/audio/wav.rs new file mode 100644 index 0000000..261ab16 --- /dev/null +++ b/packages/client-v2/src/audio/wav.rs @@ -0,0 +1,106 @@ +use anyhow::Result; +use std::fs::File; +use std::io::{BufReader, BufWriter, Read, Seek, SeekFrom, Write}; + +pub struct WavWriter { + writer: BufWriter, + data_size: u32, + sample_rate: u32, + channels: u16, +} + +impl WavWriter { + pub fn create(path: &str, sample_rate: u32, channels: u16) -> Result { + let file = File::create(path)?; + let mut writer = BufWriter::new(file); + + // Header placeholder + writer.write_all(&[0u8; 44])?; + + Ok(Self { + writer, + data_size: 0, + sample_rate, + channels, + }) + } + + pub fn write_samples(&mut self, samples: &[i16]) -> Result<()> { + for &sample in samples { + self.writer.write_all(&sample.to_le_bytes())?; + self.data_size += 2; + } + Ok(()) + } + + 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; + let byte_rate = self.sample_rate * self.channels as u32 * 2; + let block_align = self.channels * 2; + + let mut header = [0u8; 44]; + header[0..4].copy_from_slice(b"RIFF"); + header[4..8].copy_from_slice(&file_size.to_le_bytes()); + header[8..12].copy_from_slice(b"WAVE"); + header[12..16].copy_from_slice(b"fmt "); + header[16..20].copy_from_slice(&16u32.to_le_bytes()); + header[20..22].copy_from_slice(&1u16.to_le_bytes()); // PCM + header[22..24].copy_from_slice(&self.channels.to_le_bytes()); + header[24..28].copy_from_slice(&self.sample_rate.to_le_bytes()); + header[28..32].copy_from_slice(&byte_rate.to_le_bytes()); + header[32..34].copy_from_slice(&block_align.to_le_bytes()); + header[34..36].copy_from_slice(&16u16.to_le_bytes()); // bits per sample + header[36..40].copy_from_slice(b"data"); + header[40..44].copy_from_slice(&self.data_size.to_le_bytes()); + + file.write_all(&header)?; + Ok(()) + } +} + +pub struct WavReader { + reader: BufReader, + pub sample_rate: u32, + pub channels: u16, + pub data_size: u32, +} + +impl WavReader { + pub fn open(path: &str) -> Result { + let file = File::open(path)?; + let mut reader = BufReader::new(file); + let mut header = [0u8; 44]; + reader.read_exact(&mut header)?; + + if &header[0..4] != b"RIFF" || &header[8..12] != b"WAVE" { + return Err(anyhow::anyhow!("Not a WAV file")); + } + + let channels = u16::from_le_bytes([header[22], header[23]]); + let sample_rate = u32::from_le_bytes([header[24], header[25], header[26], header[27]]); + let data_size = u32::from_le_bytes([header[40], header[41], header[42], header[43]]); + + Ok(Self { + reader, + sample_rate, + channels, + data_size, + }) + } + + pub fn read_samples(&mut self, samples: &mut [i16]) -> Result { + 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]]); + } + Ok(sample_count) + } +} + diff --git a/packages/client-v2/src/bin/client.rs b/packages/client-v2/src/bin/client.rs index aede7f7..f5df377 100644 --- a/packages/client-v2/src/bin/client.rs +++ b/packages/client-v2/src/bin/client.rs @@ -6,9 +6,5 @@ async fn main() -> Result<()> { { xiao::app::client::entry::run_client().await?; } - #[cfg(not(target_os = "linux"))] - { - eprintln!("Only support Linux"); - } Ok(()) } diff --git a/packages/client-v2/src/bin/stereo.rs b/packages/client-v2/src/bin/stereo.rs deleted file mode 100644 index fa4ff1c..0000000 --- a/packages/client-v2/src/bin/stereo.rs +++ /dev/null @@ -1,14 +0,0 @@ -use anyhow::Result; - -#[tokio::main] -async fn main() -> Result<()> { - #[cfg(feature = "app")] - { - xiao::app::stereo::entry::run_stereo().await?; - } - #[cfg(not(target_os = "linux"))] - { - eprintln!("Only support Linux"); - } - Ok(()) -} diff --git a/packages/client-v2/src/net/discovery.rs b/packages/client-v2/src/net/discovery.rs index e4f069f..4d66bbc 100644 --- a/packages/client-v2/src/net/discovery.rs +++ b/packages/client-v2/src/net/discovery.rs @@ -10,13 +10,13 @@ pub const DISCOVERY_PORT: u16 = 53530; pub struct Discovery; impl Discovery { - /// 主节点：启动广播，告知从节点自己的 TCP 端口 + /// 服务端：启动广播，告知客户端自己的 TCP 端口 pub async fn start_broadcast(tcp_port: u16) -> Result<()> { let socket = UdpSocket::bind("0.0.0.0:0").await?; socket.set_broadcast(true)?; let target: SocketAddr = format!("255.255.255.255:{}", DISCOVERY_PORT).parse()?; - let msg = postcard::to_allocvec(&ControlPacket::ServerHello { udp_port: tcp_port })?; + let msg = postcard::to_allocvec(&ControlPacket::ServerHello { tcp_port })?; tokio::spawn(async move { loop { @@ -28,17 +28,17 @@ impl Discovery { Ok(()) } - /// 从节点：监听广播，发现主节点的 IP 和 TCP 端口 - pub async fn discover_master() -> Result<(IpAddr, u16)> { + /// 客户端：监听广播，发现服务端的 IP 和 TCP 端口 + pub async fn discover_server() -> Result<(IpAddr, u16)> { let socket = UdpSocket::bind(format!("0.0.0.0:{}", DISCOVERY_PORT)).await?; let mut buf = [0u8; 1024]; loop { let (len, addr) = socket.recv_from(&mut buf).await?; - if let Ok(ControlPacket::ServerHello { udp_port }) = + if let Ok(ControlPacket::ServerHello { tcp_port }) = postcard::from_bytes::(&buf[..len]) { - return Ok((addr.ip(), udp_port)); + return Ok((addr.ip(), tcp_port)); } } } diff --git a/packages/client-v2/src/net/network.rs b/packages/client-v2/src/net/network.rs index 69dd573..3040ab4 100644 --- a/packages/client-v2/src/net/network.rs +++ b/packages/client-v2/src/net/network.rs @@ -34,11 +34,6 @@ impl AudioSocket { Ok((packet, addr)) } - pub async fn punch(&self, target: SocketAddr) -> Result<()> { - self.socket.send_to(&[0u8; 1], target).await?; - Ok(()) - } - pub fn clone_inner(&self) -> Arc { self.socket.clone() } @@ -56,16 +51,20 @@ impl ControlConnection { pub async fn send_packet(&mut self, packet: &ControlPacket) -> Result<()> { let bytes = postcard::to_allocvec(packet)?; + let len = bytes.len() as u32; + self.stream.write_u32(len).await?; self.stream.write_all(&bytes).await?; Ok(()) } - pub async fn recv_packet(&mut self, buf: &mut [u8]) -> Result { - let len = self.stream.read(buf).await?; - if len == 0 { - return Err(anyhow::anyhow!("连接已关闭")); + pub async fn recv_packet(&mut self) -> Result { + let len = self.stream.read_u32().await? as usize; + if len > 10 * 1024 * 1024 { + return Err(anyhow::anyhow!("Packet too large: {}", len)); } - let packet = postcard::from_bytes(&buf[..len])?; + let mut buf = vec![0u8; len]; + self.stream.read_exact(&mut buf).await?; + let packet = postcard::from_bytes(&buf)?; Ok(packet) } @@ -77,19 +76,21 @@ impl ControlConnection { ) { self.stream.into_split() } + + pub fn peer_addr(&self) -> Result { + self.stream.peer_addr().context("Failed to get peer addr") + } } -/// 主节点网络管理器 -pub struct MasterNetwork { +/// 服务端网络管理器 +pub struct ServerNetwork { listener: TcpListener, - audio: AudioSocket, } -impl MasterNetwork { +impl ServerNetwork { pub async fn setup(port: u16) -> Result { let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await?; - let audio = AudioSocket::bind().await?; - Ok(Self { listener, audio }) + Ok(Self { listener }) } pub async fn accept(&self) -> Result<(ControlConnection, SocketAddr)> { @@ -97,30 +98,29 @@ impl MasterNetwork { Ok((ControlConnection::new(stream), addr)) } - pub fn audio_socket(&self) -> &AudioSocket { - &self.audio + pub fn local_addr(&self) -> Result { + self.listener + .local_addr() + .context("Failed to get local addr") } } -/// 从节点网络管理器 -pub struct SlaveNetwork { +/// 客户端网络管理器 +pub struct ClientNetwork { control: ControlConnection, - audio: AudioSocket, } -impl SlaveNetwork { - pub async fn connect(master_addr: SocketAddr) -> Result { - let stream = TcpStream::connect(master_addr) +impl ClientNetwork { + pub async fn connect(server_addr: SocketAddr) -> Result { + let stream = TcpStream::connect(server_addr) .await - .context(format!("无法连接到主节点 TCP 地址: {}", master_addr))?; - let audio = AudioSocket::bind().await?; + .context(format!("无法连接到服务端 TCP 地址: {}", server_addr))?; Ok(Self { control: ControlConnection::new(stream), - audio, }) } - pub fn split(self) -> (ControlConnection, AudioSocket) { - (self.control, self.audio) + pub fn into_control(self) -> ControlConnection { + self.control } } diff --git a/packages/client-v2/src/net/protocol.rs b/packages/client-v2/src/net/protocol.rs index c699019..ad5bd1d 100644 --- a/packages/client-v2/src/net/protocol.rs +++ b/packages/client-v2/src/net/protocol.rs @@ -1,45 +1,59 @@ +use crate::audio::config::AudioConfig; use serde::{Deserialize, Serialize}; -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Copy)] -pub enum ChannelRole { - Left, - Right, -} - -impl ChannelRole { - pub fn to_string(&self) -> String { - match self { - ChannelRole::Left => "左声道".to_string(), - ChannelRole::Right => "右声道".to_string(), - } - } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub struct DeviceInfo { + pub model: String, + pub mac: String, + pub version: u32, } #[derive(Serialize, Deserialize, Debug, Clone)] pub enum ControlPacket { - // 发现协议 + // 服务发现 ServerHello { - udp_port: u16, // UDP 音频流端口 + tcp_port: u16, }, - // 握手协议 + // 握手与认证 ClientIdentify { - role: ChannelRole, + info: DeviceInfo, }, - // 时间同步 (持续进行) - Ping { - client_ts: u128, - seq: u32, + IdentifyOk, + + // 音频控制 + StartRecording { + config: AudioConfig, }, - Pong { - client_ts: u128, - server_ts: u128, - seq: u32, + StopRecording, + StartPlayback { + config: AudioConfig, }, + StopPlayback, + + // RPC + RpcRequest { + id: u32, + method: String, + args: Vec, + }, + RpcResponse { + id: u32, + result: RpcResult, + }, + + // 心跳 + Ping, + Pong, +} + +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct RpcResult { + pub stdout: String, + pub stderr: String, + pub code: i32, } #[derive(Serialize, Deserialize, Debug, Clone)] pub struct AudioPacket { - pub seq: u32, // 序列号，用于丢包检测 - pub timestamp: u128, // 目标播放时间 (主节点时间) - pub data: Vec, // Opus 编码数据 + pub data: Vec, // Opus 编码数据 }