diff --git a/packages/client-v2/src/app/client/core.rs b/packages/client-v2/src/app/client/core.rs deleted file mode 100644 index aba1768..0000000 --- a/packages/client-v2/src/app/client/core.rs +++ /dev/null @@ -1,115 +0,0 @@ -#![cfg(target_os = "linux")] - -use crate::app::client::handlers; -use crate::app::client::session::ClientSession; -use crate::audio::config::AudioConfig; -use crate::net::discovery::Discovery; -use crate::net::network::{AudioSocket, ClientNetwork}; -use crate::net::protocol::{ControlPacket, DeviceInfo, RpcResult}; -use anyhow::{Context, Result}; -use std::net::SocketAddr; -use std::sync::Arc; -use tokio::sync::broadcast; - -pub struct Client { - info: DeviceInfo, - session: Arc>>>, -} - -impl Client { - pub fn new(model: &str, mac: &str, version: u32) -> Self { - Self { - info: DeviceInfo { - model: model.to_string(), - mac: mac.to_string(), - version, - }, - session: Arc::new(tokio::sync::Mutex::new(None)), - } - } - - pub async fn run(self: Arc) -> 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 (reader, writer) = control.split(); - let session = Arc::new(ClientSession::new(self.info.clone(), writer)); - *self.session.lock().await = Some(session.clone()); - - let (stop_tx, _) = broadcast::channel::<()>(1); - let audio_socket = Arc::new(AudioSocket::bind().await?); - - // 启动测试 RPC 调用的任务 - let client_clone = self.clone(); - tokio::spawn(async move { - tokio::time::sleep(std::time::Duration::from_secs(2)).await; - println!("测试向服务端发起 RPC: status"); - match client_clone.call_server("status", vec![]).await { - Ok(res) => println!("收到服务端响应: {:?}", res), - Err(e) => eprintln!("向服务端发起 RPC 失败: {}", e), - } - }); - - let mut reader = reader; - loop { - let packet = reader.recv_packet().await?; - let self_clone = self.clone(); - let session_clone = session.clone(); - let audio_socket = audio_socket.clone(); - let stop_tx = stop_tx.clone(); - - tokio::spawn(async move { - if let Err(e) = handlers::handle_packet( - self_clone, - session_clone, - packet, - audio_socket, - stop_tx, - server_addr, - ) - .await - { - eprintln!("处理控制包出错: {}", e); - } - }); - } - } - - pub async fn call_server(&self, method: &str, args: Vec) -> Result { - let session = self - .session - .lock() - .await - .clone() - .context("Session not established")?; - let (id, rx) = session.rpc.alloc_id(); - session - .writer - .lock() - .await - .send_packet(&ControlPacket::RpcRequest { - id, - method: method.to_string(), - args, - }) - .await?; - Ok(rx.await?) - } -} diff --git a/packages/client-v2/src/app/client/entry.rs b/packages/client-v2/src/app/client/entry.rs deleted file mode 100644 index 409c849..0000000 --- a/packages/client-v2/src/app/client/entry.rs +++ /dev/null @@ -1,14 +0,0 @@ -#![cfg(target_os = "linux")] - -use crate::app::client::core::Client; -use anyhow::Result; - -pub async fn run_client() -> Result<()> { - // 模拟从系统获取信息 - let model = "XiaoAi-V2-Simulated"; - let mac = "00:11:22:33:44:55"; - let version = 1; - - let client = std::sync::Arc::new(Client::new(model, mac, version)); - client.run().await -} diff --git a/packages/client-v2/src/app/client/handlers.rs b/packages/client-v2/src/app/client/handlers.rs deleted file mode 100644 index 58a82a3..0000000 --- a/packages/client-v2/src/app/client/handlers.rs +++ /dev/null @@ -1,202 +0,0 @@ -#![cfg(target_os = "linux")] - -use crate::app::client::core::Client; -use crate::app::client::session::ClientSession; -use crate::audio::codec::OpusCodec; -use crate::audio::config::AudioConfig; -use crate::audio::player::AudioPlayer; -use crate::audio::recorder::AudioRecorder; -use crate::net::network::AudioSocket; -use crate::net::protocol::{AudioPacket, ControlPacket, RpcResult}; -use anyhow::Result; -use std::net::SocketAddr; -use std::sync::Arc; -use tokio::sync::broadcast; - -pub async fn handle_packet( - client: Arc, - session: Arc, - packet: ControlPacket, - audio_socket: Arc, - stop_tx: broadcast::Sender<()>, - server_addr: SocketAddr, -) -> Result<()> { - match packet { - ControlPacket::RpcResponse { id, result } => { - session.rpc.fulfill(id, result); - } - ControlPacket::RpcRequest { id, method, args } => { - println!("收到 RPC 请求: {} {:?}", method, args); - let result = handle_rpc(&method, args).await; - let response = ControlPacket::RpcResponse { id, result }; - session.writer.lock().await.send_packet(&response).await?; - } - ControlPacket::StartRecording { config } => { - // ... - println!("开始录音: {:?}", config); - let mut stop_rx = stop_tx.subscribe(); - tokio::spawn(async move { - if let Err(e) = handle_recording(config, audio_socket, server_addr, stop_rx).await { - eprintln!("录音出错: {}", e); - } - }); - } - ControlPacket::StopRecording => { - println!("停止录音"); - let _ = stop_tx.send(()); - } - ControlPacket::StartPlayback { config } => { - println!("开始播放: {:?}", config); - let mut stop_rx = stop_tx.subscribe(); - tokio::spawn(async move { - if let Err(e) = handle_playback(config, audio_socket, stop_rx).await { - eprintln!("播放出错: {}", e); - } - }); - } - ControlPacket::StopPlayback => { - println!("停止播放"); - let _ = stop_tx.send(()); - } - ControlPacket::Ping => { - let _ = session - .writer - .lock() - .await - .send_packet(&ControlPacket::Pong) - .await; - } - _ => {} - } - Ok(()) -} - -pub async fn handle_rpc(method: &str, args: Vec) -> RpcResult { - match method { - "shell" => handle_shell(args).await, - _ => RpcResult { - stdout: "".to_string(), - stderr: format!("Unknown method: {}", method), - code: -1, - }, - } -} - -async fn handle_shell(args: Vec) -> RpcResult { - if args.is_empty() { - return RpcResult { - stdout: "".to_string(), - stderr: "Missing command argument".to_string(), - code: -1, - }; - } - let cmd = &args[0]; - println!("Executing 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!("Mock execution of {} successful", cmd), - stderr: "".to_string(), - code: 0, - } - } -} - -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; - } - 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(()) -} diff --git a/packages/client-v2/src/app/client/mod.rs b/packages/client-v2/src/app/client/mod.rs index cc15fe5..329f359 100644 --- a/packages/client-v2/src/app/client/mod.rs +++ b/packages/client-v2/src/app/client/mod.rs @@ -1,4 +1,234 @@ -pub mod core; -pub mod entry; -pub mod handlers; -pub mod session; +#![cfg(target_os = "linux")] + +use crate::audio::codec::OpusCodec; +use crate::audio::player::AudioPlayer; +use crate::audio::recorder::AudioRecorder; +use crate::net::discovery::Discovery; +use crate::net::network::{AudioSocket, Connection}; +use crate::net::protocol::{AudioPacket, ControlPacket, DeviceInfo, RpcResult}; +use crate::net::rpc::RpcManager; +use anyhow::Result; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::sync::{Mutex, broadcast}; + +pub struct Client { + info: DeviceInfo, + conn: Mutex>>, + rpc: Arc, + server_audio_addr: Mutex>, +} + +impl Client { + pub fn new() -> Self { + Self { + info: DeviceInfo::current(), + conn: Mutex::new(None), + rpc: Arc::new(RpcManager::new()), + server_audio_addr: Mutex::new(None), + } + } + + pub async fn run(self: Arc) -> Result<()> { + let (ip, tcp_port, udp_port) = Discovery::listen().await?; + let addr = SocketAddr::new(ip, tcp_port); + let audio_addr = SocketAddr::new(ip, udp_port); + println!("Found server at {}, audio at {}", addr, audio_addr); + *self.server_audio_addr.lock().await = Some(audio_addr); + + println!("Connecting to TCP server at {}...", addr); + let stream = tokio::net::TcpStream::connect(addr).await?; + let conn = Arc::new(Connection::new(stream)?); + println!("TCP connected, sending identification..."); + + let audio = Arc::new(AudioSocket::bind().await?); + conn.send(&ControlPacket::ClientIdentify { + info: self.info.clone(), + udp_port: audio.port(), + }) + .await?; + match conn.recv().await? { + ControlPacket::IdentifyOk => println!("Connected to server"), + p => return Err(anyhow::anyhow!("Handshake failed: {:?}", p)), + } + + *self.conn.lock().await = Some(conn.clone()); + let (stop_tx, _) = broadcast::channel(1); + + loop { + let packet = conn.recv().await?; + let this = self.clone(); + let audio = audio.clone(); + let stop_tx = stop_tx.clone(); + let audio_addr = self.server_audio_addr.lock().await.unwrap(); + tokio::spawn(async move { + if let Err(e) = this.handle_packet(packet, audio, stop_tx, audio_addr).await { + eprintln!("Handle packet error: {}", e); + } + }); + } + } + + async fn handle_packet( + &self, + packet: ControlPacket, + audio: Arc, + stop_tx: broadcast::Sender<()>, + server_addr: SocketAddr, + ) -> Result<()> { + match packet { + ControlPacket::RpcRequest { id, method, args } => { + let result = self.handle_rpc(&method, args).await; + if let Some(conn) = self.conn.lock().await.as_ref() { + conn.send(&ControlPacket::RpcResponse { id, result }) + .await?; + } + } + ControlPacket::RpcResponse { id, result } => self.rpc.resolve(id, result), + ControlPacket::StartRecording { config } => { + let mut stop_rx = stop_tx.subscribe(); + tokio::spawn(async move { + let (pcm_tx, mut pcm_rx) = tokio::sync::mpsc::channel::>(20); + + // 录音线程：使用 std::thread 处理阻塞的 ALSA 调用 + let config_clone = config.clone(); + std::thread::spawn(move || { + let recorder = match AudioRecorder::new(&config_clone) { + Ok(r) => r, + Err(e) => { + eprintln!("Failed to start recorder: {}", e); + return; + } + }; + loop { + let mut pcm = vec![0i16; config_clone.frame_size]; + match recorder.read(&mut pcm) { + Ok(n) => { + if pcm_tx.blocking_send(pcm[..n].to_vec()).is_err() { + break; // Receiver dropped, stop recording + } + } + Err(e) => { + eprintln!("Recorder read error: {}", e); + break; + } + } + } + }); + + let mut codec = match OpusCodec::new(&config) { + Ok(c) => c, + Err(e) => { + eprintln!("Failed to init opus codec: {}", e); + return; + } + }; + + println!("Recording started..."); + loop { + tokio::select! { + _ = stop_rx.recv() => break, + Some(pcm_data) = pcm_rx.recv() => { + let mut opus = vec![0u8; 4096]; + if let Ok(len) = codec.encode(&pcm_data, &mut opus) { + let _ = audio.send(&AudioPacket { data: opus[..len].to_vec() }, server_addr).await; + } + } + } + } + println!("Recording stopped."); + }); + } + ControlPacket::StartPlayback { config } => { + let mut stop_rx = stop_tx.subscribe(); + tokio::spawn(async move { + let (pcm_tx, mut pcm_rx) = tokio::sync::mpsc::channel::>(20); + + // 播放线程：使用 std::thread 处理阻塞的 ALSA 调用 + let config_clone = config.clone(); + std::thread::spawn(move || { + let player = match AudioPlayer::new(&config_clone) { + Ok(p) => p, + Err(e) => { + eprintln!("Failed to start player: {}", e); + return; + } + }; + while let Some(pcm_data) = pcm_rx.blocking_recv() { + let _ = player.write(&pcm_data); + } + }); + + let mut codec = match OpusCodec::new(&config) { + Ok(c) => c, + Err(e) => { + eprintln!("Failed to init opus codec: {}", e); + return; + } + }; + + let mut buf = vec![0u8; 4096]; + println!("Playback started with jitter buffer..."); + loop { + tokio::select! { + _ = stop_rx.recv() => break, + res = audio.recv(&mut buf) => { + match res { + Ok((packet, _)) => { + let mut pcm = vec![0i16; config.frame_size]; + if let Ok(n) = codec.decode(&packet.data, &mut pcm) { + let _ = pcm_tx.send(pcm[..n].to_vec()).await; + } + } + Err(e) => { + eprintln!("Audio recv error: {}", e); + break; + } + } + } + } + } + println!("Playback stopped."); + }); + } + ControlPacket::StopRecording | ControlPacket::StopPlayback => { + let _ = stop_tx.send(()); + } + ControlPacket::Ping => { + if let Some(conn) = self.conn.lock().await.as_ref() { + conn.send(&ControlPacket::Pong).await?; + } + } + _ => {} + } + Ok(()) + } + + async fn handle_rpc(&self, method: &str, args: Vec) -> RpcResult { + match method { + "shell" if !args.is_empty() => { + let output = std::process::Command::new("sh") + .arg("-c") + .arg(&args[0]) + .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 { + stderr: e.to_string(), + code: -1, + ..Default::default() + }, + } + } + _ => RpcResult { + stderr: "Unsupported method".to_string(), + code: -1, + ..Default::default() + }, + } + } +} diff --git a/packages/client-v2/src/app/client/session.rs b/packages/client-v2/src/app/client/session.rs deleted file mode 100644 index ee85af6..0000000 --- a/packages/client-v2/src/app/client/session.rs +++ /dev/null @@ -1,20 +0,0 @@ -use crate::net::network::ControlWriter; -use crate::net::protocol::DeviceInfo; -use crate::net::rpc::RpcManager; -use std::sync::Arc; - -pub struct ClientSession { - pub info: DeviceInfo, - pub writer: Arc>, - pub rpc: Arc, -} - -impl ClientSession { - pub fn new(info: DeviceInfo, writer: ControlWriter) -> Self { - Self { - info, - writer: Arc::new(tokio::sync::Mutex::new(writer)), - rpc: Arc::new(RpcManager::new()), - } - } -} diff --git a/packages/client-v2/src/app/server/core.rs b/packages/client-v2/src/app/server/core.rs deleted file mode 100644 index 2552c79..0000000 --- a/packages/client-v2/src/app/server/core.rs +++ /dev/null @@ -1,252 +0,0 @@ -use crate::app::server::handlers; -use crate::app::server::session::ServerSession; -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, RpcResult}; -use anyhow::{Context, Result}; -use parking_lot::Mutex; -use std::collections::HashMap; -use std::net::SocketAddr; -use std::sync::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 (mut reader, writer) = control.split(); - let session = Arc::new(ServerSession::new(info, writer, addr)); - - self.sessions.lock().insert(addr, session.clone()); - - // 处理控制消息循环 - loop { - let packet = reader.recv_packet().await?; - let session = session.clone(); - tokio::spawn(async move { - if let Err(e) = handlers::handle_packet(session, packet).await { - eprintln!("处理来自 {} 的包时出错: {}", addr, e); - } - }); - } - } - - // 暴露给外部调用的方法 - 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 - .writer - .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 - .writer - .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 - .writer - .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 - .writer - .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 - .writer - .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); - - 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?; - - 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 deleted file mode 100644 index 3378f5a..0000000 --- a/packages/client-v2/src/app/server/entry.rs +++ /dev/null @@ -1,57 +0,0 @@ -use crate::app::server::core::Server; -use crate::audio::config::AudioConfig; -use anyhow::Result; -use std::sync::Arc; - -pub async fn run_server() -> Result<()> { - 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/handlers.rs b/packages/client-v2/src/app/server/handlers.rs deleted file mode 100644 index c84f3f3..0000000 --- a/packages/client-v2/src/app/server/handlers.rs +++ /dev/null @@ -1,50 +0,0 @@ -use crate::app::server::session::ServerSession; -use crate::net::protocol::{ControlPacket, RpcResult}; -use anyhow::Result; -use std::sync::Arc; - -pub async fn handle_packet(session: Arc, packet: ControlPacket) -> Result<()> { - match packet { - ControlPacket::RpcResponse { id, result } => { - session.rpc.fulfill(id, result); - } - ControlPacket::RpcRequest { id, method, args } => { - println!( - "收到来自客户端 {} 的 RPC 请求: {} {:?}", - session.addr, method, args - ); - let result = handle_server_rpc(&method, args).await; - session - .writer - .lock() - .await - .send_packet(&ControlPacket::RpcResponse { id, result }) - .await?; - } - ControlPacket::Ping => { - session - .writer - .lock() - .await - .send_packet(&ControlPacket::Pong) - .await?; - } - _ => {} - } - Ok(()) -} - -async fn handle_server_rpc(method: &str, _args: Vec) -> RpcResult { - match method { - "status" => RpcResult { - stdout: "Server is running normally".to_string(), - stderr: "".to_string(), - code: 0, - }, - _ => RpcResult { - stdout: "".to_string(), - stderr: format!("Server does not support method: {}", method), - code: -1, - }, - } -} diff --git a/packages/client-v2/src/app/server/mod.rs b/packages/client-v2/src/app/server/mod.rs index cc15fe5..bb3b331 100644 --- a/packages/client-v2/src/app/server/mod.rs +++ b/packages/client-v2/src/app/server/mod.rs @@ -1,4 +1,224 @@ -pub mod core; -pub mod entry; -pub mod handlers; -pub mod session; +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, Connection}; +use crate::net::protocol::{AudioPacket, ControlPacket, DeviceInfo, RpcResult}; +use crate::net::rpc::RpcManager; +use anyhow::{Context, Result}; +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::sync::Mutex; + +pub struct Session { + pub info: DeviceInfo, + pub conn: Arc, + pub rpc: Arc, + pub audio_addr: SocketAddr, +} + +pub struct Server { + sessions: Arc>>>, + audio: Arc, +} + +impl Server { + pub async fn new() -> Result { + Ok(Self { + sessions: Arc::new(Mutex::new(HashMap::new())), + audio: Arc::new(AudioSocket::bind().await?), + }) + } + + pub async fn run(self: Arc, port: u16) -> Result<()> { + let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{}", port)).await?; + let addr = listener.local_addr()?; + println!("Server listening on TCP: {}", addr); + Discovery::broadcast(port, self.audio.port()).await?; + + loop { + let (stream, addr) = listener.accept().await?; + let server = self.clone(); + tokio::spawn(async move { + if let Err(e) = server.handle_connection(stream, addr).await { + eprintln!("Session {} error: {}", addr, e); + } + server.sessions.lock().await.remove(&addr); + }); + } + } + + async fn handle_connection( + &self, + stream: tokio::net::TcpStream, + addr: SocketAddr, + ) -> Result<()> { + println!("New TCP connection from {}", addr); + let conn = Arc::new(Connection::new(stream)?); + + let (info, client_udp_port) = match conn.recv().await? { + ControlPacket::ClientIdentify { info, udp_port } => (info, udp_port), + p => { + println!("Expected Identify from {}, got {:?}", addr, p); + return Err(anyhow::anyhow!("Expected Identify, got {:?}", p)); + } + }; + + let audio_addr = SocketAddr::new(addr.ip(), client_udp_port); + println!( + "Client identified: {} ({}) version {}, audio at {}", + info.model, addr, info.version, audio_addr + ); + conn.send(&ControlPacket::IdentifyOk).await?; + + let session = Arc::new(Session { + info, + conn: conn.clone(), + rpc: Arc::new(RpcManager::new()), + audio_addr, + }); + + self.sessions.lock().await.insert(addr, session.clone()); + + loop { + let packet = conn.recv().await?; + let session = session.clone(); + tokio::spawn(async move { + if let Err(e) = handle_packet(session, packet).await { + eprintln!("Handle packet error: {}", e); + } + }); + } + } + + pub async fn get_clients(&self) -> Vec { + self.sessions.lock().await.keys().cloned().collect() + } + + pub async fn call( + &self, + addr: SocketAddr, + method: &str, + args: Vec, + ) -> Result { + let session = self + .sessions + .lock() + .await + .get(&addr) + .cloned() + .context("Session not found")?; + let (id, rx) = session.rpc.register(); + session + .conn + .send(&ControlPacket::RpcRequest { + id, + method: method.to_string(), + args, + }) + .await?; + Ok(rx.await?) + } + + pub async fn start_record(&self, addr: SocketAddr, config: AudioConfig) -> Result<()> { + let session = self + .sessions + .lock() + .await + .get(&addr) + .cloned() + .context("Session not found")?; + session + .conn + .send(&ControlPacket::StartRecording { + config: config.clone(), + }) + .await?; + let audio = self.audio.clone(); + tokio::spawn(async move { + let mut writer = + WavWriter::create("temp/recorded.wav", config.sample_rate, config.channels) + .unwrap(); + let mut codec = OpusCodec::new(&config).unwrap(); + let mut pcm = vec![0i16; config.frame_size]; + let mut buf = vec![0u8; 4096]; + for _ in 0..500 { + // Record ~10s + if let Ok((packet, _)) = audio.recv(&mut buf).await { + if let Ok(n) = codec.decode(&packet.data, &mut pcm) { + writer.write_samples(&pcm[..n]).unwrap(); + } + } + } + writer.finalize().unwrap(); + println!("Recording saved to temp/recorded.wav"); + }); + Ok(()) + } + + pub async fn start_play(&self, addr: SocketAddr) -> Result<()> { + let session = self + .sessions + .lock() + .await + .get(&addr) + .cloned() + .context("Session not found")?; + let reader = WavReader::open("temp/test.wav")?; + + let opus_rate = if reader.sample_rate > 24000 { + 48000 + } else { + 16000 + }; + + let config = AudioConfig { + sample_rate: opus_rate, + channels: reader.channels, + frame_size: (opus_rate / 50) as usize, // 20ms + ..AudioConfig::music_48k() + }; + + session + .conn + .send(&ControlPacket::StartPlayback { + config: config.clone(), + }) + .await?; + let audio = self.audio.clone(); + let target_addr = session.audio_addr; + tokio::spawn(async move { + let mut reader = reader; + let mut codec = OpusCodec::new(&config).unwrap(); + let mut pcm = vec![0i16; config.frame_size]; + let mut opus = vec![0u8; 4096]; + while let Ok(n) = reader.read_samples(&mut pcm) { + if n == 0 { + break; + } + if let Ok(len) = codec.encode(&pcm[..n], &mut opus) { + let _ = audio + .send( + &AudioPacket { + data: opus[..len].to_vec(), + }, + target_addr, + ) + .await; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + }); + Ok(()) + } +} + +async fn handle_packet(session: Arc, packet: ControlPacket) -> Result<()> { + match packet { + ControlPacket::RpcResponse { id, result } => session.rpc.resolve(id, result), + ControlPacket::Ping => session.conn.send(&ControlPacket::Pong).await?, + _ => {} + } + Ok(()) +} diff --git a/packages/client-v2/src/app/server/session.rs b/packages/client-v2/src/app/server/session.rs deleted file mode 100644 index d3e498c..0000000 --- a/packages/client-v2/src/app/server/session.rs +++ /dev/null @@ -1,23 +0,0 @@ -use crate::net::network::ControlWriter; -use crate::net::protocol::DeviceInfo; -use crate::net::rpc::RpcManager; -use std::net::SocketAddr; -use std::sync::Arc; - -pub struct ServerSession { - pub info: DeviceInfo, - pub writer: Arc>, - pub addr: SocketAddr, - pub rpc: Arc, -} - -impl ServerSession { - pub fn new(info: DeviceInfo, writer: ControlWriter, addr: SocketAddr) -> Self { - Self { - info, - writer: Arc::new(tokio::sync::Mutex::new(writer)), - addr, - rpc: Arc::new(RpcManager::new()), - } - } -} diff --git a/packages/client-v2/src/audio/codec.rs b/packages/client-v2/src/audio/codec.rs index 641d127..ab6b5f3 100644 --- a/packages/client-v2/src/audio/codec.rs +++ b/packages/client-v2/src/audio/codec.rs @@ -9,62 +9,67 @@ pub struct OpusCodec { impl OpusCodec { pub fn new(config: &AudioConfig) -> Result { + // Opus 仅支持这些采样率，强制进行转换以防万一 + let opus_rate = match config.sample_rate { + 8000 => 8000, + 12000 => 12000, + 16000 => 16000, + 24000 => 24000, + 48000 => 48000, + _ => { + let fallback = if config.sample_rate < 24000 { + 16000 + } else { + 48000 + }; + println!( + "Warning: Opus does not support {}Hz, falling back to {}Hz", + config.sample_rate, fallback + ); + fallback + } + }; + let channels = match config.channels { 1 => Channels::Mono, 2 => Channels::Stereo, - _ => return Err(anyhow::anyhow!("Invalid channels: {}", config.channels)), + _ => return Err(anyhow::anyhow!("Unsupported channels: {}", config.channels)), }; + let mode = match config.audio_scene { AudioScene::Music => Application::Audio, AudioScene::Voice => Application::Voip, }; - let bitrate = match config.bitrate { - -1 => Bitrate::Max, - 0 => Bitrate::Auto, - _ => Bitrate::Bits(config.bitrate), + + let mut encoder = + Encoder::new(opus_rate, channels, mode).context("Opus encoder init failed")?; + + let bitrate = if config.bitrate <= 0 { + Bitrate::Auto + } else { + Bitrate::Bits(config.bitrate) }; - let mut encoder = Encoder::new(config.sample_rate, channels, mode) - .context("Failed to create Opus encoder")?; - encoder.set_bitrate(bitrate)?; - if config.vbr { - encoder.set_vbr(true)?; - } + encoder.set_vbr(config.vbr)?; if config.fec { - encoder.set_inband_fec(true)?; // 内联前向纠错 - encoder.set_packet_loss_perc(20)?; // 预期丢包率20% + encoder.set_inband_fec(true)?; + encoder.set_packet_loss_perc(10)?; } let decoder = - Decoder::new(config.sample_rate, channels).context("Failed to create Opus decoder")?; + Decoder::new(config.sample_rate, channels).context("Opus decoder init failed")?; Ok(Self { encoder, decoder }) } pub fn encode(&mut self, pcm: &[i16], out: &mut [u8]) -> Result { - self.encoder - .encode(pcm, out) - .context("Opus encoding failed") + self.encoder.encode(pcm, out).context("Opus encode failed") } pub fn decode(&mut self, opus: &[u8], out: &mut [i16]) -> Result { self.decoder .decode(opus, out, false) - .context("Opus decoding failed") - } - - /// 前向纠错(FEC) - pub fn decode_fec(&mut self, opus: &[u8], out: &mut [i16]) -> Result { - self.decoder - .decode(opus, out, true) - .context("Opus FEC decoding failed") - } - - /// 丢包补偿(PLC) - pub fn decode_loss(&mut self, out: &mut [i16]) -> Result { - self.decoder - .decode(&[], out, false) - .context("Opus PLC (decode_loss) failed") + .context("Opus decode failed") } } diff --git a/packages/client-v2/src/audio/config.rs b/packages/client-v2/src/audio/config.rs index 4d06228..4fedc0f 100644 --- a/packages/client-v2/src/audio/config.rs +++ b/packages/client-v2/src/audio/config.rs @@ -8,56 +8,49 @@ pub enum AudioScene { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AudioConfig { - // ALSA 设备参数，用于录音和播放 pub capture_device: String, pub playback_device: String, pub sample_rate: u32, pub channels: u16, - pub frame_size: usize, // 帧大小，单位为采样点 - - // Opus 编解码参数，用于音频传输 + pub frame_size: usize, pub audio_scene: AudioScene, pub bitrate: i32, - pub vbr: bool, // 是否启用 VBR（动态比特率） - pub fec: bool, // 是否启用 FEC（内联前向纠错） + pub vbr: bool, + pub fec: bool, } impl AudioConfig { - pub fn music() -> Self { + pub fn voice_16k() -> Self { Self { - audio_scene: AudioScene::Music, - sample_rate: 48_000, // 48kHz - channels: 2, - frame_size: 960, // 20ms at 48kHz - bitrate: 320_000, // 320 kbps - ..Default::default() + capture_device: "plug:Capture".to_string(), + playback_device: "plug:default".to_string(), + sample_rate: 16_000, + channels: 1, + frame_size: 320, // 20ms + audio_scene: AudioScene::Voice, + bitrate: 32_000, + vbr: true, + fec: true, } } - pub fn voice() -> Self { + pub fn music_48k() -> Self { Self { - audio_scene: AudioScene::Voice, - sample_rate: 16_000, // 16kHz - channels: 1, - frame_size: 320, // 20ms at 16kHz - bitrate: 32_000, // 32 kbps - ..Default::default() + capture_device: "plug:Capture".to_string(), + playback_device: "plug:default".to_string(), + sample_rate: 48_000, + channels: 2, + frame_size: 960, // 20ms + audio_scene: AudioScene::Music, + bitrate: 128_000, + vbr: true, + fec: true, } } } impl Default for AudioConfig { fn default() -> Self { - Self { - audio_scene: AudioScene::Voice, - capture_device: "plug:Capture".to_string(), - playback_device: "default".to_string(), - sample_rate: 16_000, - channels: 1, - frame_size: 320, // 20ms at 16kHz - bitrate: 32_000, - vbr: false, - fec: false, - } + Self::voice_16k() } } diff --git a/packages/client-v2/src/audio/player.rs b/packages/client-v2/src/audio/player.rs index fe38f5f..3aa9adb 100644 --- a/packages/client-v2/src/audio/player.rs +++ b/packages/client-v2/src/audio/player.rs @@ -11,61 +11,38 @@ pub struct AudioPlayer { impl AudioPlayer { pub fn new(config: &AudioConfig) -> Result { - let pcm = PCM::new(&config.playback_device, Direction::Playback, false) - .context("Failed to open playback PCM device")?; + let pcm = PCM::new(&config.playback_device, Direction::Playback, 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)?; - setup_pcm(&pcm, config.sample_rate, config.channels)?; + // 100ms buffer to prevent underruns + let buffer_size = (config.sample_rate as f64 * 0.1) as u32; + hwp.set_buffer_size_near(buffer_size as alsa::pcm::Frames)?; + + pcm.hw_params(&hwp)?; + } + pcm.prepare()?; Ok(Self { pcm }) } - pub fn write(&self, buffer: &[i16]) -> Result { - let res = self.pcm.io_i16()?.writei(buffer); - + pub fn write(&self, buf: &[i16]) -> Result { + let res = self.pcm.io_i16()?.writei(buf); match res { - Ok(written) => Ok(written), - Err(e) => { - // Buffer Underrun，即播放缓冲区的数据被耗尽，导致音频流中断 - if e.errno() == 32 { - // 恢复音频流状态 - self.pcm.prepare()?; - // 重新获取 IO 对象并尝试写入数据 - self.pcm - .io_i16()? - .writei(buffer) - .context("Failed to write to playback device after recovery") - } else { - Err(e).context("Failed to write to playback device") - } + 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()), } } - - pub fn prepare(&self) -> Result<()> { - self.pcm.prepare().context("Failed to prepare PCM") - } -} - -fn setup_pcm(pcm: &PCM, sample_rate: u32, channels: u16) -> Result<()> { - let hwp = HwParams::any(pcm).context("Failed to get HwParams")?; - hwp.set_access(Access::RWInterleaved)?; - hwp.set_format(Format::s16())?; - hwp.set_rate(sample_rate, alsa::ValueOr::Nearest)?; - hwp.set_channels(channels as u32)?; - - // 设置较大的缓冲区以减少由于调度抖动和设备重初始化导致的断音/卡顿 - // 使用 100ms 缓冲区，既能防止 underrun，又不会引入过大延迟 - let buffer_size = (sample_rate as f64 * 0.1) as u32; // 100ms 缓冲 - let period_size = buffer_size / 4; // 25ms 周期 - hwp.set_buffer_size_near(buffer_size as alsa::pcm::Frames)?; - hwp.set_period_size_near(period_size as alsa::pcm::Frames, alsa::ValueOr::Nearest)?; - - pcm.hw_params(&hwp).context("Failed to set HwParams")?; - - let swp = pcm.sw_params_current()?; - // 设置 start_threshold，当缓冲区有 1 个 period 数据时就开始播放 - // 这样可以快速启动，同时保持足够的缓冲余量 - swp.set_start_threshold(period_size as alsa::pcm::Frames)?; - pcm.sw_params(&swp)?; - pcm.prepare()?; - Ok(()) } diff --git a/packages/client-v2/src/audio/recorder.rs b/packages/client-v2/src/audio/recorder.rs index 952e73b..33c9069 100644 --- a/packages/client-v2/src/audio/recorder.rs +++ b/packages/client-v2/src/audio/recorder.rs @@ -11,31 +11,32 @@ pub struct AudioRecorder { impl AudioRecorder { pub fn new(config: &AudioConfig) -> Result { - let pcm = PCM::new(&config.capture_device, Direction::Capture, false) - .context("Failed to open capture PCM device")?; - - setup_pcm(&pcm, config.sample_rate, config.channels)?; + 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 }) } - pub fn read(&self, buffer: &mut [i16]) -> Result { - self.pcm - .io_i16()? - .readi(buffer) - .context("Failed to read from capture device") + pub fn read(&self, buf: &mut [i16]) -> Result { + 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()), + } } } - -fn setup_pcm(pcm: &PCM, sample_rate: u32, channels: u16) -> Result<()> { - let hwp = HwParams::any(pcm).context("Failed to get HwParams")?; - hwp.set_access(Access::RWInterleaved)?; - hwp.set_format(Format::s16())?; - hwp.set_rate(sample_rate, alsa::ValueOr::Nearest)?; - hwp.set_channels(channels as u32)?; - pcm.hw_params(&hwp).context("Failed to set HwParams")?; - - let swp = pcm.sw_params_current()?; - pcm.sw_params(&swp)?; - pcm.prepare()?; - Ok(()) -} diff --git a/packages/client-v2/src/bin/client.rs b/packages/client-v2/src/bin/client.rs index f5df377..3fd3bd4 100644 --- a/packages/client-v2/src/bin/client.rs +++ b/packages/client-v2/src/bin/client.rs @@ -1,10 +1,9 @@ -use anyhow::Result; +use xiao::app::client::Client; +use std::sync::Arc; #[tokio::main] -async fn main() -> Result<()> { - #[cfg(feature = "app")] - { - xiao::app::client::entry::run_client().await?; - } +async fn main() -> anyhow::Result<()> { + let client = Arc::new(Client::new()); + client.run().await?; Ok(()) } diff --git a/packages/client-v2/src/bin/server.rs b/packages/client-v2/src/bin/server.rs index dc6e979..f6ed176 100644 --- a/packages/client-v2/src/bin/server.rs +++ b/packages/client-v2/src/bin/server.rs @@ -1,10 +1,47 @@ -use anyhow::Result; +use std::sync::Arc; +use xiao::app::server::Server; +use xiao::audio::config::AudioConfig; #[tokio::main] -async fn main() -> Result<()> { - #[cfg(feature = "app")] - { - xiao::app::server::entry::run_server().await?; +async fn main() -> anyhow::Result<()> { + let server = Arc::new(Server::new().await?); + let s = server.clone(); + + tokio::spawn(async move { + s.run(8080).await.unwrap(); + }); + + println!("Server is running. Waiting for a client to connect..."); + + loop { + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + let clients = server.get_clients().await; + if !clients.is_empty() { + let addr = clients[0]; + println!("Client connected: {}. Starting tests...", addr); + + println!("1. Testing Shell RPC..."); + let res = server + .call(addr, "shell", vec!["echo 'Hello from Mars!'".to_string()]) + .await?; + println!( + "RPC Result: stdout={}, code={}", + res.stdout.trim(), + res.code + ); + + println!("2. Testing Audio Recording (10s)..."); + server.start_record(addr, AudioConfig::voice_16k()).await?; + tokio::time::sleep(std::time::Duration::from_secs(12)).await; + + println!("3. Testing Audio Playback (from temp/test.wav)..."); + server.start_play(addr).await?; + + break; + } } + + println!("Tests completed. Press Ctrl+C to exit."); + tokio::signal::ctrl_c().await?; Ok(()) } diff --git a/packages/client-v2/src/lib.rs b/packages/client-v2/src/lib.rs index d1c3b68..db47768 100644 --- a/packages/client-v2/src/lib.rs +++ b/packages/client-v2/src/lib.rs @@ -1,6 +1,3 @@ +pub mod app; pub mod audio; pub mod net; -pub mod utils; - -#[cfg(feature = "app")] -pub mod app; diff --git a/packages/client-v2/src/net/discovery.rs b/packages/client-v2/src/net/discovery.rs index 4d66bbc..a1b4c8b 100644 --- a/packages/client-v2/src/net/discovery.rs +++ b/packages/client-v2/src/net/discovery.rs @@ -5,18 +5,21 @@ use std::time::Duration; use tokio::net::UdpSocket; pub const DISCOVERY_PORT: u16 = 53530; +const DISCOVERY_MAGIC: &[u8] = b"XIAO_DISCOVERY_V2"; -/// 服务发现模块，用于主从节点的自动发现 pub struct Discovery; impl Discovery { - /// 服务端：启动广播，告知客户端自己的 TCP 端口 - pub async fn start_broadcast(tcp_port: u16) -> Result<()> { + pub async fn broadcast(tcp_port: u16, udp_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 { tcp_port })?; + + let mut msg = DISCOVERY_MAGIC.to_vec(); + msg.extend(postcard::to_allocvec(&ControlPacket::ServerHello { + tcp_port, + udp_port, + })?); tokio::spawn(async move { loop { @@ -24,21 +27,23 @@ impl Discovery { tokio::time::sleep(Duration::from_secs(1)).await; } }); - Ok(()) } - /// 客户端：监听广播，发现服务端的 IP 和 TCP 端口 - pub async fn discover_server() -> Result<(IpAddr, u16)> { + pub async fn listen() -> Result<(IpAddr, u16, 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 { tcp_port }) = - postcard::from_bytes::(&buf[..len]) - { - return Ok((addr.ip(), tcp_port)); + let data = &buf[..len]; + + if data.starts_with(DISCOVERY_MAGIC) { + let packet_data = &data[DISCOVERY_MAGIC.len()..]; + if let Ok(ControlPacket::ServerHello { tcp_port, udp_port }) = + postcard::from_bytes(packet_data) + { + return Ok((addr.ip(), tcp_port, udp_port)); + } } } } diff --git a/packages/client-v2/src/net/network.rs b/packages/client-v2/src/net/network.rs index 758fe1d..adfbf74 100644 --- a/packages/client-v2/src/net/network.rs +++ b/packages/client-v2/src/net/network.rs @@ -1,13 +1,61 @@ use crate::net::protocol::{AudioPacket, ControlPacket}; -use anyhow::{Context, Result}; +use anyhow::Result; use std::net::SocketAddr; use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; -use tokio::net::{TcpListener, TcpStream, UdpSocket}; +use tokio::net::{TcpStream, UdpSocket}; +use tokio::sync::Mutex; -/// UDP 音频传输 -// ... (rest of AudioSocket is same) +pub struct NetConfig { + pub tcp_port: u16, + pub udp_port: u16, +} + +/// A unified control connection over TCP +pub struct Connection { + reader: Mutex, + writer: Mutex, + peer_addr: SocketAddr, +} + +impl Connection { + pub fn new(stream: TcpStream) -> Result { + let peer_addr = stream.peer_addr()?; + let (r, w) = stream.into_split(); + Ok(Self { + reader: Mutex::new(r), + writer: Mutex::new(w), + peer_addr, + }) + } + + pub async fn send(&self, packet: &ControlPacket) -> Result<()> { + let bytes = postcard::to_allocvec(packet)?; + let mut writer = self.writer.lock().await; + writer.write_u32(bytes.len() as u32).await?; + writer.write_all(&bytes).await?; + writer.flush().await?; + Ok(()) + } + + pub async fn recv(&self) -> Result { + let mut reader = self.reader.lock().await; + let len = reader.read_u32().await? as usize; + if len > 1024 * 1024 { + return Err(anyhow::anyhow!("Packet too large: {}", len)); + } + let mut buf = vec![0u8; len]; + reader.read_exact(&mut buf).await?; + let packet = postcard::from_bytes(&buf)?; + Ok(packet) + } + + pub fn peer_addr(&self) -> SocketAddr { + self.peer_addr + } +} + +/// UDP Socket for audio transmission pub struct AudioSocket { socket: Arc, } @@ -20,140 +68,19 @@ impl AudioSocket { }) } - pub fn local_port(&self) -> Result { - Ok(self.socket.local_addr()?.port()) + pub fn port(&self) -> u16 { + self.socket.local_addr().unwrap().port() } - pub async fn send_packet(&self, packet: &AudioPacket, target: SocketAddr) -> Result<()> { + pub async fn send(&self, packet: &AudioPacket, target: SocketAddr) -> Result<()> { let bytes = postcard::to_allocvec(packet)?; self.socket.send_to(&bytes, target).await?; Ok(()) } - pub async fn recv_packet(&self, buf: &mut [u8]) -> Result<(AudioPacket, SocketAddr)> { + pub async fn recv(&self, buf: &mut [u8]) -> Result<(AudioPacket, SocketAddr)> { let (len, addr) = self.socket.recv_from(buf).await?; let packet = postcard::from_bytes(&buf[..len])?; Ok((packet, addr)) } - - pub fn clone_inner(&self) -> Arc { - self.socket.clone() - } -} - -/// TCP 控制连接读取端 -pub struct ControlReader { - reader: OwnedReadHalf, -} - -impl ControlReader { - pub async fn recv_packet(&mut self) -> Result { - let len = self.reader.read_u32().await? as usize; - if len > 10 * 1024 * 1024 { - return Err(anyhow::anyhow!("Packet too large: {}", len)); - } - let mut buf = vec![0u8; len]; - self.reader.read_exact(&mut buf).await?; - let packet = postcard::from_bytes(&buf)?; - Ok(packet) - } -} - -/// TCP 控制连接写入端 -pub struct ControlWriter { - writer: OwnedWriteHalf, -} - -impl ControlWriter { - pub async fn send_packet(&mut self, packet: &ControlPacket) -> Result<()> { - let bytes = postcard::to_allocvec(packet)?; - let len = bytes.len() as u32; - self.writer.write_u32(len).await?; - self.writer.write_all(&bytes).await?; - self.writer.flush().await?; - Ok(()) - } -} - -/// TCP 控制连接 -pub struct ControlConnection { - stream: TcpStream, -} - -impl ControlConnection { - pub fn new(stream: TcpStream) -> Self { - Self { stream } - } - - 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?; - self.stream.flush().await?; - Ok(()) - } - - 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 mut buf = vec![0u8; len]; - self.stream.read_exact(&mut buf).await?; - let packet = postcard::from_bytes(&buf)?; - Ok(packet) - } - - pub fn split(self) -> (ControlReader, ControlWriter) { - let (r, w) = self.stream.into_split(); - (ControlReader { reader: r }, ControlWriter { writer: w }) - } - - pub fn peer_addr(&self) -> Result { - self.stream.peer_addr().context("Failed to get peer addr") - } -} - -/// 服务端网络管理器 -pub struct ServerNetwork { - listener: TcpListener, -} - -impl ServerNetwork { - pub async fn setup(port: u16) -> Result { - let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await?; - Ok(Self { listener }) - } - - pub async fn accept(&self) -> Result<(ControlConnection, SocketAddr)> { - let (stream, addr) = self.listener.accept().await?; - Ok((ControlConnection::new(stream), addr)) - } - - pub fn local_addr(&self) -> Result { - self.listener - .local_addr() - .context("Failed to get local addr") - } -} - -/// 客户端网络管理器 -pub struct ClientNetwork { - control: ControlConnection, -} - -impl ClientNetwork { - pub async fn connect(server_addr: SocketAddr) -> Result { - let stream = TcpStream::connect(server_addr) - .await - .context(format!("无法连接到服务端 TCP 地址: {}", server_addr))?; - Ok(Self { - control: ControlConnection::new(stream), - }) - } - - 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 ad5bd1d..d469e93 100644 --- a/packages/client-v2/src/net/protocol.rs +++ b/packages/client-v2/src/net/protocol.rs @@ -8,19 +8,31 @@ pub struct DeviceInfo { pub version: u32, } +impl DeviceInfo { + pub fn current() -> Self { + Self { + model: "Open-XiaoAi-V2".to_string(), + mac: "00:00:00:00:00:00".to_string(), // TODO: Get actual MAC + version: 1, + } + } +} + #[derive(Serialize, Deserialize, Debug, Clone)] pub enum ControlPacket { - // 服务发现 + // Discovery ServerHello { tcp_port: u16, + udp_port: u16, }, - // 握手与认证 + // Handshake ClientIdentify { info: DeviceInfo, + udp_port: u16, }, IdentifyOk, - // 音频控制 + // Audio Control StartRecording { config: AudioConfig, }, @@ -41,12 +53,12 @@ pub enum ControlPacket { result: RpcResult, }, - // 心跳 + // Heartbeat Ping, Pong, } -#[derive(Serialize, Deserialize, Debug, Clone)] +#[derive(Serialize, Deserialize, Debug, Clone, Default)] pub struct RpcResult { pub stdout: String, pub stderr: String, @@ -55,5 +67,5 @@ pub struct RpcResult { #[derive(Serialize, Deserialize, Debug, Clone)] pub struct AudioPacket { - pub data: Vec, // Opus 编码数据 + pub data: Vec, } diff --git a/packages/client-v2/src/net/rpc.rs b/packages/client-v2/src/net/rpc.rs index 3b1c4fd..f4e56ce 100644 --- a/packages/client-v2/src/net/rpc.rs +++ b/packages/client-v2/src/net/rpc.rs @@ -17,14 +17,14 @@ impl RpcManager { } } - pub fn alloc_id(&self) -> (u32, oneshot::Receiver) { + pub fn register(&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) { + pub fn resolve(&self, id: u32, result: RpcResult) { if let Some(tx) = self.pending.lock().remove(&id) { let _ = tx.send(result); } diff --git a/packages/client-v2/src/utils/alsa.rs b/packages/client-v2/src/utils/alsa.rs deleted file mode 100644 index 42a876c..0000000 --- a/packages/client-v2/src/utils/alsa.rs +++ /dev/null @@ -1,85 +0,0 @@ -#![cfg(target_os = "linux")] - -use anyhow::{Context, Result}; -use std::fs; -use std::process::Command; - -const FIFO_PATH: &str = "/tmp/xiao_out.fifo"; -const REAL_ASOUND_CONF: &str = "/etc/asound.conf"; -const TEMP_ASOUND_CONF: &str = "/tmp/asound.xiao.conf"; - -/// ALSA 音频重定向器，用于拦截系统音频输出到 FIFO 管道 -pub struct AlsaRedirector; - -impl AlsaRedirector { - pub fn new() -> Result { - Self::cleanup(); // 确保环境干净 - - let original_conf = fs::read_to_string(REAL_ASOUND_CONF).unwrap_or_default(); - - if !original_conf.contains("pcm.original_default") { - // 重命名原有的 default 逻辑，插入拦截器 - let mut new_conf = original_conf.replace("pcm.!default", "pcm.original_default"); - new_conf.push_str(&format!( - "\npcm.!default {{ type plug slave {{ pcm \"xiao_interceptor\" format S16_LE rate 48000 channels 2 }} }}\n\ - pcm.xiao_interceptor {{ type file slave.pcm \"null\" file \"{}\" format \"raw\" }}\n", - FIFO_PATH - )); - - fs::write(TEMP_ASOUND_CONF, new_conf)?; - - // 挂载覆盖 /etc/asound.conf - let status = Command::new("mount") - .arg("--bind") - .arg(TEMP_ASOUND_CONF) - .arg(REAL_ASOUND_CONF) - .status() - .context("执行 mount 命令失败")?; - - if !status.success() { - return Err(anyhow::anyhow!("挂载 asound.conf 失败")); - } - - Self::restart_applications(); - } - - // 创建 FIFO 管道 - let _ = Command::new("mkfifo").arg(FIFO_PATH).status(); - let _ = Command::new("chmod").arg("666").arg(FIFO_PATH).status(); - - Ok(Self) - } - - pub fn cleanup() { - let _ = Command::new("sh") - .arg("-c") - .arg(format!("umount -l {} >/dev/null 2>&1", REAL_ASOUND_CONF)) - .status(); - let _ = fs::remove_file(TEMP_ASOUND_CONF); - let _ = fs::remove_file(FIFO_PATH); - Self::restart_applications(); - } - - pub fn fifo_path() -> &'static str { - FIFO_PATH - } - - pub fn restart_applications() { - // 重启媒体播放器 - let _ = Command::new("sh") - .arg("-c") - .arg("/etc/init.d/mediaplayer restart >/dev/null 2>&1") - .status(); - // 重启蓝牙 - let _ = Command::new("sh") - .arg("-c") - .arg("/etc/init.d/bluetooth restart >/dev/null 2>&1") - .status(); - } -} - -impl Drop for AlsaRedirector { - fn drop(&mut self) { - Self::cleanup(); - } -} diff --git a/packages/client-v2/src/utils/mod.rs b/packages/client-v2/src/utils/mod.rs deleted file mode 100644 index 8eaa5c4..0000000 --- a/packages/client-v2/src/utils/mod.rs +++ /dev/null @@ -1,2 +0,0 @@ -pub mod alsa; -pub mod sync; diff --git a/packages/client-v2/src/utils/sync.rs b/packages/client-v2/src/utils/sync.rs deleted file mode 100644 index 354ab8d..0000000 --- a/packages/client-v2/src/utils/sync.rs +++ /dev/null @@ -1,224 +0,0 @@ -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, - /// 当前估计的时钟偏移 (server_time - client_time) - pub current_offset: i128, - /// RTT 样本窗口 - rtts: VecDeque, - /// 当前估计的最小 RTT - min_rtt: i128, - /// 窗口大小 - window_size: usize, - /// 时钟漂移率 (ppm: parts per million) - /// 正值表示从节点时钟比主节点快 - drift_rate: f64, - /// 上次更新时间 - last_update_time: u128, - /// 漂移率估计窗口 - drift_samples: VecDeque, -} - -#[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 = 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::>()); - let offset_variance = self.calculate_variance( - &self.offsets.iter().map(|s| s.offset).collect::>(), - ); - - // 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::() as f64 / samples.len() as f64; - let variance = samples - .iter() - .map(|&x| { - let diff = x as f64 - mean; - diff * diff - }) - .sum::() - / samples.len() as f64; - variance.sqrt() - } -}