From cce3dd630a27cb93da341a01c91e2e4ba0a4ae96 Mon Sep 17 00:00:00 2001 From: Del Wang Date: Sun, 4 Jan 2026 11:24:37 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E9=87=8D=E6=9E=84=20Open-XiaoAI=20?= =?UTF-8?q?Client=20V2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/client-v2/README.md | 201 +++++- .../client-v2/src/app/client/audio_manager.rs | 151 ----- packages/client-v2/src/app/client/mod.rs | 497 ++++++++++----- packages/client-v2/src/app/client/pipeline.rs | 276 ++++++++ packages/client-v2/src/app/client/session.rs | 264 ++++++++ .../client-v2/src/app/server/audio_bus.rs | 250 ++++++++ .../client-v2/src/app/server/audio_manager.rs | 208 ------- packages/client-v2/src/app/server/mod.rs | 588 ++++++++++++------ packages/client-v2/src/app/server/session.rs | 326 ++++++++++ packages/client-v2/src/app/server/stream.rs | 302 +++++++++ packages/client-v2/src/bin/client.rs | 107 +++- packages/client-v2/src/bin/server.rs | 147 ++++- packages/client-v2/src/net/command.rs | 324 ++++++++++ packages/client-v2/src/net/event.rs | 286 +++++++++ packages/client-v2/src/net/mod.rs | 12 + packages/client-v2/src/net/protocol.rs | 94 ++- packages/client-v2/src/net/rpc.rs | 149 ++++- 17 files changed, 3408 insertions(+), 774 deletions(-) delete mode 100644 packages/client-v2/src/app/client/audio_manager.rs create mode 100644 packages/client-v2/src/app/client/pipeline.rs create mode 100644 packages/client-v2/src/app/client/session.rs create mode 100644 packages/client-v2/src/app/server/audio_bus.rs delete mode 100644 packages/client-v2/src/app/server/audio_manager.rs create mode 100644 packages/client-v2/src/app/server/session.rs create mode 100644 packages/client-v2/src/app/server/stream.rs create mode 100644 packages/client-v2/src/net/command.rs create mode 100644 packages/client-v2/src/net/event.rs diff --git a/packages/client-v2/README.md b/packages/client-v2/README.md index 3162c40..e17039f 100644 --- a/packages/client-v2/README.md +++ b/packages/client-v2/README.md @@ -1,6 +1,205 @@ # Open-XiaoAI Client V2 -> 开发中，敬请期待 +> 开发中，敬请期待... + +实时音频流推送服务，支持多客户端连接、音频录制/播放、RPC 远程调用和实时事件推送。 + +## 架构 + +``` +┌─────────────────────────────────────────────────────────────────────────┐ +│ Server │ +│ │ +│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ +│ │ SessionManager │ │ AudioBus │ │ EventBus │ │ +│ │ │ │ (Pub/Sub) │ │ (Broadcast) │ │ +│ │ ├─ Session 1 │ │ │ │ │ │ +│ │ ├─ Session 2 │◄──►│ ├─ Receiver │ │ ├─ Server → │ │ +│ │ └─ Session N │ │ └─ Broadcaster │ │ └─ Client → │ │ +│ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ +│ │ │ │ │ +│ ▼ ▼ ▼ │ +│ ┌─────────────────────────────────────────────────────────────────┐ │ +│ │ Command Handler │ │ +│ │ ├─ Shell (执行远程命令) │ │ +│ │ ├─ GetInfo (获取设备信息) │ │ +│ │ ├─ SetVolume (设置音量) │ │ +│ │ ├─ File (文件操作) │ │ +│ │ └─ System (系统控制) │ │ +│ └─────────────────────────────────────────────────────────────────┘ │ +└─────────────────────────────────────────────────────────────────────────┘ + │ TCP │ UDP + ▼ ▼ +┌─────────────────────────────────────────────────────────────────────────┐ +│ Client │ +│ │ +│ ┌─────────────────┐ ┌─────────────────────────────────────────┐ │ +│ │ Session │ │ Audio Pipelines │ │ +│ │ │ │ │ │ +│ │ ├─ Connection │ │ ┌───────────┐ ┌───────────────┐ │ │ +│ │ ├─ RPC Manager │ │ │ Record │ │ Playback │ │ │ +│ │ └─ Pipelines │◄──►│ │ Pipeline │ │ Pipeline │ │ │ +│ └─────────────────┘ │ │ │ │ │ │ │ +│ │ │ Mic→Opus │ │ Opus→Speaker │ │ │ +│ │ │ →UDP │ │ ←UDP │ │ │ +│ │ └───────────┘ └───────────────┘ │ │ +│ └─────────────────────────────────────────┘ │ +└─────────────────────────────────────────────────────────────────────────┘ +``` + +## 项目结构 + +``` +src/ +├── lib.rs +├── app/ +│ ├── server/ # 服务端 +│ │ ├── mod.rs # Server 主体 +│ │ ├── audio_bus.rs # 音频总线 (发布-订阅) +│ │ ├── session.rs # 会话管理 +│ │ └── stream.rs # 音频流 (录音/播放/转发) +│ │ +│ └── client/ # 客户端 +│ ├── mod.rs # Client 主体 +│ ├── session.rs # 会话管理 +│ └── pipeline.rs # 音频管道 +│ +├── audio/ # 音频处理 +│ ├── codec.rs # Opus 编解码 +│ ├── config.rs # 音频配置 +│ ├── player.rs # ALSA 播放 +│ ├── recorder.rs # ALSA 录音 +│ └── wav.rs # WAV 文件读写 +│ +├── net/ # 网络层 +│ ├── command.rs # RPC 命令类型 +│ ├── discovery.rs # 服务发现 +│ ├── event.rs # 实时事件系统 +│ ├── network.rs # TCP/UDP 连接 +│ ├── protocol.rs # 通信协议 +│ └── rpc.rs # RPC 管理 +│ +└── bin/ + ├── client.rs # 客户端 demo + └── server.rs # 服务端 demo +``` + +## 功能特性 + +### RPC 命令系统 + +支持多种类型的远程调用： + +| 命令 | 描述 | 请求 | 响应 | +| ----------- | --------------- | --------------------------------- | ------------------------------------------- | +| `Shell` | 执行 Shell 命令 | command, cwd, env, timeout | stdout, stderr, exit_code | +| `GetInfo` | 获取设备信息 | - | model, serial, version, uptime, audio_state | +| `SetVolume` | 设置音量 | volume (0-100) | previous, current | +| `File` | 文件操作 | Read/Write/Delete/List/Stat | data/entries/stat | +| `System` | 系统控制 | Reboot/Shutdown/GetLoad/GetMemory | Accepted/Load/Memory | +| `Ping` | 延迟测量 | timestamp | timestamp, server_time | + +### 实时事件系统 + +**服务端事件 (Server → Client):** + +- `AudioStatusChanged` - 音频状态变化 +- `ClientJoined/Left` - 客户端加入/离开 +- `Notification` - 通知消息 +- `RecordingComplete` - 录音完成 +- `PlaybackComplete` - 播放完成 + +**客户端事件 (Client → Server):** + +- `StatusUpdate` - 状态更新 (CPU/内存/温度) +- `AudioLevel` - 音频电平 +- `KeyPress` - 按键事件 +- `Alert` - 警告/错误 + +### 音频功能 + +- **录音**: 从客户端麦克风录制，服务端保存为 WAV +- **播放**: 服务端推送音频文件到客户端播放 +- **编码**: Opus 编解码，支持 16kHz/48kHz +- **传输**: UDP 低延迟传输 + +## 使用方法 + +### 启动服务端 + +```bash +cargo run --bin server --release +``` + +### 启动客户端 (Linux) + +```bash +cargo run --bin client --release +``` + +### 交叉编译 (ARM) + +```bash +make build-arm +``` + +## 配置 + +通过环境变量配置认证： + +```bash +export XIAO_SERVER_AUTH="your-server-secret" +export XIAO_CLIENT_AUTH="your-client-secret" +``` + +## 示例代码 + +### 服务端 + +```rust +use xiao::app::server::Server; +use xiao::net::command::Command; + +let server = Arc::new(Server::new().await?); + +// 启动服务器 +tokio::spawn(async move { + server.run(8080).await.unwrap(); +}); + +// 执行远程命令 +let result = server.shell(client_addr, "uname -a").await?; +println!("Output: {}", result.stdout); + +// 开始录音 +server.start_record(client_addr, AudioConfig::voice_16k()).await?; + +// 广播事件 +server.broadcast_event(ServerEvent::Notification { + level: NotificationLevel::Info, + title: "Notice".to_string(), + message: "Hello everyone!".to_string(), +}).await; +``` + +### 客户端 + +```rust +use xiao::app::client::{Client, ClientConfig}; + +let client = Arc::new(Client::new(ClientConfig::default())); + +// 订阅服务端事件 +let mut events = client.subscribe_events(); +tokio::spawn(async move { + while let Ok(event) = events.recv().await { + println!("Event: {:?}", event); + } +}); + +// 运行客户端 +client.run().await?; +``` ## License diff --git a/packages/client-v2/src/app/client/audio_manager.rs b/packages/client-v2/src/app/client/audio_manager.rs deleted file mode 100644 index a2fb3cb..0000000 --- a/packages/client-v2/src/app/client/audio_manager.rs +++ /dev/null @@ -1,151 +0,0 @@ -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; -use std::net::SocketAddr; -use std::sync::Arc; -use tokio::sync::{RwLock, mpsc}; -use tokio_util::sync::CancellationToken; - -pub struct ClientAudioManager { - audio_socket: Arc, - audio_addr: SocketAddr, - session_cancel: CancellationToken, - record_cancel: RwLock>, - play_cancel: RwLock>, -} - -impl ClientAudioManager { - pub fn new( - audio_socket: Arc, - audio_addr: SocketAddr, - session_cancel: CancellationToken, - ) -> Self { - Self { - audio_socket, - audio_addr, - session_cancel, - record_cancel: RwLock::new(None), - play_cancel: RwLock::new(None), - } - } - - pub async fn stop_recorder(&self) { - let mut cancel_guard = self.record_cancel.write().await; - if let Some(token) = cancel_guard.take() { - token.cancel(); - } - } - - pub async fn stop_player(&self) { - let mut cancel_guard = self.play_cancel.write().await; - if let Some(token) = cancel_guard.take() { - token.cancel(); - } - } - - pub async fn start_recording(&self, config: AudioConfig) { - self.stop_recorder().await; - let token = self.session_cancel.child_token(); - *self.record_cancel.write().await = Some(token.clone()); - - let audio_socket = self.audio_socket.clone(); - let audio_addr = self.audio_addr; - - tokio::spawn(async move { - let (pcm_tx, mut pcm_rx) = mpsc::channel::>(32); - let conf = config.clone(); - - // 录音线程 (ALSA 阻塞) - std::thread::spawn(move || { - let recorder = match AudioRecorder::new(&conf) { - Ok(r) => r, - Err(e) => { - eprintln!("Failed to start recorder: {}", e); - return; - } - }; - let mut buf = vec![0i16; conf.frame_size]; - while let Ok(n) = recorder.read(&mut buf) { - if pcm_tx.blocking_send(buf[..n].to_vec()).is_err() { - 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! { - _ = token.cancelled() => break, - Some(pcm) = pcm_rx.recv() => { - let mut out = vec![0u8; 4096]; - if let Ok(len) = codec.encode(&pcm, &mut out) { - let _ = audio_socket.send(&AudioPacket { data: out[..len].to_vec() }, audio_addr).await; - } - } - } - } - println!("Recording stopped."); - }); - } - - pub async fn start_playback(&self, config: AudioConfig) { - self.stop_player().await; - let token = self.session_cancel.child_token(); - *self.play_cancel.write().await = Some(token.clone()); - - let audio_socket = self.audio_socket.clone(); - - tokio::spawn(async move { - let (pcm_tx, mut pcm_rx) = mpsc::channel::>(32); - let conf = config.clone(); - - // 播放线程 (ALSA 阻塞) - std::thread::spawn(move || { - let player = match AudioPlayer::new(&conf) { - Ok(p) => p, - Err(e) => { - eprintln!("Failed to start player: {}", e); - return; - } - }; - while let Some(pcm) = pcm_rx.blocking_recv() { - let _ = player.write(&pcm); - } - }); - - let mut codec = match OpusCodec::new(&config) { - Ok(c) => c, - Err(e) => { - eprintln!("Failed to init opus codec: {}", e); - return; - } - }; - let mut udp_buf = vec![0u8; 4096]; - println!("Playback started..."); - loop { - tokio::select! { - _ = token.cancelled() => break, - res = audio_socket.recv(&mut udp_buf) => { - if let Ok((packet, _)) = res { - 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; - } - } - } - } - } - println!("Playback stopped."); - }); - } -} diff --git a/packages/client-v2/src/app/client/mod.rs b/packages/client-v2/src/app/client/mod.rs index 873229f..bf7aaad 100644 --- a/packages/client-v2/src/app/client/mod.rs +++ b/packages/client-v2/src/app/client/mod.rs @@ -1,166 +1,227 @@ +//! # Client 模块 +//! +//! 音频客户端，支持： +//! - 服务发现和自动连接 +//! - 音频录制和播放 +//! - RPC 远程调用 +//! - 实时事件处理 +//! +//! ## 架构 +//! +//! ```text +//! ┌─────────────────────────────────────┐ +//! │ Client │ +//! │ │ +//! Server ◀──TCP─────┼──▶ Session │ +//! │ ├─ Connection │ +//! │ ├─ RPC Manager │ +//! │ └─ Active Pipelines │ +//! │ │ +//! Server ◀──UDP─────┼──▶ Audio Pipelines │ +//! │ ├─ RecordPipeline │ +//! │ └─ PlaybackPipeline │ +//! │ │ +//! Events ◀──────────┼──▶ Event Handlers │ +//! └─────────────────────────────────────┘ +//! ``` + #![cfg(target_os = "linux")] -mod audio_manager; +mod pipeline; +mod session; +pub use pipeline::{PipelineHandle, PlaybackPipeline, RecordPipeline}; +pub use session::Session; + +use crate::net::command::{ + AudioState, Command, CommandError, CommandResult, DeviceInfo, SetVolumeResponse, ShellRequest, + ShellResponse, +}; use crate::net::discovery::Discovery; +use crate::net::event::{ClientEvent, NotificationLevel, ServerEvent}; use crate::net::network::{AudioSocket, Connection}; -use crate::net::protocol::{ClientInfo, ControlPacket, RpcResult}; -use crate::net::rpc::RpcManager; +use crate::net::protocol::{ClientInfo, ControlPacket}; use anyhow::{Result, anyhow}; -use audio_manager::ClientAudioManager; +use session::handshake; use std::net::SocketAddr; use std::sync::Arc; -use tokio::sync::{RwLock}; +use tokio::sync::{RwLock, broadcast}; use tokio_util::sync::CancellationToken; -struct ActiveSession { - conn: Arc, - audio_manager: ClientAudioManager, - session_cancel: CancellationToken, +/// 客户端配置 +pub struct ClientConfig { + /// 客户端型号 + pub model: String, + /// 序列号 + pub serial_number: String, + /// 心跳间隔（秒） + pub heartbeat_interval: u64, + /// 连接超时（秒） + pub timeout: u64, } +impl Default for ClientConfig { + fn default() -> Self { + Self { + model: "Open-XiaoAi-V2".to_string(), + serial_number: "00:00:00:00:00:00".to_string(), + heartbeat_interval: 10, + timeout: 60, + } + } +} + +/// 音频客户端 pub struct Client { - session: RwLock>>, - rpc: Arc, + /// 配置 + config: ClientConfig, + /// 当前活动会话 + session: RwLock>>, + /// 全局取消令牌 + cancel: CancellationToken, + /// 服务端事件广播 + server_events: broadcast::Sender, + /// 客户端启动时间 + started_at: std::time::Instant, } impl Client { - pub fn new() -> Self { + /// 创建新客户端 + pub fn new(config: ClientConfig) -> Self { + let (server_events, _) = broadcast::channel(64); Self { + config, session: RwLock::new(None), - rpc: Arc::new(RpcManager::new()), + cancel: CancellationToken::new(), + server_events, + started_at: std::time::Instant::now(), } } + /// 使用默认配置创建客户端 + pub fn with_defaults() -> Self { + Self::new(ClientConfig::default()) + } + + /// 订阅服务端事件 + pub fn subscribe_events(&self) -> broadcast::Receiver { + self.server_events.subscribe() + } + + /// 运行客户端（自动发现服务器并连接） pub async fn run(self: Arc) -> Result<()> { loop { - println!("Searching for server..."); - let (ip, tcp_port) = match Discovery::listen().await { - Ok(res) => res, - Err(e) => { - eprintln!("Discovery listen error: {}", e); - tokio::time::sleep(std::time::Duration::from_secs(1)).await; - continue; + tokio::select! { + _ = self.cancel.cancelled() => { + println!("[Client] Shutting down..."); + break; } - }; - - let addr = SocketAddr::new(ip, tcp_port); - println!("Found server at {}", addr); - - match tokio::net::TcpStream::connect(addr).await { - Err(e) => eprintln!("Failed to connect to {}: {}", addr, e), - Ok(stream) => { - println!("Connected to TCP server at {}", addr); - if let Err(e) = self.clone().handle_session(stream, addr).await { - eprintln!("Session error: {:?}", e); + result = self.discover_and_connect() => { + if let Err(e) = result { + eprintln!("[Client] Connection error: {}", e); } + // 连接断开后清理并重试 + self.cleanup().await; + tokio::time::sleep(std::time::Duration::from_secs(1)).await; } } - - self.cleanup().await; - tokio::time::sleep(std::time::Duration::from_secs(1)).await; } + Ok(()) } - async fn cleanup(&self) { - let mut session_guard = self.session.write().await; - if let Some(s) = session_guard.take() { - s.session_cancel.cancel(); - } + /// 发现服务器并连接 + async fn discover_and_connect(&self) -> Result<()> { + println!("[Client] Searching for server..."); + + let (ip, tcp_port) = Discovery::listen().await?; + let server_addr = SocketAddr::new(ip, tcp_port); + println!("[Client] Found server at {}", server_addr); + + let stream = tokio::net::TcpStream::connect(server_addr).await?; + println!("[Client] Connected to {}", server_addr); + + self.handle_session(stream, server_addr).await } + /// 处理会话 async fn handle_session( - self: Arc, + &self, stream: tokio::net::TcpStream, - addr: SocketAddr, + server_addr: SocketAddr, ) -> Result<()> { let conn = Arc::new(Connection::new(stream)?); let audio_socket = Arc::new(AudioSocket::bind().await?); - // --- 握手 (Handshake) --- - let version = env!("CARGO_PKG_VERSION").to_string(); - let server_auth = - std::env::var("XIAO_SERVER_AUTH").unwrap_or_else(|_| "xiao-server".to_string()); - let client_auth = - std::env::var("XIAO_CLIENT_AUTH").unwrap_or_else(|_| "xiao-client".to_string()); - - conn.send(&ControlPacket::ClientHello { - auth: server_auth, - version: version.clone(), - udp_port: audio_socket.port(), - // todo client info - info: ClientInfo { - model: "Open-XiaoAi-V2".to_string(), - serial_number: "00:00:00:00:00:00".to_string(), - }, - }) - .await?; - - let server_udp_port = match conn.recv().await? { - ControlPacket::ServerHello { - version: v, - udp_port, - auth, - } => { - if v != version { - return Err(anyhow!("Server version mismatch: {} != {}", v, version)); - } - if auth != client_auth { - return Err(anyhow!("Invalid server auth")); - } - udp_port - } - _ => return Err(anyhow!("Handshake failed")), + // 执行握手 + let client_info = ClientInfo { + model: self.config.model.clone(), + serial_number: self.config.serial_number.clone(), }; - let audio_addr = SocketAddr::new(addr.ip(), server_udp_port); + let handshake_result = + handshake(&conn, audio_socket.port(), server_addr, client_info).await?; + println!( - "Handshake successful with {}, audio at {}", - addr, audio_addr + "[Client] Handshake OK, server audio at {}", + handshake_result.server_audio_addr ); - // --- 初始化 Session --- - let session_cancel = CancellationToken::new(); - let session = Arc::new(ActiveSession { - conn: conn.clone(), - audio_manager: ClientAudioManager::new( - audio_socket, - audio_addr, - session_cancel.clone(), - ), - session_cancel, - }); + // 创建会话 + let session_cancel = self.cancel.child_token(); + let session = Arc::new(Session::new( + conn.clone(), + audio_socket, + handshake_result.server_audio_addr, + session_cancel.clone(), + )); + + // 存储会话 *self.session.write().await = Some(session.clone()); - // 心跳 - let hb_session = session.clone(); + // 启动心跳 + self.spawn_heartbeat(session.clone()); + + // 运行消息循环 + self.message_loop(session).await + } + + /// 启动心跳任务 + fn spawn_heartbeat(&self, session: Arc) { + let interval = std::time::Duration::from_secs(self.config.heartbeat_interval); + tokio::spawn(async move { - let mut interval = tokio::time::interval(std::time::Duration::from_secs(10)); + let mut ticker = tokio::time::interval(interval); loop { tokio::select! { - _ = hb_session.session_cancel.cancelled() => break, - _ = interval.tick() => { - if hb_session.conn.send(&ControlPacket::Ping).await.is_err() { break; } + _ = session.cancel.cancelled() => break, + _ = ticker.tick() => { + if session.send(&ControlPacket::Ping).await.is_err() { + break; + } } } } }); + } + + /// 消息主循环 + async fn message_loop(&self, session: Arc) -> Result<()> { + let timeout = std::time::Duration::from_secs(self.config.timeout); - // 消息主循环 loop { tokio::select! { - _ = session.session_cancel.cancelled() => break, - res = tokio::time::timeout(std::time::Duration::from_secs(60), conn.recv()) => { - match res { + _ = session.cancel.cancelled() => break, + result = tokio::time::timeout(timeout, session.recv()) => { + match result { Ok(Ok(packet)) => { - if let Err(e) = self.process_packet(packet, &session).await { - eprintln!("Process packet error: {}", e); - } + self.handle_packet(&session, packet).await?; } Ok(Err(e)) => { - return Err(anyhow!("Connection receive error: {}", e)); + return Err(anyhow!("Connection error: {}", e)); + } + Err(_) => { + return Err(anyhow!("Connection timeout")); } - Err(_) => return Err(anyhow!("Connection timeout")), } } } @@ -169,87 +230,213 @@ impl Client { Ok(()) } - // todo 考虑有些操作比较耗时，需要非阻塞处理 - async fn process_packet( - &self, - packet: ControlPacket, - session: &Arc, - ) -> Result<()> { + /// 处理控制包 + async fn handle_packet(&self, session: &Arc, packet: ControlPacket) -> Result<()> { match packet { ControlPacket::Ping => { - session.conn.send(&ControlPacket::Pong).await?; + session.send(&ControlPacket::Pong).await?; } ControlPacket::Pong => {} ControlPacket::RpcResponse { id, result } => { - self.rpc.resolve(id, result); + session.resolve_rpc(id, result); } - ControlPacket::RpcRequest { id, method, args } => { - let result = self.handle_rpc(&method, args).await; + ControlPacket::RpcRequest { id, command } => { + let result = self.handle_command(session, command).await; session - .conn .send(&ControlPacket::RpcResponse { id, result }) .await?; } - ControlPacket::StartRecording { config } => { - session.audio_manager.start_recording(config).await; + ControlPacket::ServerEvent(event) => { + self.handle_server_event(session, event).await; } - ControlPacket::StartPlayback { config } => { - session.audio_manager.start_playback(config).await; + ControlPacket::StartRecording { config } => { + println!("[Client] Starting recording..."); + let handle = RecordPipeline::spawn( + config, + session.audio_socket.clone(), + session.server_audio_addr, + session.cancel.clone(), + ); + session.start_recording(handle); } ControlPacket::StopRecording => { - session.audio_manager.stop_recorder().await; + println!("[Client] Stopping recording..."); + session.stop_recording(); + } + ControlPacket::StartPlayback { config } => { + println!("[Client] Starting playback..."); + let handle = PlaybackPipeline::spawn( + config, + session.audio_socket.clone(), + session.cancel.clone(), + ); + session.start_playback(handle); } ControlPacket::StopPlayback => { - session.audio_manager.stop_player().await; + println!("[Client] Stopping playback..."); + session.stop_playback(); } _ => {} } 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() - }, + /// 处理 RPC 命令 + async fn handle_command(&self, session: &Arc, command: Command) -> CommandResult { + match command { + Command::Shell(req) => self.handle_shell(req), + Command::GetInfo => self.handle_get_info(session), + Command::Ping { timestamp } => { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + CommandResult::Pong { + timestamp, + server_time: now, } } - _ => RpcResult { - stderr: "Unsupported method".to_string(), - code: -1, - ..Default::default() - }, + Command::SetVolume(req) => { + let prev = session.set_volume(req.volume); + CommandResult::Volume(SetVolumeResponse { + previous: prev, + current: session.volume(), + }) + } + _ => CommandResult::Error(CommandError::not_implemented()), } } - pub async fn call(&self, method: &str, args: Vec) -> Result { - let (id, rx) = self.rpc.register(); + /// 处理 Shell 命令 + fn handle_shell(&self, req: ShellRequest) -> CommandResult { + let mut cmd = std::process::Command::new("sh"); + cmd.arg("-c").arg(&req.command); + + if let Some(cwd) = &req.cwd { + cmd.current_dir(cwd); + } + + if let Some(env) = &req.env { + for (k, v) in env { + cmd.env(k, v); + } + } + + match cmd.output() { + Ok(output) => CommandResult::Shell(ShellResponse { + stdout: String::from_utf8_lossy(&output.stdout).to_string(), + stderr: String::from_utf8_lossy(&output.stderr).to_string(), + exit_code: output.status.code().unwrap_or(-1), + }), + Err(e) => CommandResult::Error(CommandError::internal(e.to_string())), + } + } + + /// 处理获取设备信息 + fn handle_get_info(&self, session: &Arc) -> CommandResult { + CommandResult::Info(DeviceInfo { + model: self.config.model.clone(), + serial_number: self.config.serial_number.clone(), + version: env!("CARGO_PKG_VERSION").to_string(), + uptime_secs: session.uptime_secs(), + audio_state: AudioState { + is_recording: session.is_recording(), + is_playing: session.is_playing(), + volume: session.volume(), + }, + }) + } + + /// 处理服务端事件 + async fn handle_server_event(&self, _session: &Arc, event: ServerEvent) { + // 广播给订阅者 + let _ = self.server_events.send(event.clone()); + + // 本地处理 + match &event { + ServerEvent::Notification { + level, + title, + message, + } => { + println!("[Event] [{:?}] {}: {}", level, title, message); + } + ServerEvent::AudioStatusChanged { + is_recording, + is_playing, + } => { + println!( + "[Event] Audio status: recording={}, playing={}", + is_recording, is_playing + ); + } + ServerEvent::ClientJoined { addr, model } => { + println!("[Event] Client joined: {} ({})", model, addr); + } + ServerEvent::ClientLeft { addr, model } => { + println!("[Event] Client left: {} ({})", model, addr); + } + _ => {} + } + } + + /// 清理资源 + async fn cleanup(&self) { + let mut session_guard = self.session.write().await; + if let Some(session) = session_guard.take() { + session.cleanup(); + println!("[Client] Session cleaned up"); + } + } + + // ==================== 公开 API ==================== + + /// 执行 RPC 命令 + pub async fn execute(&self, command: Command) -> Result { let session_guard = self.session.read().await; if let Some(session) = session_guard.as_ref() { - session - .conn - .send(&ControlPacket::RpcRequest { - id, - method: method.to_string(), - args, - }) - .await?; - Ok(rx.await?) + session.execute(command).await } else { Err(anyhow!("Not connected")) } } + + /// 执行 Shell 命令 + pub async fn shell(&self, cmd: &str) -> Result { + let result = self.execute(Command::shell(cmd)).await?; + match result { + CommandResult::Shell(resp) => Ok(resp), + CommandResult::Error(e) => Err(anyhow!("{}", e)), + _ => Err(anyhow!("Unexpected response type")), + } + } + + /// 发送客户端事件 + pub async fn send_event(&self, event: ClientEvent) -> Result<()> { + let session_guard = self.session.read().await; + if let Some(session) = session_guard.as_ref() { + session.send(&ControlPacket::ClientEvent(event)).await + } else { + Err(anyhow!("Not connected")) + } + } + + /// 发送警告事件 + pub async fn send_alert(&self, level: NotificationLevel, message: &str) -> Result<()> { + self.send_event(ClientEvent::Alert { + level, + message: message.to_string(), + }) + .await + } + + /// 检查是否已连接 + pub async fn is_connected(&self) -> bool { + self.session.read().await.is_some() + } + + /// 关闭客户端 + pub fn shutdown(&self) { + self.cancel.cancel(); + } } diff --git a/packages/client-v2/src/app/client/pipeline.rs b/packages/client-v2/src/app/client/pipeline.rs new file mode 100644 index 0000000..03eb6fc --- /dev/null +++ b/packages/client-v2/src/app/client/pipeline.rs @@ -0,0 +1,276 @@ +//! # Audio Pipeline - 音频管道抽象 +//! +//! 提供录音和播放的管道化处理。 +//! +//! ## 设计 +//! +//! ```text +//! RecordPipeline: +//! ┌──────────┐ ┌────────────┐ ┌────────────┐ +//! │ ALSA Mic │────▶│ Opus Encode│────▶│ UDP Send │ +//! │ (thread) │ │ (async) │ │ (async) │ +//! └──────────┘ └────────────┘ └────────────┘ +//! +//! PlaybackPipeline: +//! ┌────────────┐ ┌────────────┐ ┌────────────┐ +//! │ UDP Recv │────▶│ Opus Decode│────▶│ ALSA Play │ +//! │ (async) │ │ (async) │ │ (thread) │ +//! └────────────┘ └────────────┘ └────────────┘ +//! ``` + +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; +use std::net::SocketAddr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; + +/// 管道句柄 - 用于控制正在运行的音频管道 +pub struct PipelineHandle { + cancel: CancellationToken, + /// 用于通知阻塞线程停止 + stop_flag: Arc, +} + +impl PipelineHandle { + fn new(cancel: CancellationToken, stop_flag: Arc) -> Self { + Self { cancel, stop_flag } + } + + /// 停止管道 + pub fn stop(&self) { + self.stop_flag.store(true, Ordering::SeqCst); + self.cancel.cancel(); + } + + /// 检查是否已停止 + pub fn is_stopped(&self) -> bool { + self.cancel.is_cancelled() + } +} + +impl Drop for PipelineHandle { + fn drop(&mut self) { + self.stop(); + } +} + +/// 录音管道 +/// 从麦克风捕获 -> Opus 编码 -> UDP 发送到服务器 +pub struct RecordPipeline; + +impl RecordPipeline { + /// 启动录音管道 + /// + /// # Arguments + /// * `config` - 音频配置 + /// * `socket` - UDP socket + /// * `target` - 目标服务器地址 + /// * `parent_cancel` - 父级取消令牌 + pub fn spawn( + config: AudioConfig, + socket: Arc, + target: SocketAddr, + parent_cancel: CancellationToken, + ) -> PipelineHandle { + let cancel = parent_cancel.child_token(); + let stop_flag = Arc::new(AtomicBool::new(false)); + let handle = PipelineHandle::new(cancel.clone(), stop_flag.clone()); + + let token = cancel.clone(); + + tokio::spawn(async move { + if let Err(e) = Self::run(config, socket, target, token, stop_flag).await { + eprintln!("[RecordPipeline] Error: {}", e); + } + }); + + handle + } + + async fn run( + config: AudioConfig, + socket: Arc, + target: SocketAddr, + cancel: CancellationToken, + stop_flag: Arc, + ) -> anyhow::Result<()> { + // 创建 PCM 数据通道 + let (pcm_tx, mut pcm_rx) = mpsc::channel::>(32); + + // 启动 ALSA 录音线程（阻塞 I/O） + let recorder_config = config.clone(); + let recorder_stop = stop_flag.clone(); + + std::thread::spawn(move || { + let recorder = match AudioRecorder::new(&recorder_config) { + Ok(r) => r, + Err(e) => { + eprintln!("[RecordPipeline] Failed to create recorder: {}", e); + return; + } + }; + + let mut buf = vec![0i16; recorder_config.frame_size]; + + // 使用 stop_flag 来优雅退出 + while !recorder_stop.load(Ordering::SeqCst) { + match recorder.read(&mut buf) { + Ok(n) if n > 0 => { + if pcm_tx.blocking_send(buf[..n].to_vec()).is_err() { + break; + } + } + Ok(_) => { + // 空读取，继续 + } + Err(e) => { + eprintln!("[RecordPipeline] Read error: {}", e); + break; + } + } + } + }); + + // 创建 Opus 编码器 + let mut codec = OpusCodec::new(&config)?; + let mut opus_buf = vec![0u8; 4096]; + + println!("[RecordPipeline] Started -> {}", target); + + // 主循环：从录音线程接收 PCM，编码后发送 + loop { + tokio::select! { + _ = cancel.cancelled() => break, + pcm = pcm_rx.recv() => { + match pcm { + Some(samples) => { + if let Ok(len) = codec.encode(&samples, &mut opus_buf) { + let packet = AudioPacket { + data: opus_buf[..len].to_vec(), + }; + let _ = socket.send(&packet, target).await; + } + } + None => { + // 录音线程已退出 + break; + } + } + } + } + } + + println!("[RecordPipeline] Stopped"); + Ok(()) + } +} + +/// 播放管道 +/// 从 UDP 接收 -> Opus 解码 -> 扬声器播放 +pub struct PlaybackPipeline; + +impl PlaybackPipeline { + /// 启动播放管道 + /// + /// # Arguments + /// * `config` - 音频配置 + /// * `socket` - UDP socket + /// * `parent_cancel` - 父级取消令牌 + pub fn spawn( + config: AudioConfig, + socket: Arc, + parent_cancel: CancellationToken, + ) -> PipelineHandle { + let cancel = parent_cancel.child_token(); + let stop_flag = Arc::new(AtomicBool::new(false)); + let handle = PipelineHandle::new(cancel.clone(), stop_flag.clone()); + + let token = cancel.clone(); + + tokio::spawn(async move { + if let Err(e) = Self::run(config, socket, token, stop_flag).await { + eprintln!("[PlaybackPipeline] Error: {}", e); + } + }); + + handle + } + + async fn run( + config: AudioConfig, + socket: Arc, + cancel: CancellationToken, + stop_flag: Arc, + ) -> anyhow::Result<()> { + // 创建 PCM 数据通道 + let (pcm_tx, pcm_rx) = mpsc::channel::>(32); + + // 启动 ALSA 播放线程（阻塞 I/O） + let player_config = config.clone(); + let player_stop = stop_flag.clone(); + + std::thread::spawn(move || { + let player = match AudioPlayer::new(&player_config) { + Ok(p) => p, + Err(e) => { + eprintln!("[PlaybackPipeline] Failed to create player: {}", e); + return; + } + }; + + // 使用 blocking_recv 在线程中接收 + let mut rx = pcm_rx; + while !player_stop.load(Ordering::SeqCst) { + match rx.blocking_recv() { + Some(samples) => { + if let Err(e) = player.write(&samples) { + eprintln!("[PlaybackPipeline] Write error: {}", e); + break; + } + } + None => { + // 通道关闭 + break; + } + } + } + }); + + // 创建 Opus 解码器 + let mut codec = OpusCodec::new(&config)?; + let mut pcm_buf = vec![0i16; config.frame_size]; + let mut udp_buf = vec![0u8; 4096]; + + println!("[PlaybackPipeline] Started"); + + // 主循环：从 UDP 接收，解码后发送给播放线程 + loop { + tokio::select! { + _ = cancel.cancelled() => break, + result = socket.recv(&mut udp_buf) => { + match result { + Ok((packet, _src)) => { + if let Ok(n) = codec.decode(&packet.data, &mut pcm_buf) { + // 使用 try_send 避免阻塞 + let _ = pcm_tx.try_send(pcm_buf[..n].to_vec()); + } + } + Err(e) => { + eprintln!("[PlaybackPipeline] Recv error: {}", e); + } + } + } + } + } + + println!("[PlaybackPipeline] Stopped"); + Ok(()) + } +} + diff --git a/packages/client-v2/src/app/client/session.rs b/packages/client-v2/src/app/client/session.rs new file mode 100644 index 0000000..fe05247 --- /dev/null +++ b/packages/client-v2/src/app/client/session.rs @@ -0,0 +1,264 @@ +//! # Client Session - 客户端会话管理 +//! +//! 轻量级的会话结构，负责： +//! - TCP 控制连接 +//! - RPC 管理 +//! - 会话生命周期 +//! - 活动音频流追踪 + +use crate::net::command::{Command, CommandResult}; +use crate::net::network::{AudioSocket, Connection}; +use crate::net::protocol::{ClientInfo, ControlPacket}; +use crate::net::rpc::RpcManager; +use anyhow::{Context, Result, anyhow}; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio_util::sync::CancellationToken; + +use super::pipeline::PipelineHandle; + +/// 活动管道追踪 +pub struct ActivePipelines { + pub recorder: Option, + pub player: Option, +} + +impl Default for ActivePipelines { + fn default() -> Self { + Self { + recorder: None, + player: None, + } + } +} + +impl ActivePipelines { + pub fn stop_all(&mut self) { + if let Some(h) = self.recorder.take() { + h.stop(); + } + if let Some(h) = self.player.take() { + h.stop(); + } + } + + pub fn start_recording(&mut self, handle: PipelineHandle) { + if let Some(h) = self.recorder.take() { + h.stop(); + } + self.recorder = Some(handle); + } + + pub fn stop_recording(&mut self) { + if let Some(h) = self.recorder.take() { + h.stop(); + } + } + + pub fn start_playback(&mut self, handle: PipelineHandle) { + if let Some(h) = self.player.take() { + h.stop(); + } + self.player = Some(handle); + } + + pub fn stop_playback(&mut self) { + if let Some(h) = self.player.take() { + h.stop(); + } + } + + pub fn is_recording(&self) -> bool { + self.recorder.is_some() + } + + pub fn is_playing(&self) -> bool { + self.player.is_some() + } +} + +/// 客户端会话 +pub struct Session { + /// TCP 控制连接 + pub conn: Arc, + + /// UDP 音频 socket + pub audio_socket: Arc, + + /// 服务器音频地址 + pub server_audio_addr: SocketAddr, + + /// RPC 管理器 + pub rpc: Arc, + + /// 会话取消令牌 + pub cancel: CancellationToken, + + /// 活动管道 + pipelines: parking_lot::Mutex, + + /// 会话创建时间 + created_at: std::time::Instant, + + /// 当前音量 + volume: parking_lot::Mutex, +} + +impl Session { + /// 创建新会话 + pub fn new( + conn: Arc, + audio_socket: Arc, + server_audio_addr: SocketAddr, + cancel: CancellationToken, + ) -> Self { + Self { + conn, + audio_socket, + server_audio_addr, + rpc: Arc::new(RpcManager::new()), + cancel, + pipelines: parking_lot::Mutex::new(ActivePipelines::default()), + created_at: std::time::Instant::now(), + volume: parking_lot::Mutex::new(100), + } + } + + /// 检查会话是否仍然有效 + pub fn is_alive(&self) -> bool { + !self.cancel.is_cancelled() + } + + /// 获取会话运行时间 + pub fn uptime_secs(&self) -> u64 { + self.created_at.elapsed().as_secs() + } + + /// 发送控制包 + pub async fn send(&self, packet: &ControlPacket) -> Result<()> { + self.conn.send(packet).await + } + + /// 接收控制包 + pub async fn recv(&self) -> Result { + self.conn.recv().await + } + + /// 发起 RPC 调用（新版） + pub async fn execute(&self, command: Command) -> Result { + let (id, rx) = self.rpc.register(); + self.conn + .send(&ControlPacket::RpcRequest { id, command }) + .await?; + rx.await.context("RPC channel closed") + } + + /// 处理 RPC 响应 + pub fn resolve_rpc(&self, id: u32, result: CommandResult) { + self.rpc.resolve(id, result); + } + + /// 开始录音管道 + pub fn start_recording(&self, handle: PipelineHandle) { + self.pipelines.lock().start_recording(handle); + } + + /// 停止录音管道 + pub fn stop_recording(&self) { + self.pipelines.lock().stop_recording(); + } + + /// 开始播放管道 + pub fn start_playback(&self, handle: PipelineHandle) { + self.pipelines.lock().start_playback(handle); + } + + /// 停止播放管道 + pub fn stop_playback(&self) { + self.pipelines.lock().stop_playback(); + } + + /// 检查是否正在录音 + pub fn is_recording(&self) -> bool { + self.pipelines.lock().is_recording() + } + + /// 检查是否正在播放 + pub fn is_playing(&self) -> bool { + self.pipelines.lock().is_playing() + } + + /// 获取当前音量 + pub fn volume(&self) -> u8 { + *self.volume.lock() + } + + /// 设置音量 + pub fn set_volume(&self, vol: u8) -> u8 { + let mut v = self.volume.lock(); + let prev = *v; + *v = vol.min(100); + prev + } + + /// 清理所有资源 + pub fn cleanup(&self) { + self.cancel.cancel(); + self.pipelines.lock().stop_all(); + } +} + +impl Drop for Session { + fn drop(&mut self) { + self.cleanup(); + } +} + +/// 握手结果 +pub struct HandshakeResult { + pub server_audio_addr: SocketAddr, +} + +/// 执行客户端握手 +pub async fn handshake( + conn: &Connection, + audio_port: u16, + server_addr: SocketAddr, + client_info: ClientInfo, +) -> Result { + let version = env!("CARGO_PKG_VERSION").to_string(); + let server_auth = + std::env::var("XIAO_SERVER_AUTH").unwrap_or_else(|_| "xiao-server".to_string()); + let client_auth = + std::env::var("XIAO_CLIENT_AUTH").unwrap_or_else(|_| "xiao-client".to_string()); + + // 发送 ClientHello + conn.send(&ControlPacket::ClientHello { + auth: server_auth, + version: version.clone(), + udp_port: audio_port, + info: client_info, + }) + .await?; + + // 等待 ServerHello + let server_udp_port = match conn.recv().await? { + ControlPacket::ServerHello { + version: v, + udp_port, + auth, + } => { + if v != version { + return Err(anyhow!("Server version mismatch: {} != {}", v, version)); + } + if auth != client_auth { + return Err(anyhow!("Invalid server auth")); + } + udp_port + } + _ => return Err(anyhow!("Expected ServerHello")), + }; + + let server_audio_addr = SocketAddr::new(server_addr.ip(), server_udp_port); + Ok(HandshakeResult { server_audio_addr }) +} diff --git a/packages/client-v2/src/app/server/audio_bus.rs b/packages/client-v2/src/app/server/audio_bus.rs new file mode 100644 index 0000000..4cc0fbc --- /dev/null +++ b/packages/client-v2/src/app/server/audio_bus.rs @@ -0,0 +1,250 @@ +//! # AudioBus - 音频总线 +//! +//! 核心的音频路由模块，采用发布-订阅模式处理多客户端音频流。 +//! +//! ## 设计理念 +//! +//! ```text +//! ┌──────────────────────────────────────────┐ +//! │ AudioBus │ +//! │ │ +//! Client A ──UDP──▶│ ┌─────────┐ ┌──────────────────┐ │ +//! │ │ Ingress │──────▶ BroadcastChannel │ │──▶ Client B, C... +//! Client B ──UDP──▶│ │ Router │ └──────────────────┘ │ +//! │ └─────────┘ │ +//! │ │ │ +//! │ ▼ │ +//! │ ┌─────────────┐ │ +//! │ │ Recorder │──▶ WAV File │ +//! │ └─────────────┘ │ +//! │ │ +//! │ ┌─────────────┐ │ +//! │ │ Playback │◀── WAV/MP3 File │──▶ Client X +//! │ └─────────────┘ │ +//! └──────────────────────────────────────────┘ +//! ``` +//! +//! ## 核心概念 +//! +//! - **Subscriber**: 订阅者，可以是客户端或录音器 +//! - **Publisher**: 发布者，可以是客户端麦克风或文件播放器 +//! - **Channel**: 频道，用于隔离不同的音频流组 + +use crate::net::network::AudioSocket; +use crate::net::protocol::AudioPacket; +use dashmap::DashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use tokio::sync::broadcast; + +/// 订阅者 ID +pub type SubscriberId = u64; + +/// 音频包及其来源 +#[derive(Clone, Debug)] +pub struct AudioFrame { + /// 音频数据包 + pub packet: AudioPacket, + /// 发送者地址（如果来自网络） + pub source: Option, + /// 时间戳（单调递增） + pub timestamp: u64, +} + +/// 订阅者信息 +#[derive(Clone)] +pub struct Subscriber { + pub id: SubscriberId, + /// UDP 目标地址 + pub addr: SocketAddr, + /// 是否过滤自己的音频（防止回声） + pub filter_self: bool, +} + +/// 音频总线 - 负责音频流的路由和分发 +pub struct AudioBus { + /// UDP socket 用于收发音频 + socket: Arc, + + /// 广播通道 - 发布订阅模式的核心 + /// 所有接收到的音频都会广播到这个 channel + broadcast_tx: broadcast::Sender, + + /// 订阅者注册表 + /// key: SocketAddr (UDP 地址) + /// value: Subscriber + subscribers: DashMap, + + /// 地址到订阅者 ID 的反向映射 + addr_to_id: DashMap, + + /// ID 生成器 + next_id: AtomicU64, + + /// 全局时间戳 + timestamp: AtomicU64, +} + +impl AudioBus { + /// 创建新的音频总线 + pub async fn new() -> anyhow::Result { + let socket = Arc::new(AudioSocket::bind().await?); + // 广播 channel 容量，设置较大以容纳多个订阅者 + let (broadcast_tx, _) = broadcast::channel(256); + + Ok(Self { + socket, + broadcast_tx, + subscribers: DashMap::new(), + addr_to_id: DashMap::new(), + next_id: AtomicU64::new(1), + timestamp: AtomicU64::new(0), + }) + } + + /// 获取 UDP 端口 + pub fn port(&self) -> u16 { + self.socket.port() + } + + /// 获取 socket 引用（用于外部发送） + pub fn socket(&self) -> Arc { + self.socket.clone() + } + + /// 注册一个订阅者 + pub fn register(&self, addr: SocketAddr, filter_self: bool) -> SubscriberId { + let id = self.next_id.fetch_add(1, Ordering::SeqCst); + let subscriber = Subscriber { + id, + addr, + filter_self, + }; + self.subscribers.insert(addr, subscriber); + self.addr_to_id.insert(addr, id); + println!("[AudioBus] Subscriber {id} registered at {addr}"); + id + } + + /// 注销订阅者 + pub fn unregister(&self, addr: &SocketAddr) { + if let Some((_, sub)) = self.subscribers.remove(addr) { + self.addr_to_id.remove(addr); + println!("[AudioBus] Subscriber {} unregistered", sub.id); + } + } + + /// 订阅广播频道，返回一个 Receiver + pub fn subscribe(&self) -> broadcast::Receiver { + self.broadcast_tx.subscribe() + } + + /// 发布音频帧到总线 + pub fn publish(&self, packet: AudioPacket, source: Option) { + let ts = self.timestamp.fetch_add(1, Ordering::SeqCst); + let frame = AudioFrame { + packet, + source, + timestamp: ts, + }; + // 忽略没有订阅者的情况 + let _ = self.broadcast_tx.send(frame); + } + + /// 广播音频帧到所有订阅者（除了发送者自己） + pub async fn broadcast(&self, frame: &AudioFrame) { + for entry in self.subscribers.iter() { + let sub = entry.value(); + + // 如果启用了自过滤，跳过发送者自己 + if sub.filter_self { + if let Some(src) = &frame.source { + if src == &sub.addr { + continue; + } + } + } + + // 发送音频包 + if let Err(e) = self.socket.send(&frame.packet, sub.addr).await { + eprintln!("[AudioBus] Failed to send to {}: {}", sub.addr, e); + } + } + } + + /// 发送音频包到指定地址 + pub async fn send_to(&self, packet: &AudioPacket, addr: SocketAddr) -> anyhow::Result<()> { + self.socket.send(packet, addr).await + } + + /// 启动 UDP 接收循环 + /// 这是一个独立的任务，负责： + /// 1. 接收 UDP 音频包 + /// 2. 发布到广播频道 + pub async fn run_receiver(&self) { + let mut buf = vec![0u8; 4096]; + loop { + match self.socket.recv(&mut buf).await { + Ok((packet, src_addr)) => { + // 只处理已注册的发送者 + if self.subscribers.contains_key(&src_addr) { + self.publish(packet, Some(src_addr)); + } + } + Err(e) => { + eprintln!("[AudioBus] Recv error: {}", e); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + } + } + } + + /// 启动广播分发循环 + /// 订阅广播频道并将音频转发给所有订阅者 + pub async fn run_broadcaster(self: Arc) { + let mut rx = self.subscribe(); + loop { + match rx.recv().await { + Ok(frame) => { + self.broadcast(&frame).await; + } + Err(broadcast::error::RecvError::Lagged(n)) => { + eprintln!("[AudioBus] Broadcaster lagged {} frames", n); + } + Err(broadcast::error::RecvError::Closed) => { + break; + } + } + } + } + + /// 获取当前订阅者数量 + pub fn subscriber_count(&self) -> usize { + self.subscribers.len() + } + + /// 检查地址是否已注册 + pub fn is_registered(&self, addr: &SocketAddr) -> bool { + self.subscribers.contains_key(addr) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_audio_bus_basic() { + let bus = AudioBus::new().await.unwrap(); + let addr: SocketAddr = "127.0.0.1:12345".parse().unwrap(); + + let id = bus.register(addr, true); + assert!(bus.is_registered(&addr)); + assert_eq!(bus.subscriber_count(), 1); + + bus.unregister(&addr); + assert!(!bus.is_registered(&addr)); + assert_eq!(bus.subscriber_count(), 0); + } +} diff --git a/packages/client-v2/src/app/server/audio_manager.rs b/packages/client-v2/src/app/server/audio_manager.rs deleted file mode 100644 index 789f8b0..0000000 --- a/packages/client-v2/src/app/server/audio_manager.rs +++ /dev/null @@ -1,208 +0,0 @@ -use crate::audio::codec::OpusCodec; -use crate::audio::config::AudioConfig; -use crate::audio::wav::{WavReader, WavWriter}; -use crate::net::network::AudioSocket; -use crate::net::protocol::AudioPacket; -use anyhow::Result; -use parking_lot::Mutex; -use std::net::SocketAddr; -use std::sync::Arc; -use tokio::sync::mpsc; -use tokio_util::sync::CancellationToken; -use tokio_util::task::TaskTracker; - -pub enum RecorderCommand { - Start { - config: AudioConfig, - filename: String, - }, - Stop, -} - -pub struct ServerAudioManager { - session_cancel: CancellationToken, - record_cancel: Mutex>, - play_cancel: Mutex>, - audio_tx: mpsc::Sender, - recorder_tx: mpsc::Sender, - tracker: TaskTracker, -} - -impl ServerAudioManager { - pub fn new( - session_cancel: CancellationToken, - tracker: TaskTracker, - ) -> ( - Self, - mpsc::Receiver, - mpsc::Receiver, - ) { - let (audio_tx, audio_rx) = mpsc::channel(1024); - let (recorder_tx, recorder_rx) = mpsc::channel(64); - - let manager = Self { - session_cancel, - record_cancel: Mutex::new(None), - play_cancel: Mutex::new(None), - audio_tx, - recorder_tx, - tracker, - }; - - (manager, audio_rx, recorder_rx) - } - - pub fn audio_tx(&self) -> mpsc::Sender { - self.audio_tx.clone() - } - - pub async fn start_recording(&self, config: AudioConfig, filename: String) -> Result<()> { - // 仅停止之前的录音任务 - { - let mut guard = self.record_cancel.lock(); - if let Some(token) = guard.take() { - token.cancel(); - } - *guard = Some(self.session_cancel.child_token()); - } - - self.recorder_tx - .send(RecorderCommand::Start { config, filename }) - .await?; - Ok(()) - } - - pub async fn stop_recording(&self) -> Result<()> { - if let Some(token) = self.record_cancel.lock().take() { - token.cancel(); - } - let _ = self.recorder_tx.send(RecorderCommand::Stop).await; - Ok(()) - } - - pub async fn start_playback( - &self, - config: AudioConfig, - reader: WavReader, - audio_socket: Arc, - target_addr: SocketAddr, - ) -> Result<()> { - let token = { - let mut guard = self.play_cancel.lock(); - if let Some(token) = guard.take() { - token.cancel(); - } - let token = self.session_cancel.child_token(); - *guard = Some(token.clone()); - token - }; - - let session_cancel = self.session_cancel.clone(); - - self.tracker.spawn(async move { - let mut reader = reader; - let mut codec = match OpusCodec::new(&config) { - Ok(c) => c, - Err(e) => { - eprintln!("Failed to create opus codec: {}", e); - return; - } - }; - let mut pcm = vec![0i16; config.frame_size]; - let mut opus = vec![0u8; 4096]; - let mut interval = tokio::time::interval(std::time::Duration::from_millis(20)); - - loop { - tokio::select! { - _ = token.cancelled() => break, - _ = session_cancel.cancelled() => break, - _ = interval.tick() => { - match reader.read_samples(&mut pcm) { - Ok(0) => break, - Ok(n) => { - if let Ok(len) = codec.encode(&pcm[..n], &mut opus) { - let _ = audio_socket - .send( - &AudioPacket { - data: opus[..len].to_vec(), - }, - target_addr, - ) - .await; - } - } - Err(e) => { - eprintln!("Failed to read samples: {}", e); - break; - } - } - } - } - } - }); - - Ok(()) - } - - pub fn stop_playback(&self) { - if let Some(token) = self.play_cancel.lock().take() { - token.cancel(); - } - } - - pub fn spawn_audio_processor( - &self, - mut audio_rx: mpsc::Receiver, - mut recorder_rx: mpsc::Receiver, - ) { - let session_cancel = self.session_cancel.clone(); - self.tracker.spawn(async move { - let mut active_recorder: Option<(WavWriter, OpusCodec, usize)> = None; - loop { - tokio::select! { - _ = session_cancel.cancelled() => break, - cmd = recorder_rx.recv() => { - match cmd { - Some(RecorderCommand::Start { config, filename }) => { - if let Some((writer, _, _)) = active_recorder.take() { - let _ = writer.finalize(); - } - match WavWriter::create(&filename, config.sample_rate, config.channels) { - Ok(writer) => { - match OpusCodec::new(&config) { - Ok(codec) => active_recorder = Some((writer, codec, config.frame_size)), - Err(e) => eprintln!("Failed to create opus codec: {}", e), - } - } - Err(e) => eprintln!("Failed to create wav writer: {}", e), - } - } - Some(RecorderCommand::Stop) => { - if let Some((writer, _, _)) = active_recorder.take() { - let _ = writer.finalize(); - } - } - None => break, - } - } - packet = audio_rx.recv() => { - match packet { - Some(packet) => { - if let Some((writer, codec, frame_size)) = &mut active_recorder { - let mut pcm = vec![0i16; *frame_size]; - if let Ok(n) = codec.decode(&packet.data, &mut pcm) { - let _ = writer.write_samples(&pcm[..n]); - } - } - } - None => break, - } - } - } - } - if let Some((writer, _, _)) = active_recorder.take() { - let _ = writer.finalize(); - } - }); - } -} diff --git a/packages/client-v2/src/app/server/mod.rs b/packages/client-v2/src/app/server/mod.rs index c063840..69c4ea5 100644 --- a/packages/client-v2/src/app/server/mod.rs +++ b/packages/client-v2/src/app/server/mod.rs @@ -1,109 +1,228 @@ -mod audio_manager; +//! # Server 模块 +//! +//! 实时音频流服务器，支持： +//! - 多客户端连接管理 +//! - 音频流发布-订阅 +//! - 录音和播放 +//! - RPC 远程调用 +//! - 实时事件推送 +//! +//! ## 架构 +//! +//! ```text +//! ┌─────────────────────────────────────────────┐ +//! │ Server │ +//! │ │ +//! Client 1 ──TCP────────┼──▶ SessionManager │ +//! Client 2 ──TCP────────┼──▶ ├─ Session 1 │ +//! Client N ──TCP────────┼──▶ ├─ Session 2 │ +//! │ └─ Session N │ +//! │ │ │ +//! ┌────────────────┼────────────┴────────────────────────┐ │ +//! │ │ │ │ +//! ▼ │ ▼ │ +//! ┌─────────┐ │ ┌─────────────────────────────┐ │ +//! │Command │ │ │ AudioBus │ │ +//! │Handler │ │ │ │ │ +//! └─────────┘ │ │ ┌─────────┐ ┌──────────┐ │ │ +//! │ │ │ │Receiver │ │Broadcaster│ │ │ +//! ▼ │ │ │ Loop │─▶│ Loop │──┼───┼──▶ All Clients +//! ┌─────────┐ │ │ └─────────┘ └──────────┘ │ │ +//! │ Event │ │ └─────────────────────────────┘ │ +//! │ Bus │───────────┼─────────────────────────────────────────────┼──▶ Events +//! └─────────┘ │ │ +//! └─────────────────────────────────────────────┘ +//! ``` + +mod audio_bus; +mod session; +mod stream; + +pub use audio_bus::{AudioBus, AudioFrame}; +pub use session::{Session, SessionManager}; +pub use stream::{FilePlaybackStream, RecorderStream, StreamHandle}; use crate::audio::config::AudioConfig; use crate::audio::wav::WavReader; +use crate::net::command::{ + AudioState, Command, CommandError, CommandResult, DeviceInfo, ShellResponse, +}; use crate::net::discovery::Discovery; -use crate::net::network::{AudioSocket, Connection}; -use crate::net::protocol::{ClientInfo, ControlPacket, RpcResult}; -use crate::net::rpc::RpcManager; +use crate::net::event::{ClientEvent, ServerEvent, ServerEventBus}; +use crate::net::network::Connection; +use crate::net::protocol::ControlPacket; use anyhow::{Context, Result, anyhow}; -use audio_manager::ServerAudioManager; -use dashmap::DashMap; use std::net::SocketAddr; use std::sync::Arc; use tokio_util::sync::CancellationToken; -use tokio_util::task::TaskTracker; - -pub struct Session { - pub info: ClientInfo, - pub conn: Arc, - pub rpc: Arc, - pub tcp_addr: SocketAddr, - pub audio_addr: SocketAddr, - pub session_cancel: CancellationToken, - pub audio_manager: ServerAudioManager, - pub tracker: TaskTracker, -} +/// 实时音频服务器 pub struct Server { - sessions: DashMap>, - udp_to_tcp: DashMap, - audio: Arc, + /// 会话管理器 + sessions: Arc, + /// 音频总线 + audio_bus: Arc, + /// 服务端事件总线 + event_bus: Arc, + /// 服务器取消令牌 + cancel: CancellationToken, + /// 服务器启动时间 + started_at: std::time::Instant, } impl Server { + /// 创建新服务器 pub async fn new() -> Result { + let audio_bus = Arc::new(AudioBus::new().await?); + Ok(Self { - sessions: DashMap::new(), - udp_to_tcp: DashMap::new(), - audio: Arc::new(AudioSocket::bind().await?), + sessions: Arc::new(SessionManager::new()), + audio_bus, + event_bus: Arc::new(ServerEventBus::default()), + cancel: CancellationToken::new(), + started_at: std::time::Instant::now(), }) } + /// 获取事件总线（用于外部订阅） + pub fn event_bus(&self) -> Arc { + self.event_bus.clone() + } + + /// 启动服务器 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); + println!("[Server] Listening on TCP: {}", addr); + println!("[Server] Audio UDP port: {}", self.audio_bus.port()); + + // 广播服务发现 Discovery::broadcast(port).await?; - // Audio Dispatcher - let this = self.clone(); + // 启动音频总线接收循环 + let bus = self.audio_bus.clone(); tokio::spawn(async move { - let mut buf = vec![0u8; 4096]; - loop { - match this.audio.recv(&mut buf).await { - Ok((packet, src_addr)) => { - if let Some(tcp_addr) = this.udp_to_tcp.get(&src_addr) { - if let Some(session) = this.sessions.get(tcp_addr.value()) { - // 使用 try_send 避免某一个客户端阻塞导致全局音频延迟 - let _ = session.audio_manager.audio_tx().try_send(packet); - } + bus.run_receiver().await; + }); + + // 启动音频总线广播循环 + let bus = self.audio_bus.clone(); + tokio::spawn(async move { + bus.run_broadcaster().await; + }); + + // TCP 连接接受循环 + loop { + tokio::select! { + _ = self.cancel.cancelled() => { + println!("[Server] Shutting down..."); + break; + } + result = listener.accept() => { + match result { + Ok((stream, addr)) => { + let server = self.clone(); + tokio::spawn(async move { + if let Err(e) = server.handle_connection(stream, addr).await { + eprintln!("[Server] Connection {} error: {}", addr, e); + } + }); + } + Err(e) => { + eprintln!("[Server] Accept error: {}", e); } - } - Err(e) => { - eprintln!("Audio socket recv error: {}", e); - tokio::time::sleep(std::time::Duration::from_millis(100)).await; } } } - }); - - loop { - let (stream, addr) = listener.accept().await?; - let server = self.clone(); - tokio::spawn(async move { - if let Err(e) = server.clone().handle_session(stream, addr).await { - eprintln!("Session {} error: {}", addr, e); - } - server.remove_session(&addr).await; - }); } + + Ok(()) } - async fn remove_session(&self, tcp_addr: &SocketAddr) { - if let Some((_, session)) = self.sessions.remove(tcp_addr) { - session.session_cancel.cancel(); - session.tracker.close(); - self.udp_to_tcp.remove(&session.audio_addr); - println!("Session {} ({}) closed", tcp_addr, session.info.model); - } - } - - async fn handle_session( + /// 处理新连接 + async fn handle_connection( self: Arc, stream: tokio::net::TcpStream, addr: SocketAddr, ) -> Result<()> { - println!("New TCP connection from {}", addr); + println!("[Server] New connection from {}", addr); let conn = Arc::new(Connection::new(stream)?); - // --- 握手 (Handshake) --- + // --- 握手 --- + let (info, audio_addr) = self.handshake(&conn, addr).await?; + println!( + "[Server] Client identified: {} ({}) audio: {}", + info.model, info.serial_number, audio_addr + ); + + // --- 创建 Session --- + let session_cancel = self.cancel.child_token(); + let session = Arc::new(Session::new( + info.clone(), + conn.clone(), + addr, + audio_addr, + session_cancel.clone(), + )); + + // 注册到 SessionManager 和 AudioBus + self.sessions.register(session.clone()); + self.audio_bus.register(audio_addr, true); + + // 发布客户端加入事件 + self.event_bus.publish(ServerEvent::ClientJoined { + addr: addr.to_string(), + model: info.model.clone(), + }); + + // 广播给其他客户端 + self.sessions + .broadcast_except( + &ControlPacket::ServerEvent(ServerEvent::ClientJoined { + addr: addr.to_string(), + model: info.model.clone(), + }), + &addr, + ) + .await; + + // --- 主循环 --- + let result = self.session_loop(session.clone()).await; + + // --- 清理 --- + self.audio_bus.unregister(&audio_addr); + self.sessions.unregister(&addr); + + // 发布客户端离开事件 + self.event_bus.publish(ServerEvent::ClientLeft { + addr: addr.to_string(), + model: info.model.clone(), + }); + + // 广播给其他客户端 + self.sessions + .broadcast(&ControlPacket::ServerEvent(ServerEvent::ClientLeft { + addr: addr.to_string(), + model: info.model, + })) + .await; + + result + } + + /// 握手流程 + async fn handshake( + &self, + conn: &Arc, + addr: SocketAddr, + ) -> Result<(crate::net::protocol::ClientInfo, SocketAddr)> { let version = env!("CARGO_PKG_VERSION").to_string(); let server_auth = std::env::var("XIAO_SERVER_AUTH").unwrap_or_else(|_| "xiao-server".to_string()); let client_auth = std::env::var("XIAO_CLIENT_AUTH").unwrap_or_else(|_| "xiao-client".to_string()); + // 等待客户端 Hello let (info, client_audio_port) = match conn.recv().await? { ControlPacket::ClientHello { auth, @@ -112,71 +231,45 @@ impl Server { info, } => { if v != version { - return Err(anyhow!("Client version mismatch: {} != {}", v, version)); + return Err(anyhow!("Version mismatch: {} != {}", v, version)); } if auth != server_auth { return Err(anyhow!("Invalid client auth")); } (info, udp_port) } - _ => return Err(anyhow!("Handshake failed")), + _ => return Err(anyhow!("Expected ClientHello")), }; + // 发送服务器 Hello conn.send(&ControlPacket::ServerHello { auth: client_auth, - version: env!("CARGO_PKG_VERSION").to_string(), - udp_port: self.audio.port(), + version, + udp_port: self.audio_bus.port(), }) .await?; let audio_addr = SocketAddr::new(addr.ip(), client_audio_port); - println!( - "Client identified: {} ({}), audio at {}", - info.model, info.serial_number, audio_addr - ); + Ok((info, audio_addr)) + } - // --- 初始化 Session --- - let tracker = TaskTracker::new(); - let session_cancel = CancellationToken::new(); - let (audio_manager, audio_rx, recorder_rx) = - ServerAudioManager::new(session_cancel.clone(), tracker.clone()); + /// Session 消息循环 + async fn session_loop(&self, session: Arc) -> Result<()> { + let timeout = std::time::Duration::from_secs(60); - let session = Arc::new(Session { - info, - conn: conn.clone(), - rpc: Arc::new(RpcManager::new()), - tcp_addr: addr, - audio_addr, - session_cancel, - audio_manager, - tracker: tracker.clone(), - }); - - session - .audio_manager - .spawn_audio_processor(audio_rx, recorder_rx); - - self.sessions.insert(addr, session.clone()); - self.udp_to_tcp.insert(audio_addr, addr); - - // 消息主循环 loop { tokio::select! { - _ = session.session_cancel.cancelled() => break, - res = tokio::time::timeout(std::time::Duration::from_secs(60), conn.recv()) => { - match res { + _ = session.cancel.cancelled() => break, + result = tokio::time::timeout(timeout, session.recv()) => { + match result { Ok(Ok(packet)) => { - if let Err(e) = self.process_packet(session.clone(), packet).await { - eprintln!("Process packet error: {}", e); - } + self.handle_packet(&session, packet).await?; } Ok(Err(e)) => { - eprintln!("Session {} connection error: {}", addr, e); - break; + return Err(anyhow!("Connection error: {}", e)); } Err(_) => { - eprintln!("Session {} timeout", addr); - break; + return Err(anyhow!("Connection timeout")); } } } @@ -186,119 +279,209 @@ impl Server { Ok(()) } - pub async fn get_clients(&self) -> Vec { - self.sessions.iter().map(|r| *r.key()).collect() - } - - // todo 考虑有些操作比较耗时，需要非阻塞处理 - async fn process_packet(&self, session: Arc, packet: ControlPacket) -> Result<()> { + /// 处理控制包 + async fn handle_packet(&self, session: &Arc, packet: ControlPacket) -> Result<()> { match packet { ControlPacket::Ping => { - session.conn.send(&ControlPacket::Pong).await?; + session.send(&ControlPacket::Pong).await?; } ControlPacket::Pong => {} ControlPacket::RpcResponse { id, result } => { - session.rpc.resolve(id, result); + session.resolve_rpc(id, result); } - ControlPacket::RpcRequest { id, method, args } => { - let result = self.handle_rpc(&session, &method, args).await; + ControlPacket::RpcRequest { id, command } => { + let result = self.handle_command(session, command).await; session - .conn .send(&ControlPacket::RpcResponse { id, result }) .await?; } + ControlPacket::ClientEvent(event) => { + self.handle_client_event(session, event).await; + } _ => {} } Ok(()) } - async fn handle_rpc( - &self, - _session: &Arc, - method: &str, - args: Vec, - ) -> RpcResult { - match method { - "hello" => RpcResult { - stdout: format!("Hello from server! Args: {:?}", args), - ..Default::default() - }, - _ => RpcResult { - stderr: format!("Unknown method: {}", method), - code: -1, - ..Default::default() - }, + /// 处理 RPC 命令 + async fn handle_command(&self, session: &Arc, command: Command) -> CommandResult { + match command { + Command::Shell(req) => { + // 转发给客户端执行 + match session.execute(Command::Shell(req)).await { + Ok(result) => result, + Err(e) => CommandResult::Error(CommandError::internal(e.to_string())), + } + } + Command::GetInfo => { + // 返回服务器信息 + CommandResult::Info(DeviceInfo { + model: "XiaoAi-Server".to_string(), + serial_number: "SERVER-001".to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + uptime_secs: self.started_at.elapsed().as_secs(), + audio_state: AudioState { + is_recording: session.is_recording(), + is_playing: session.is_playing(), + volume: 100, + }, + }) + } + Command::Ping { timestamp } => { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + CommandResult::Pong { + timestamp, + server_time: now, + } + } + Command::SetVolume(_) => { + // 转发给客户端 + match session.execute(command).await { + Ok(result) => result, + Err(e) => CommandResult::Error(CommandError::internal(e.to_string())), + } + } + _ => CommandResult::Error(CommandError::not_implemented()), } } - pub async fn call( - &self, - addr: SocketAddr, - method: &str, - args: Vec, - ) -> Result { - let session = self - .sessions - .get(&addr) - .map(|r| r.value().clone()) - .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?) + /// 处理客户端事件 + async fn handle_client_event(&self, session: &Arc, event: ClientEvent) { + match &event { + ClientEvent::Alert { level, message } => { + println!( + "[Event] Alert from {}: [{:?}] {}", + session.tcp_addr, level, message + ); + } + ClientEvent::AudioLevel { + level_db, + is_silent, + } => { + println!( + "[Event] Audio level from {}: {:.1}dB (silent: {})", + session.tcp_addr, level_db, is_silent + ); + } + _ => {} + } + + // 可以在这里将客户端事件转发给其他订阅者 } + // ==================== 公开 API ==================== + + /// 获取所有已连接的客户端地址 + pub async fn get_clients(&self) -> Vec { + self.sessions.all_addrs() + } + + /// 获取客户端数量 + pub fn client_count(&self) -> usize { + self.sessions.count() + } + + /// 获取服务器运行时间 + pub fn uptime_secs(&self) -> u64 { + self.started_at.elapsed().as_secs() + } + + /// 向客户端发起 RPC 调用（新版） + pub async fn execute(&self, addr: SocketAddr, command: Command) -> Result { + let session = self.sessions.get(&addr).context("Session not found")?; + session.execute(command).await + } + + /// 执行 Shell 命令 + pub async fn shell(&self, addr: SocketAddr, cmd: &str) -> Result { + let result = self.execute(addr, Command::shell(cmd)).await?; + match result { + CommandResult::Shell(resp) => Ok(resp), + CommandResult::Error(e) => Err(anyhow!("{}", e)), + _ => Err(anyhow!("Unexpected response type")), + } + } + + /// 推送事件给所有客户端 + pub async fn broadcast_event(&self, event: ServerEvent) { + self.event_bus.publish(event.clone()); + self.sessions + .broadcast(&ControlPacket::ServerEvent(event)) + .await; + } + + /// 推送事件给指定客户端 + pub async fn send_event(&self, addr: SocketAddr, event: ServerEvent) -> Result<()> { + let session = self.sessions.get(&addr).context("Session not found")?; + self.event_bus.publish_to(addr, event.clone()); + session.send(&ControlPacket::ServerEvent(event)).await + } + + /// 开始录音 pub async fn start_record(&self, addr: SocketAddr, config: AudioConfig) -> Result<()> { - let session = self - .sessions - .get(&addr) - .map(|r| r.value().clone()) - .context("Session not found")?; + let session = self.sessions.get(&addr).context("Session not found")?; let filename = format!( "temp/recorded_{}.wav", session.info.serial_number.replace(":", "") ); - // 通知 Audio Manager 开始录音 - session - .audio_manager - .start_recording(config.clone(), filename) - .await?; + // 创建录音流，订阅音频总线 + let handle = RecorderStream::spawn( + config.clone(), + filename.clone(), + self.audio_bus.subscribe(), + Some(session.audio_addr), + session.cancel.clone(), + ); + session.start_recording(handle, config.clone()); + + // 通知客户端开始发送音频 session - .conn .send(&ControlPacket::StartRecording { config }) .await?; + // 发布事件 + session + .send(&ControlPacket::ServerEvent( + ServerEvent::AudioStatusChanged { + is_recording: true, + is_playing: session.is_playing(), + }, + )) + .await?; + + println!("[Server] Recording started for {} -> {}", addr, filename); Ok(()) } + /// 停止录音 pub async fn stop_record(&self, addr: SocketAddr) -> Result<()> { - let session = self - .sessions - .get(&addr) - .map(|r| r.value().clone()) - .context("Session not found")?; + let session = self.sessions.get(&addr).context("Session not found")?; + session.stop_recording(); + session.send(&ControlPacket::StopRecording).await?; - // 停止录音任务 - session.audio_manager.stop_recording().await?; - session.conn.send(&ControlPacket::StopRecording).await?; + // 发布事件 + session + .send(&ControlPacket::ServerEvent( + ServerEvent::AudioStatusChanged { + is_recording: false, + is_playing: session.is_playing(), + }, + )) + .await?; + + println!("[Server] Recording stopped for {}", addr); Ok(()) } + /// 开始播放 pub async fn start_play(&self, addr: SocketAddr, file_path: &str) -> Result<()> { - let session = self - .sessions - .get(&addr) - .map(|r| r.value().clone()) - .context("Session not found")?; + let session = self.sessions.get(&addr).context("Session not found")?; let reader = WavReader::open(file_path)?; let opus_rate = if reader.sample_rate > 24000 { @@ -314,31 +497,60 @@ impl Server { ..AudioConfig::music_48k() }; + // 通知客户端准备接收音频 session - .conn .send(&ControlPacket::StartPlayback { config: config.clone(), }) .await?; + // 创建播放流 + let handle = FilePlaybackStream::spawn( + config, + reader, + self.audio_bus.socket(), + session.audio_addr, + session.cancel.clone(), + ); + + session.start_playback(handle); + + // 发布事件 session - .audio_manager - .start_playback(config, reader, self.audio.clone(), session.audio_addr) + .send(&ControlPacket::ServerEvent( + ServerEvent::AudioStatusChanged { + is_recording: session.is_recording(), + is_playing: true, + }, + )) .await?; + println!("[Server] Playback started for {} from {}", addr, file_path); Ok(()) } + /// 停止播放 pub async fn stop_play(&self, addr: SocketAddr) -> Result<()> { - let session = self - .sessions - .get(&addr) - .map(|r| r.value().clone()) - .context("Session not found")?; + let session = self.sessions.get(&addr).context("Session not found")?; + session.stop_playback(); + session.send(&ControlPacket::StopPlayback).await?; - // 停止播放任务 - session.audio_manager.stop_playback(); - session.conn.send(&ControlPacket::StopPlayback).await?; + // 发布事件 + session + .send(&ControlPacket::ServerEvent( + ServerEvent::AudioStatusChanged { + is_recording: session.is_recording(), + is_playing: false, + }, + )) + .await?; + + println!("[Server] Playback stopped for {}", addr); Ok(()) } + + /// 关闭服务器 + pub fn shutdown(&self) { + self.cancel.cancel(); + } } diff --git a/packages/client-v2/src/app/server/session.rs b/packages/client-v2/src/app/server/session.rs new file mode 100644 index 0000000..05cc33e --- /dev/null +++ b/packages/client-v2/src/app/server/session.rs @@ -0,0 +1,326 @@ +//! # Session - 客户端会话管理 +//! +//! 轻量级的会话结构，专注于： +//! - TCP 控制连接 +//! - RPC 管理 +//! - 会话生命周期 +//! +//! 音频流的实际处理由 AudioBus 和 Stream 模块负责。 + +use crate::audio::config::AudioConfig; +use crate::net::command::{Command, CommandResult}; +use crate::net::network::Connection; +use crate::net::protocol::{ClientInfo, ControlPacket}; +use crate::net::rpc::RpcManager; +use anyhow::{Context, Result}; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio_util::sync::CancellationToken; + +use super::stream::StreamHandle; + +/// 活动流追踪 +/// 用于追踪当前 session 的活动音频流 +pub struct ActiveStreams { + /// 当前录音流句柄 + pub recorder: Option, + /// 当前播放流句柄 + pub playback: Option, +} + +impl Default for ActiveStreams { + fn default() -> Self { + Self { + recorder: None, + playback: None, + } + } +} + +impl ActiveStreams { + /// 停止所有活动流 + pub fn stop_all(&mut self) { + if let Some(h) = self.recorder.take() { + h.stop(); + } + if let Some(h) = self.playback.take() { + h.stop(); + } + } + + /// 开始录音（停止之前的录音） + pub fn start_recording(&mut self, handle: StreamHandle) { + if let Some(h) = self.recorder.take() { + h.stop(); + } + self.recorder = Some(handle); + } + + /// 停止录音 + pub fn stop_recording(&mut self) { + if let Some(h) = self.recorder.take() { + h.stop(); + } + } + + /// 开始播放（停止之前的播放） + pub fn start_playback(&mut self, handle: StreamHandle) { + if let Some(h) = self.playback.take() { + h.stop(); + } + self.playback = Some(handle); + } + + /// 停止播放 + pub fn stop_playback(&mut self) { + if let Some(h) = self.playback.take() { + h.stop(); + } + } + + /// 检查是否正在录音 + pub fn is_recording(&self) -> bool { + self.recorder.is_some() + } + + /// 检查是否正在播放 + pub fn is_playing(&self) -> bool { + self.playback.is_some() + } +} + +/// 客户端会话 +pub struct Session { + /// 客户端信息 + pub info: ClientInfo, + + /// TCP 控制连接 + pub conn: Arc, + + /// RPC 管理器 + pub rpc: Arc, + + /// TCP 地址（用作会话 ID） + pub tcp_addr: SocketAddr, + + /// UDP 音频地址 + pub audio_addr: SocketAddr, + + /// 会话取消令牌 + pub cancel: CancellationToken, + + /// 活动流 + streams: parking_lot::Mutex, + + /// 当前录音配置（如果正在录音） + recording_config: parking_lot::Mutex>, + + /// 会话创建时间 + created_at: std::time::Instant, +} + +impl Session { + /// 创建新会话 + pub fn new( + info: ClientInfo, + conn: Arc, + tcp_addr: SocketAddr, + audio_addr: SocketAddr, + cancel: CancellationToken, + ) -> Self { + Self { + info, + conn, + rpc: Arc::new(RpcManager::new()), + tcp_addr, + audio_addr, + cancel, + streams: parking_lot::Mutex::new(ActiveStreams::default()), + recording_config: parking_lot::Mutex::new(None), + created_at: std::time::Instant::now(), + } + } + + /// 获取会话 ID（使用 TCP 地址） + pub fn id(&self) -> SocketAddr { + self.tcp_addr + } + + /// 检查会话是否仍然有效 + pub fn is_alive(&self) -> bool { + !self.cancel.is_cancelled() + } + + /// 获取会话运行时间（秒） + pub fn uptime_secs(&self) -> u64 { + self.created_at.elapsed().as_secs() + } + + /// 发送控制包 + pub async fn send(&self, packet: &ControlPacket) -> Result<()> { + self.conn.send(packet).await + } + + /// 接收控制包 + pub async fn recv(&self) -> Result { + self.conn.recv().await + } + + /// 执行 RPC 调用（新版，使用 Command） + pub async fn execute(&self, command: Command) -> Result { + let (id, rx) = self.rpc.register(); + self.conn + .send(&ControlPacket::RpcRequest { id, command }) + .await?; + rx.await.context("RPC channel closed") + } + + /// 处理 RPC 响应 + pub fn resolve_rpc(&self, id: u32, result: CommandResult) { + self.rpc.resolve(id, result); + } + + /// 开始录音流 + pub fn start_recording(&self, handle: StreamHandle, config: AudioConfig) { + self.streams.lock().start_recording(handle); + *self.recording_config.lock() = Some(config); + } + + /// 停止录音流 + pub fn stop_recording(&self) { + self.streams.lock().stop_recording(); + *self.recording_config.lock() = None; + } + + /// 开始播放流 + pub fn start_playback(&self, handle: StreamHandle) { + self.streams.lock().start_playback(handle); + } + + /// 停止播放流 + pub fn stop_playback(&self) { + self.streams.lock().stop_playback(); + } + + /// 检查是否正在录音 + pub fn is_recording(&self) -> bool { + self.streams.lock().is_recording() + } + + /// 检查是否正在播放 + pub fn is_playing(&self) -> bool { + self.streams.lock().is_playing() + } + + /// 清理所有资源 + pub fn cleanup(&self) { + self.cancel.cancel(); + self.streams.lock().stop_all(); + } + + /// 获取当前录音配置 + pub fn recording_config(&self) -> Option { + self.recording_config.lock().clone() + } +} + +impl Drop for Session { + fn drop(&mut self) { + self.cleanup(); + } +} + +/// 会话管理器 +/// 负责管理所有客户端会话 +pub struct SessionManager { + sessions: dashmap::DashMap>, + /// 从 UDP 地址到 TCP 地址的映射 + udp_to_tcp: dashmap::DashMap, +} + +impl SessionManager { + pub fn new() -> Self { + Self { + sessions: dashmap::DashMap::new(), + udp_to_tcp: dashmap::DashMap::new(), + } + } + + /// 注册新会话 + pub fn register(&self, session: Arc) { + let tcp_addr = session.tcp_addr; + let audio_addr = session.audio_addr; + + self.udp_to_tcp.insert(audio_addr, tcp_addr); + self.sessions.insert(tcp_addr, session); + + println!( + "[SessionManager] Registered: {} (audio: {})", + tcp_addr, audio_addr + ); + } + + /// 注销会话 + pub fn unregister(&self, tcp_addr: &SocketAddr) -> Option> { + if let Some((_, session)) = self.sessions.remove(tcp_addr) { + self.udp_to_tcp.remove(&session.audio_addr); + session.cleanup(); + println!( + "[SessionManager] Unregistered: {} ({})", + tcp_addr, session.info.model + ); + Some(session) + } else { + None + } + } + + /// 通过 TCP 地址获取会话 + pub fn get(&self, tcp_addr: &SocketAddr) -> Option> { + self.sessions.get(tcp_addr).map(|r| r.value().clone()) + } + + /// 通过 UDP 地址获取会话 + pub fn get_by_udp(&self, udp_addr: &SocketAddr) -> Option> { + self.udp_to_tcp + .get(udp_addr) + .and_then(|tcp_addr| self.get(tcp_addr.value())) + } + + /// 获取所有会话地址 + pub fn all_addrs(&self) -> Vec { + self.sessions.iter().map(|r| *r.key()).collect() + } + + /// 获取所有会话 + pub fn all_sessions(&self) -> Vec> { + self.sessions.iter().map(|r| r.value().clone()).collect() + } + + /// 获取会话数量 + pub fn count(&self) -> usize { + self.sessions.len() + } + + /// 广播控制包到所有会话 + pub async fn broadcast(&self, packet: &ControlPacket) { + for entry in self.sessions.iter() { + let _ = entry.value().send(packet).await; + } + } + + /// 广播控制包到所有会话（除了指定的） + pub async fn broadcast_except(&self, packet: &ControlPacket, except: &SocketAddr) { + for entry in self.sessions.iter() { + if entry.key() != except { + let _ = entry.value().send(packet).await; + } + } + } +} + +impl Default for SessionManager { + fn default() -> Self { + Self::new() + } +} diff --git a/packages/client-v2/src/app/server/stream.rs b/packages/client-v2/src/app/server/stream.rs new file mode 100644 index 0000000..f3e2040 --- /dev/null +++ b/packages/client-v2/src/app/server/stream.rs @@ -0,0 +1,302 @@ +//! # AudioStream - 音频流抽象 +//! +//! 提供统一的音频流处理接口，支持多种输入源和输出目标。 +//! +//! ## 设计 +//! +//! ```text +//! ┌─────────────────────────────────────────────────────────────┐ +//! │ Stream Types │ +//! │ │ +//! │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ +//! │ │ FileSource │ │ BusSource │ │ NetworkSink │ │ +//! │ │ (WAV Reader) │ │ (From Bus) │ │ (To Client) │ │ +//! │ └──────────────┘ └──────────────┘ └──────────────┘ │ +//! │ │ +//! │ ┌──────────────┐ ┌──────────────┐ │ +//! │ │ FileSink │ │ BusSink │ │ +//! │ │ (WAV Writer) │ │ (To Bus) │ │ +//! │ └──────────────┘ └──────────────┘ │ +//! └─────────────────────────────────────────────────────────────┘ +//! ``` + +use crate::audio::codec::OpusCodec; +use crate::audio::config::AudioConfig; +use crate::audio::wav::{WavReader, WavWriter}; +use crate::net::network::AudioSocket; +use crate::net::protocol::AudioPacket; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::sync::broadcast; +use tokio_util::sync::CancellationToken; + +use super::audio_bus::AudioFrame; + +/// 音频流任务句柄 +/// 用于控制正在运行的音频流任务 +pub struct StreamHandle { + cancel: CancellationToken, +} + +impl StreamHandle { + pub fn new(cancel: CancellationToken) -> Self { + Self { cancel } + } + + /// 停止流 + pub fn stop(&self) { + self.cancel.cancel(); + } + + /// 检查是否已停止 + pub fn is_stopped(&self) -> bool { + self.cancel.is_cancelled() + } +} + +impl Drop for StreamHandle { + fn drop(&mut self) { + self.cancel.cancel(); + } +} + +/// 文件播放流 - 从 WAV 文件读取并发送到指定客户端 +pub struct FilePlaybackStream; + +impl FilePlaybackStream { + /// 启动文件播放流 + /// + /// # Arguments + /// * `config` - 音频配置（用于 Opus 编码） + /// * `reader` - WAV 文件读取器 + /// * `socket` - UDP socket + /// * `target` - 目标客户端地址 + /// * `parent_cancel` - 父级取消令牌（用于 session 级别取消） + pub fn spawn( + config: AudioConfig, + reader: WavReader, + socket: Arc, + target: SocketAddr, + parent_cancel: CancellationToken, + ) -> StreamHandle { + let cancel = parent_cancel.child_token(); + let token = cancel.clone(); + + tokio::spawn(async move { + if let Err(e) = Self::run(config, reader, socket, target, token).await { + eprintln!("[FilePlayback] Error: {}", e); + } + }); + + StreamHandle::new(cancel) + } + + async fn run( + config: AudioConfig, + mut reader: WavReader, + socket: Arc, + target: SocketAddr, + cancel: CancellationToken, + ) -> anyhow::Result<()> { + let mut codec = OpusCodec::new(&config)?; + let mut pcm = vec![0i16; config.frame_size]; + let mut opus_buf = vec![0u8; 4096]; + let frame_duration = std::time::Duration::from_millis(20); + let mut interval = tokio::time::interval(frame_duration); + + println!("[FilePlayback] Started -> {}", target); + + loop { + tokio::select! { + _ = cancel.cancelled() => break, + _ = interval.tick() => { + match reader.read_samples(&mut pcm) { + Ok(0) => { + println!("[FilePlayback] EOF reached"); + break; + } + Ok(n) => { + if let Ok(len) = codec.encode(&pcm[..n], &mut opus_buf) { + let packet = AudioPacket { + data: opus_buf[..len].to_vec(), + }; + let _ = socket.send(&packet, target).await; + } + } + Err(e) => { + eprintln!("[FilePlayback] Read error: {}", e); + break; + } + } + } + } + } + + println!("[FilePlayback] Stopped"); + Ok(()) + } +} + +/// 录音流 - 从总线订阅音频并写入 WAV 文件 +pub struct RecorderStream; + +impl RecorderStream { + /// 启动录音流 + /// + /// # Arguments + /// * `config` - 音频配置 + /// * `filename` - 输出文件路径 + /// * `bus_rx` - 音频总线接收器 + /// * `source_filter` - 仅录制来自此地址的音频（None 表示全部录制） + /// * `parent_cancel` - 父级取消令牌 + pub fn spawn( + config: AudioConfig, + filename: String, + bus_rx: broadcast::Receiver, + source_filter: Option, + parent_cancel: CancellationToken, + ) -> StreamHandle { + let cancel = parent_cancel.child_token(); + let token = cancel.clone(); + + tokio::spawn(async move { + if let Err(e) = Self::run(config, filename, bus_rx, source_filter, token).await { + eprintln!("[Recorder] Error: {}", e); + } + }); + + StreamHandle::new(cancel) + } + + async fn run( + config: AudioConfig, + filename: String, + mut bus_rx: broadcast::Receiver, + source_filter: Option, + cancel: CancellationToken, + ) -> anyhow::Result<()> { + let mut writer = WavWriter::create(&filename, config.sample_rate, config.channels)?; + let mut codec = OpusCodec::new(&config)?; + let mut pcm = vec![0i16; config.frame_size]; + + println!( + "[Recorder] Started -> {} (filter: {:?})", + filename, source_filter + ); + + loop { + tokio::select! { + _ = cancel.cancelled() => break, + result = bus_rx.recv() => { + match result { + Ok(frame) => { + // 应用源过滤 + if let Some(filter_addr) = &source_filter { + if frame.source.as_ref() != Some(filter_addr) { + continue; + } + } + + // 解码并写入 + if let Ok(n) = codec.decode(&frame.packet.data, &mut pcm) { + let _ = writer.write_samples(&pcm[..n]); + } + } + Err(broadcast::error::RecvError::Lagged(n)) => { + eprintln!("[Recorder] Lagged {} frames", n); + } + Err(broadcast::error::RecvError::Closed) => { + break; + } + } + } + } + } + + // 确保正确关闭文件 + writer.finalize()?; + println!("[Recorder] Stopped, file saved: {}", filename); + Ok(()) + } +} + +/// 音频转发流 - 从总线订阅并转发到指定客户端 +/// 用于实现"监听"功能或者服务端音频源推送 +pub struct ForwardStream; + +impl ForwardStream { + /// 启动转发流 + /// + /// # Arguments + /// * `socket` - UDP socket + /// * `target` - 目标地址 + /// * `bus_rx` - 音频总线接收器 + /// * `source_filter` - 源过滤（可选） + /// * `parent_cancel` - 父级取消令牌 + pub fn spawn( + socket: Arc, + target: SocketAddr, + bus_rx: broadcast::Receiver, + source_filter: Option, + parent_cancel: CancellationToken, + ) -> StreamHandle { + let cancel = parent_cancel.child_token(); + let token = cancel.clone(); + + tokio::spawn(async move { + if let Err(e) = Self::run(socket, target, bus_rx, source_filter, token).await { + eprintln!("[Forward] Error: {}", e); + } + }); + + StreamHandle::new(cancel) + } + + async fn run( + socket: Arc, + target: SocketAddr, + mut bus_rx: broadcast::Receiver, + source_filter: Option, + cancel: CancellationToken, + ) -> anyhow::Result<()> { + println!( + "[Forward] Started -> {} (filter: {:?})", + target, source_filter + ); + + loop { + tokio::select! { + _ = cancel.cancelled() => break, + result = bus_rx.recv() => { + match result { + Ok(frame) => { + // 应用源过滤 + if let Some(filter_addr) = &source_filter { + if frame.source.as_ref() != Some(filter_addr) { + continue; + } + } + + // 不转发给自己 + if frame.source.as_ref() == Some(&target) { + continue; + } + + let _ = socket.send(&frame.packet, target).await; + } + Err(broadcast::error::RecvError::Lagged(n)) => { + eprintln!("[Forward] Lagged {} frames", n); + } + Err(broadcast::error::RecvError::Closed) => { + break; + } + } + } + } + } + + println!("[Forward] Stopped"); + Ok(()) + } +} + diff --git a/packages/client-v2/src/bin/client.rs b/packages/client-v2/src/bin/client.rs index 7850a1c..c3bdbad 100644 --- a/packages/client-v2/src/bin/client.rs +++ b/packages/client-v2/src/bin/client.rs @@ -1,7 +1,19 @@ +//! # Client Demo +//! +//! 演示客户端的主要功能： +//! - 自动服务发现 +//! - 响应 RPC 调用 +//! - 音频录制和播放 +//! - 事件处理 + #[cfg(target_os = "linux")] use std::sync::Arc; #[cfg(target_os = "linux")] -use xiao::app::client::Client; +use xiao::app::client::{Client, ClientConfig}; +#[cfg(target_os = "linux")] +use xiao::net::command::Command; +#[cfg(target_os = "linux")] +use xiao::net::event::NotificationLevel; #[cfg(not(target_os = "linux"))] fn main() { @@ -11,23 +23,98 @@ fn main() { #[cfg(target_os = "linux")] #[tokio::main] async fn main() -> anyhow::Result<()> { - let client = Arc::new(Client::new()); - let c = client.clone(); + println!("╔═══════════════════════════════════════════════════════╗"); + println!("║ XiaoAi Audio Client v{} ║", env!("CARGO_PKG_VERSION")); + println!("╚═══════════════════════════════════════════════════════╝"); + println!(); + + // 创建客户端（可以自定义配置） + let config = ClientConfig { + model: "Open-XiaoAi-Demo".to_string(), + serial_number: get_mac_address(), + heartbeat_interval: 10, + timeout: 60, + }; + + let client = Arc::new(Client::new(config)); + + // 启动事件监听器 + let event_client = client.clone(); tokio::spawn(async move { - if let Err(e) = c.run().await { + let mut rx = event_client.subscribe_events(); + while let Ok(event) = rx.recv().await { + println!("📨 [ServerEvent] {:?}", event); + } + }); + + // 启动客户端主循环 + let run_client = client.clone(); + tokio::spawn(async move { + if let Err(e) = run_client.run().await { eprintln!("Client error: {}", e); } }); - // Wait for connection - tokio::time::sleep(std::time::Duration::from_secs(3)).await; + println!("Client is running, searching for server...\n"); - println!("Testing RPC call to server..."); - match client.call("hello", vec!["world".to_string()]).await { - Ok(res) => println!("Server RPC response: {}", res.stdout), - Err(e) => eprintln!("Server RPC call failed: {}", e), + // 等待连接 + loop { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + if client.is_connected().await { + break; + } } + println!("\n═══════════════════════════════════════════════════════"); + println!("Connected to server!"); + println!("Running client-side tests...\n"); + + // 1. 测试向服务器发送 Ping + println!("1️⃣ Testing Ping to server..."); + match client.execute(Command::ping()).await { + Ok(result) => println!(" ✅ Pong received: {:?}", result), + Err(e) => println!(" ❌ Ping failed: {}", e), + } + + // 2. 测试获取服务器信息 + println!("\n2️⃣ Testing GetInfo from server..."); + match client.execute(Command::GetInfo).await { + Ok(result) => println!(" ✅ Server info: {:?}", result), + Err(e) => println!(" ❌ GetInfo failed: {}", e), + } + + // 3. 发送客户端事件 + println!("\n3️⃣ Sending alert event to server..."); + match client + .send_alert(NotificationLevel::Info, "Client started successfully!") + .await + { + Ok(_) => println!(" ✅ Alert sent"), + Err(e) => println!(" ❌ Failed to send alert: {}", e), + } + + println!("\n═══════════════════════════════════════════════════════"); + println!("✅ Client tests completed!"); + println!("\nClient is now ready to receive commands from server."); + println!("Press Ctrl+C to exit.\n"); + + // 保持运行，等待服务器命令 tokio::signal::ctrl_c().await?; + println!("\nShutting down client..."); + client.shutdown(); Ok(()) } + +#[cfg(target_os = "linux")] +fn get_mac_address() -> String { + // 尝试获取 MAC 地址 + if let Ok(output) = std::process::Command::new("cat") + .arg("/sys/class/net/eth0/address") + .output() + { + if output.status.success() { + return String::from_utf8_lossy(&output.stdout).trim().to_string(); + } + } + "00:00:00:00:00:00".to_string() +} diff --git a/packages/client-v2/src/bin/server.rs b/packages/client-v2/src/bin/server.rs index 531f5b9..9fcd011 100644 --- a/packages/client-v2/src/bin/server.rs +++ b/packages/client-v2/src/bin/server.rs @@ -1,47 +1,144 @@ +//! # Server Demo +//! +//! 演示服务端的主要功能： +//! - 多客户端管理 +//! - RPC 调用 +//! - 音频录制和播放 +//! - 事件广播 + use std::sync::Arc; use xiao::app::server::Server; use xiao::audio::config::AudioConfig; +use xiao::net::command::Command; +use xiao::net::event::{NotificationLevel, ServerEvent}; #[tokio::main] async fn main() -> anyhow::Result<()> { + println!("╔═══════════════════════════════════════════════════════╗"); + println!("║ XiaoAi Audio Server v{} ║", env!("CARGO_PKG_VERSION")); + println!("╚═══════════════════════════════════════════════════════╝"); + println!(); + let server = Arc::new(Server::new().await?); let s = server.clone(); + // 启动服务器 tokio::spawn(async move { - s.run(8080).await.unwrap(); + if let Err(e) = s.run(8080).await { + eprintln!("Server error: {}", e); + } }); - println!("Server is running. Waiting for a client to connect..."); + // 启动事件监听器 + let event_server = server.clone(); + tokio::spawn(async move { + let mut rx = event_server.event_bus().subscribe(); + while let Some((addr, event)) = rx.recv().await { + match event { + ServerEvent::ClientJoined { model, .. } => { + println!("📱 [Event] Client joined: {} ({:?})", model, addr); + } + ServerEvent::ClientLeft { model, .. } => { + println!("📴 [Event] Client left: {} ({:?})", model, addr); + } + _ => {} + } + } + }); + println!("Server is running on port 8080"); + println!("Waiting for clients to connect...\n"); + + // 主循环：等待客户端并执行测试 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, "temp/test.wav").await?; - - break; + if clients.is_empty() { + continue; } + + let addr = clients[0]; + println!("\n═══════════════════════════════════════════════════════"); + println!("Client connected: {}", addr); + println!("Running demo tests...\n"); + + // 1. 测试 Ping + println!("1️⃣ Testing Ping..."); + match server.execute(addr, Command::ping()).await { + Ok(result) => println!(" ✅ Ping result: {:?}", result), + Err(e) => println!(" ❌ Ping failed: {}", e), + } + + // 2. 测试获取设备信息 + println!("\n2️⃣ Testing GetInfo..."); + match server.execute(addr, Command::GetInfo).await { + Ok(result) => println!(" ✅ Device info: {:?}", result), + Err(e) => println!(" ❌ GetInfo failed: {}", e), + } + + // 3. 测试 Shell 命令 + println!("\n3️⃣ Testing Shell RPC..."); + match server.shell(addr, "echo 'Hello from XiaoAi!'").await { + Ok(resp) => { + println!(" ✅ stdout: {}", resp.stdout.trim()); + println!(" ✅ exit_code: {}", resp.exit_code); + } + Err(e) => println!(" ❌ Shell failed: {}", e), + } + + // 4. 测试事件广播 + println!("\n4️⃣ Broadcasting notification event..."); + server + .broadcast_event(ServerEvent::Notification { + level: NotificationLevel::Info, + title: "Test".to_string(), + message: "This is a test notification from server".to_string(), + }) + .await; + println!(" ✅ Event broadcasted"); + + // 5. 测试音频录制 + println!("\n5️⃣ Testing Audio Recording (5 seconds)..."); + match server.start_record(addr, AudioConfig::voice_16k()).await { + Ok(_) => { + println!(" ⏺️ Recording started..."); + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + server.stop_record(addr).await?; + println!(" ⏹️ Recording stopped"); + } + Err(e) => println!(" ❌ Recording failed: {}", e), + } + + // 6. 测试音频播放（如果有测试文件） + println!("\n6️⃣ Testing Audio Playback..."); + if std::path::Path::new("temp/test.wav").exists() { + match server.start_play(addr, "temp/test.wav").await { + Ok(_) => { + println!(" ▶️ Playback started..."); + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + server.stop_play(addr).await?; + println!(" ⏹️ Playback stopped"); + } + Err(e) => println!(" ❌ Playback failed: {}", e), + } + } else { + println!(" ⚠️ No test file found at temp/test.wav, skipping..."); + } + + println!("\n═══════════════════════════════════════════════════════"); + println!("✅ All tests completed!"); + println!("\nServer status:"); + println!(" • Connected clients: {}", server.client_count()); + println!(" • Uptime: {} seconds", server.uptime_secs()); + println!("\nPress Ctrl+C to exit."); + + break; } - println!("Tests completed. Press Ctrl+C to exit."); + // 等待退出信号 tokio::signal::ctrl_c().await?; + println!("\nShutting down server..."); + server.shutdown(); Ok(()) } diff --git a/packages/client-v2/src/net/command.rs b/packages/client-v2/src/net/command.rs new file mode 100644 index 0000000..150c717 --- /dev/null +++ b/packages/client-v2/src/net/command.rs @@ -0,0 +1,324 @@ +//! # Command - RPC 命令类型系统 +//! +//! 支持多种类型的命令，每种命令有独立的请求和响应结构。 +//! +//! ## 设计 +//! +//! ```text +//! ┌─────────────────────────────────────────────────────────────┐ +//! │ Command Types │ +//! │ │ +//! │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ +//! │ │ Shell │ │ GetInfo │ │ SetVolume │ ... │ +//! │ │ cmd → out │ │ () → Info │ │ vol → () │ │ +//! │ └──────────────┘ └──────────────┘ └──────────────┘ │ +//! │ │ +//! │ ┌──────────────────┐ │ +//! │ │ RpcRequest │ │ +//! │ │ id + Command │ │ +//! │ └────────┬─────────┘ │ +//! │ │ │ +//! │ ▼ │ +//! │ ┌──────────────────┐ │ +//! │ │ RpcResponse │ │ +//! │ │ id + Result │ │ +//! │ └──────────────────┘ │ +//! └─────────────────────────────────────────────────────────────┘ +//! ``` + +use serde::{Deserialize, Serialize}; + +// ==================== 命令请求类型 ==================== + +/// Shell 命令请求 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct ShellRequest { + /// 要执行的命令 + pub command: String, + /// 工作目录（可选） + pub cwd: Option, + /// 环境变量（可选） + pub env: Option>, + /// 超时秒数（可选） + pub timeout_secs: Option, +} + +/// Shell 命令响应 +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct ShellResponse { + pub stdout: String, + pub stderr: String, + pub exit_code: i32, +} + +/// 获取设备信息请求 +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct GetInfoRequest; + +/// 设备信息响应 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct DeviceInfo { + pub model: String, + pub serial_number: String, + pub version: String, + pub uptime_secs: u64, + pub audio_state: AudioState, +} + +/// 音频状态 +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct AudioState { + pub is_recording: bool, + pub is_playing: bool, + pub volume: u8, +} + +/// 设置音量请求 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct SetVolumeRequest { + pub volume: u8, // 0-100 +} + +/// 设置音量响应 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct SetVolumeResponse { + pub previous: u8, + pub current: u8, +} + +/// 文件操作请求 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub enum FileRequest { + /// 读取文件 + Read { path: String }, + /// 写入文件 + Write { path: String, data: Vec }, + /// 删除文件 + Delete { path: String }, + /// 列出目录 + List { path: String }, + /// 获取文件信息 + Stat { path: String }, +} + +/// 文件操作响应 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub enum FileResponse { + /// 读取结果 + Data(Vec), + /// 写入成功 + Written { bytes: usize }, + /// 删除成功 + Deleted, + /// 目录列表 + Entries(Vec), + /// 文件信息 + Stat(FileStat), +} + +/// 文件条目 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct FileEntry { + pub name: String, + pub is_dir: bool, + pub size: u64, +} + +/// 文件状态 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct FileStat { + pub size: u64, + pub is_dir: bool, + pub modified: u64, +} + +/// 系统控制请求 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub enum SystemRequest { + /// 重启 + Reboot, + /// 关机 + Shutdown, + /// 获取系统负载 + GetLoad, + /// 获取内存使用 + GetMemory, +} + +/// 系统控制响应 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub enum SystemResponse { + /// 操作已接受 + Accepted, + /// 系统负载 + Load { one: f32, five: f32, fifteen: f32 }, + /// 内存使用 + Memory { total: u64, used: u64, free: u64 }, +} + +/// 自定义命令请求（用于扩展） +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct CustomRequest { + pub name: String, + pub payload: Vec, +} + +/// 自定义命令响应 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct CustomResponse { + pub payload: Vec, +} + +// ==================== 统一命令枚举 ==================== + +/// RPC 命令 - 统一的请求类型 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub enum Command { + /// 执行 Shell 命令 + Shell(ShellRequest), + /// 获取设备信息 + GetInfo, + /// 设置音量 + SetVolume(SetVolumeRequest), + /// 文件操作 + File(FileRequest), + /// 系统控制 + System(SystemRequest), + /// 自定义命令 + Custom(CustomRequest), + /// Ping（用于测量延迟） + Ping { timestamp: u64 }, +} + +/// RPC 结果 - 统一的响应类型 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub enum CommandResult { + /// Shell 命令结果 + Shell(ShellResponse), + /// 设备信息 + Info(DeviceInfo), + /// 音量设置结果 + Volume(SetVolumeResponse), + /// 文件操作结果 + File(FileResponse), + /// 系统控制结果 + System(SystemResponse), + /// 自定义命令结果 + Custom(CustomResponse), + /// Pong 响应 + Pong { timestamp: u64, server_time: u64 }, + /// 错误 + Error(CommandError), +} + +/// 命令错误 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct CommandError { + pub code: i32, + pub message: String, +} + +impl CommandError { + pub fn new(code: i32, message: impl Into) -> Self { + Self { + code, + message: message.into(), + } + } + + pub fn not_found(msg: impl Into) -> Self { + Self::new(-1, msg) + } + + pub fn invalid_args(msg: impl Into) -> Self { + Self::new(-2, msg) + } + + pub fn permission_denied(msg: impl Into) -> Self { + Self::new(-3, msg) + } + + pub fn internal(msg: impl Into) -> Self { + Self::new(-500, msg) + } + + pub fn timeout(msg: impl Into) -> Self { + Self::new(-408, msg) + } + + pub fn not_implemented() -> Self { + Self::new(-501, "Not implemented") + } +} + +impl std::fmt::Display for CommandError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "[{}] {}", self.code, self.message) + } +} + +impl std::error::Error for CommandError {} + +// ==================== 便捷构造方法 ==================== + +impl Command { + /// 创建 Shell 命令 + pub fn shell(cmd: impl Into) -> Self { + Self::Shell(ShellRequest { + command: cmd.into(), + cwd: None, + env: None, + timeout_secs: None, + }) + } + + /// 创建带超时的 Shell 命令 + pub fn shell_with_timeout(cmd: impl Into, timeout: u32) -> Self { + Self::Shell(ShellRequest { + command: cmd.into(), + cwd: None, + env: None, + timeout_secs: Some(timeout), + }) + } + + /// 创建 Ping 命令 + pub fn ping() -> Self { + Self::Ping { + timestamp: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64, + } + } +} + +impl CommandResult { + /// 创建成功的 Shell 响应 + pub fn shell_ok(stdout: String) -> Self { + Self::Shell(ShellResponse { + stdout, + stderr: String::new(), + exit_code: 0, + }) + } + + /// 创建错误响应 + pub fn error(err: CommandError) -> Self { + Self::Error(err) + } + + /// 检查是否是错误 + pub fn is_error(&self) -> bool { + matches!(self, Self::Error(_)) + } + + /// 获取错误（如果有） + pub fn as_error(&self) -> Option<&CommandError> { + match self { + Self::Error(e) => Some(e), + _ => None, + } + } +} + diff --git a/packages/client-v2/src/net/event.rs b/packages/client-v2/src/net/event.rs new file mode 100644 index 0000000..2dbefd6 --- /dev/null +++ b/packages/client-v2/src/net/event.rs @@ -0,0 +1,286 @@ +//! # Event - 实时事件系统 +//! +//! 支持双向的实时事件推送，包括： +//! - 服务端事件（推送给客户端） +//! - 客户端事件（推送给服务端） +//! - 事件订阅和过滤 +//! +//! ## 设计 +//! +//! ```text +//! ┌─────────────────────────────────────────────────────────────┐ +//! │ Event System │ +//! │ │ +//! │ Server Events: │ +//! │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ +//! │ │ AudioStatus │ │ ClientJoined │ │ Message │ ... │ +//! │ └──────────────┘ └──────────────┘ └──────────────┘ │ +//! │ │ +//! │ Client Events: │ +//! │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ +//! │ │ StatusUpdate │ │ Metrics │ │ Alert │ ... │ +//! │ └──────────────┘ └──────────────┘ └──────────────┘ │ +//! │ │ +//! │ ┌─────────────────────┐ │ +//! │ │ EventBus │ │ +//! │ │ broadcast channel │ │ +//! │ └─────────────────────┘ │ +//! └─────────────────────────────────────────────────────────────┘ +//! ``` + +use serde::{Deserialize, Serialize}; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::sync::broadcast; + +// ==================== 事件类型 ==================== + +/// 服务端事件（Server → Client） +#[derive(Serialize, Deserialize, Debug, Clone)] +pub enum ServerEvent { + /// 音频状态变化 + AudioStatusChanged { + is_recording: bool, + is_playing: bool, + }, + + /// 客户端加入（广播给其他客户端） + ClientJoined { + addr: String, + model: String, + }, + + /// 客户端离开 + ClientLeft { + addr: String, + model: String, + }, + + /// 服务器消息/通知 + Notification { + level: NotificationLevel, + title: String, + message: String, + }, + + /// 录音完成 + RecordingComplete { + filename: String, + duration_secs: f32, + size_bytes: u64, + }, + + /// 播放完成 + PlaybackComplete { + filename: String, + }, + + /// 服务器状态更新 + ServerStatus { + connected_clients: u32, + uptime_secs: u64, + }, + + /// 自定义事件 + Custom { + name: String, + payload: Vec, + }, +} + +/// 客户端事件（Client → Server） +#[derive(Serialize, Deserialize, Debug, Clone)] +pub enum ClientEvent { + /// 状态更新 + StatusUpdate { + cpu_usage: f32, + memory_usage: f32, + temperature: Option, + }, + + /// 音频电平 + AudioLevel { + level_db: f32, + is_silent: bool, + }, + + /// 按键事件 + KeyPress { + key: String, + action: KeyAction, + }, + + /// 警告/错误 + Alert { + level: NotificationLevel, + message: String, + }, + + /// 自定义事件 + Custom { + name: String, + payload: Vec, + }, +} + +/// 通知级别 +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)] +pub enum NotificationLevel { + Debug, + Info, + Warning, + Error, +} + +/// 按键动作 +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)] +pub enum KeyAction { + Press, + Release, + LongPress, +} + +// ==================== 事件总线 ==================== + +/// 事件订阅者信息 +pub struct EventSubscription { + pub receiver: broadcast::Receiver, +} + +impl EventSubscription { + /// 接收下一个事件 + pub async fn recv(&mut self) -> Option { + match self.receiver.recv().await { + Ok(event) => Some(event), + Err(broadcast::error::RecvError::Lagged(_)) => { + // 跳过丢失的事件，继续接收 + Box::pin(self.recv()).await + } + Err(broadcast::error::RecvError::Closed) => None, + } + } +} + +/// 服务端事件总线 +pub struct ServerEventBus { + tx: broadcast::Sender<(Option, ServerEvent)>, +} + +impl ServerEventBus { + pub fn new(capacity: usize) -> Self { + let (tx, _) = broadcast::channel(capacity); + Self { tx } + } + + /// 发布事件（广播给所有客户端） + pub fn publish(&self, event: ServerEvent) { + let _ = self.tx.send((None, event)); + } + + /// 发布事件给指定客户端 + pub fn publish_to(&self, addr: SocketAddr, event: ServerEvent) { + let _ = self.tx.send((Some(addr), event)); + } + + /// 订阅事件 + pub fn subscribe(&self) -> EventSubscription<(Option, ServerEvent)> { + EventSubscription { + receiver: self.tx.subscribe(), + } + } +} + +impl Default for ServerEventBus { + fn default() -> Self { + Self::new(256) + } +} + +/// 客户端事件总线 +pub struct ClientEventBus { + tx: broadcast::Sender<(SocketAddr, ClientEvent)>, +} + +impl ClientEventBus { + pub fn new(capacity: usize) -> Self { + let (tx, _) = broadcast::channel(capacity); + Self { tx } + } + + /// 发布客户端事件 + pub fn publish(&self, client_addr: SocketAddr, event: ClientEvent) { + let _ = self.tx.send((client_addr, event)); + } + + /// 订阅事件 + pub fn subscribe(&self) -> EventSubscription<(SocketAddr, ClientEvent)> { + EventSubscription { + receiver: self.tx.subscribe(), + } + } +} + +impl Default for ClientEventBus { + fn default() -> Self { + Self::new(256) + } +} + +// ==================== 事件过滤器 ==================== + +/// 服务端事件过滤器 +pub trait ServerEventFilter: Send + Sync { + fn should_process(&self, event: &ServerEvent) -> bool; +} + +/// 接受所有事件 +pub struct AcceptAll; + +impl ServerEventFilter for AcceptAll { + fn should_process(&self, _: &ServerEvent) -> bool { + true + } +} + +/// 只接受通知事件 +pub struct NotificationsOnly; + +impl ServerEventFilter for NotificationsOnly { + fn should_process(&self, event: &ServerEvent) -> bool { + matches!(event, ServerEvent::Notification { .. }) + } +} + +/// 只接受音频相关事件 +pub struct AudioEventsOnly; + +impl ServerEventFilter for AudioEventsOnly { + fn should_process(&self, event: &ServerEvent) -> bool { + matches!( + event, + ServerEvent::AudioStatusChanged { .. } + | ServerEvent::RecordingComplete { .. } + | ServerEvent::PlaybackComplete { .. } + ) + } +} + +// ==================== 事件处理器 ==================== + +/// 事件处理器 trait +#[allow(async_fn_in_trait)] +pub trait EventHandler: Send + Sync { + async fn handle(&self, event: E); +} + +/// 运行事件处理循环 +pub async fn run_event_loop(mut subscription: EventSubscription, handler: Arc) +where + E: Clone + Send + 'static, + H: EventHandler + 'static, +{ + while let Some(event) = subscription.recv().await { + handler.handle(event).await; + } +} + diff --git a/packages/client-v2/src/net/mod.rs b/packages/client-v2/src/net/mod.rs index af1fd80..dc26e0c 100644 --- a/packages/client-v2/src/net/mod.rs +++ b/packages/client-v2/src/net/mod.rs @@ -1,4 +1,16 @@ +//! # Net 模块 +//! +//! 网络通信相关模块： +//! - `command` - RPC 命令类型系统 +//! - `discovery` - 服务发现 +//! - `event` - 实时事件系统 +//! - `network` - 底层网络连接 +//! - `protocol` - 通信协议定义 +//! - `rpc` - RPC 调用管理 + +pub mod command; pub mod discovery; +pub mod event; pub mod network; pub mod protocol; pub mod rpc; diff --git a/packages/client-v2/src/net/protocol.rs b/packages/client-v2/src/net/protocol.rs index 8c9a344..81b4f67 100644 --- a/packages/client-v2/src/net/protocol.rs +++ b/packages/client-v2/src/net/protocol.rs @@ -1,59 +1,90 @@ +//! # Protocol - 通信协议定义 +//! +//! 定义 Client 和 Server 之间的所有通信协议。 + use crate::audio::config::AudioConfig; +use crate::net::command::{Command, CommandResult}; +use crate::net::event::{ClientEvent, ServerEvent}; use serde::{Deserialize, Serialize}; +// ==================== 基础类型 ==================== + +/// 客户端信息 #[derive(Serialize, Deserialize, Debug, Clone)] pub struct ClientInfo { pub model: String, pub serial_number: String, } +// ==================== 控制包 ==================== + +/// 控制包 - TCP 通道传输的所有消息类型 #[derive(Serialize, Deserialize, Debug, Clone)] pub enum ControlPacket { - // Discovery - Discovery { - protocol: String, - port: u16, - }, + // ========== 服务发现 ========== + /// 服务发现广播 + Discovery { protocol: String, port: u16 }, - // Handshake + // ========== 握手 ========== + /// 服务端握手 ServerHello { auth: String, version: String, - udp_port: u16, // for audio + udp_port: u16, }, + /// 客户端握手 ClientHello { auth: String, version: String, - udp_port: u16, // for audio + udp_port: u16, info: ClientInfo, }, - // Heartbeat + // ========== 心跳 ========== Ping, Pong, - // RPC - RpcRequest { - id: u32, - method: String, - args: Vec, - }, - RpcResponse { - id: u32, - result: RpcResult, - }, + // ========== RPC ========== + /// RPC 请求（新版，使用 Command 类型） + RpcRequest { id: u32, command: Command }, + /// RPC 响应（新版，使用 CommandResult 类型） + RpcResponse { id: u32, result: CommandResult }, - // Audio Control - StartRecording { - config: AudioConfig, - }, + // ========== 事件 ========== + /// 服务端事件推送 + ServerEvent(ServerEvent), + /// 客户端事件推送 + ClientEvent(ClientEvent), + + // ========== 音频控制 ========== + /// 开始录音 + StartRecording { config: AudioConfig }, + /// 停止录音 StopRecording, - StartPlayback { - config: AudioConfig, - }, + /// 开始播放 + StartPlayback { config: AudioConfig }, + /// 停止播放 StopPlayback, + + // ========== 订阅管理 ========== + /// 订阅事件类型 + Subscribe { event_types: Vec }, + /// 取消订阅 + Unsubscribe { event_types: Vec }, } +// ==================== 音频包 ==================== + +/// 音频数据包 - UDP 通道传输 +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct AudioPacket { + /// Opus 编码的音频数据 + pub data: Vec, +} + +// ==================== 兼容性：旧版 RPC 结果 ==================== + +/// 旧版 RPC 结果（保持向后兼容） #[derive(Serialize, Deserialize, Debug, Clone, Default)] pub struct RpcResult { pub stdout: String, @@ -61,7 +92,12 @@ pub struct RpcResult { pub code: i32, } -#[derive(Serialize, Deserialize, Debug, Clone)] -pub struct AudioPacket { - pub data: Vec, +impl From for RpcResult { + fn from(resp: crate::net::command::ShellResponse) -> Self { + Self { + stdout: resp.stdout, + stderr: resp.stderr, + code: resp.exit_code, + } + } } diff --git a/packages/client-v2/src/net/rpc.rs b/packages/client-v2/src/net/rpc.rs index f4e56ce..9cc3ce8 100644 --- a/packages/client-v2/src/net/rpc.rs +++ b/packages/client-v2/src/net/rpc.rs @@ -1,32 +1,167 @@ -use crate::net::protocol::RpcResult; +//! # RPC Manager - RPC 调用管理 +//! +//! 管理 RPC 请求的生命周期，包括： +//! - 请求 ID 生成 +//! - 请求/响应匹配 +//! - 超时处理 + +use crate::net::command::CommandResult; use parking_lot::Mutex; use std::collections::HashMap; use std::sync::atomic::{AtomicU32, Ordering}; +use std::time::Duration; use tokio::sync::oneshot; +/// RPC 调用错误 +#[derive(Debug)] +pub enum RpcError { + /// 超时 + Timeout, + /// 通道关闭 + Cancelled, + /// 未连接 + NotConnected, +} + +impl std::fmt::Display for RpcError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + RpcError::Timeout => write!(f, "RPC timeout"), + RpcError::Cancelled => write!(f, "RPC cancelled"), + RpcError::NotConnected => write!(f, "Not connected"), + } + } +} + +impl std::error::Error for RpcError {} + +/// 等待中的 RPC 请求 +struct PendingRequest { + tx: oneshot::Sender, + created_at: std::time::Instant, +} + +/// RPC 管理器 pub struct RpcManager { next_id: AtomicU32, - pending: Mutex>>, + pending: Mutex>, + default_timeout: Duration, } impl RpcManager { + /// 创建新的 RPC 管理器 pub fn new() -> Self { + Self::with_timeout(Duration::from_secs(30)) + } + + /// 创建带自定义超时的 RPC 管理器 + pub fn with_timeout(timeout: Duration) -> Self { Self { next_id: AtomicU32::new(1), pending: Mutex::new(HashMap::new()), + default_timeout: timeout, } } - pub fn register(&self) -> (u32, oneshot::Receiver) { + /// 注册一个新的 RPC 请求 + /// + /// 返回 (请求ID, 响应接收器) + 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); + self.pending.lock().insert( + id, + PendingRequest { + tx, + created_at: std::time::Instant::now(), + }, + ); (id, rx) } - pub fn resolve(&self, id: u32, result: RpcResult) { - if let Some(tx) = self.pending.lock().remove(&id) { - let _ = tx.send(result); + /// 解析 RPC 响应 + pub fn resolve(&self, id: u32, result: CommandResult) { + if let Some(req) = self.pending.lock().remove(&id) { + let _ = req.tx.send(result); } } + + /// 取消指定的 RPC 请求 + pub fn cancel(&self, id: u32) { + self.pending.lock().remove(&id); + } + + /// 获取待处理请求数量 + pub fn pending_count(&self) -> usize { + self.pending.lock().len() + } + + /// 清理超时的请求 + pub fn cleanup_expired(&self) { + let mut pending = self.pending.lock(); + let now = std::time::Instant::now(); + + pending.retain(|_, req| { + if now.duration_since(req.created_at) > self.default_timeout { + false + } else { + true + } + }); + } + + /// 获取默认超时时间 + pub fn default_timeout(&self) -> Duration { + self.default_timeout + } +} + +impl Default for RpcManager { + fn default() -> Self { + Self::new() + } +} + +/// RPC 调用辅助函数 +pub async fn call_with_timeout( + timeout: Duration, + register_fn: F, +) -> Result +where + F: FnOnce() -> Fut, + Fut: std::future::Future, RpcError>>, +{ + let rx = register_fn().await?; + + match tokio::time::timeout(timeout, rx).await { + Ok(Ok(result)) => Ok(result), + Ok(Err(_)) => Err(RpcError::Cancelled), + Err(_) => Err(RpcError::Timeout), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::net::command::{CommandResult, ShellResponse}; + + #[tokio::test] + async fn test_rpc_manager() { + let mgr = RpcManager::new(); + + let (id, rx) = mgr.register(); + assert_eq!(mgr.pending_count(), 1); + + let result = CommandResult::Shell(ShellResponse { + stdout: "hello".to_string(), + stderr: String::new(), + exit_code: 0, + }); + + mgr.resolve(id, result.clone()); + assert_eq!(mgr.pending_count(), 0); + + let received = rx.await.unwrap(); + assert!(matches!(received, CommandResult::Shell(_))); + } }