From 6f550216512a413461711a3b4f483ba6d41ac1d5 Mon Sep 17 00:00:00 2001 From: Del Wang Date: Fri, 2 Jan 2026 15:16:31 +0800 Subject: [PATCH] refactor: handlers --- packages/client-v2/Makefile | 7 +- packages/client-v2/src/app/client/core.rs | 235 +++++------------- packages/client-v2/src/app/client/entry.rs | 2 +- packages/client-v2/src/app/client/handlers.rs | 202 +++++++++++++++ packages/client-v2/src/app/client/mod.rs | 2 + packages/client-v2/src/app/client/session.rs | 20 ++ packages/client-v2/src/app/server/core.rs | 87 ++----- packages/client-v2/src/app/server/handlers.rs | 50 ++++ packages/client-v2/src/app/server/mod.rs | 2 + packages/client-v2/src/app/server/session.rs | 23 ++ packages/client-v2/src/net/mod.rs | 1 + packages/client-v2/src/net/network.rs | 47 +++- packages/client-v2/src/net/rpc.rs | 32 +++ 13 files changed, 458 insertions(+), 252 deletions(-) create mode 100644 packages/client-v2/src/app/client/handlers.rs create mode 100644 packages/client-v2/src/app/client/session.rs create mode 100644 packages/client-v2/src/app/server/handlers.rs create mode 100644 packages/client-v2/src/app/server/session.rs create mode 100644 packages/client-v2/src/net/rpc.rs diff --git a/packages/client-v2/Makefile b/packages/client-v2/Makefile index 9c898b8..f8bc61e 100644 --- a/packages/client-v2/Makefile +++ b/packages/client-v2/Makefile @@ -1,13 +1,16 @@ build-server: - clear && cargo build --release --bin server --features app + cargo build --release --bin server --features app build-client: - clear && docker run --rm -v $(shell pwd):/app idootop/open-xiaoai-runtime:oh2p \ + docker run --rm -v $(shell pwd):/app idootop/open-xiaoai-runtime:oh2p \ cargo build --target armv7-unknown-linux-gnueabihf --release --bin client --features app run-server: target/release/server +test: + clear && make build-client && make deploy && make build-server && make run-server + # 部署到小爱音箱（调试自用） deploy: dd if=target/armv7-unknown-linux-gnueabihf/release/client \ diff --git a/packages/client-v2/src/app/client/core.rs b/packages/client-v2/src/app/client/core.rs index 2c8ef5d..aba1768 100644 --- a/packages/client-v2/src/app/client/core.rs +++ b/packages/client-v2/src/app/client/core.rs @@ -1,12 +1,11 @@ #![cfg(target_os = "linux")] -use crate::audio::codec::OpusCodec; +use crate::app::client::handlers; +use crate::app::client::session::ClientSession; use crate::audio::config::AudioConfig; -use crate::audio::player::AudioPlayer; -use crate::audio::recorder::AudioRecorder; use crate::net::discovery::Discovery; -use crate::net::network::{AudioSocket, ClientNetwork, ControlConnection}; -use crate::net::protocol::{AudioPacket, ControlPacket, DeviceInfo, RpcResult}; +use crate::net::network::{AudioSocket, ClientNetwork}; +use crate::net::protocol::{ControlPacket, DeviceInfo, RpcResult}; use anyhow::{Context, Result}; use std::net::SocketAddr; use std::sync::Arc; @@ -14,6 +13,7 @@ use tokio::sync::broadcast; pub struct Client { info: DeviceInfo, + session: Arc>>>, } impl Client { @@ -24,10 +24,11 @@ impl Client { mac: mac.to_string(), version, }, + session: Arc::new(tokio::sync::Mutex::new(None)), } } - pub async fn run(self) -> Result<()> { + pub async fn run(self: Arc) -> Result<()> { println!("正在寻找服务端..."); let (server_ip, server_port) = Discovery::discover_server().await?; let server_addr = SocketAddr::new(server_ip, server_port); @@ -48,177 +49,67 @@ impl Client { p => return Err(anyhow::anyhow!("认证失败: {:?}", p)), } + let (reader, writer) = control.split(); + let session = Arc::new(ClientSession::new(self.info.clone(), writer)); + *self.session.lock().await = Some(session.clone()); + let (stop_tx, _) = broadcast::channel::<()>(1); let audio_socket = Arc::new(AudioSocket::bind().await?); + // 启动测试 RPC 调用的任务 + let client_clone = self.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + println!("测试向服务端发起 RPC: status"); + match client_clone.call_server("status", vec![]).await { + Ok(res) => println!("收到服务端响应: {:?}", res), + Err(e) => eprintln!("向服务端发起 RPC 失败: {}", e), + } + }); + + let mut reader = reader; loop { - tokio::select! { - packet = control.recv_packet() => { - let packet = packet?; - match packet { - ControlPacket::StartRecording { config } => { - println!("开始录音: {:?}", config); - let socket = audio_socket.clone(); - let mut stop_rx = stop_tx.subscribe(); - let server_addr = control.peer_addr()?; - tokio::spawn(async move { - if let Err(e) = handle_recording(config, socket, server_addr, stop_rx).await { - eprintln!("录音出错: {}", e); - } - }); - } - ControlPacket::StopRecording => { - println!("停止录音"); - let _ = stop_tx.send(()); - } - ControlPacket::StartPlayback { config } => { - println!("开始播放: {:?}", config); - let socket = audio_socket.clone(); - let mut stop_rx = stop_tx.subscribe(); - tokio::spawn(async move { - if let Err(e) = handle_playback(config, socket, stop_rx).await { - eprintln!("播放出错: {}", e); - } - }); - } - ControlPacket::StopPlayback => { - println!("停止播放"); - let _ = stop_tx.send(()); - } - ControlPacket::RpcRequest { id, method, args } => { - println!("收到 RPC 请求: {} {:?}", method, args); - let result = handle_rpc(&method, args).await; - control.send_packet(&ControlPacket::RpcResponse { id, result }).await?; - } - ControlPacket::Ping => { - control.send_packet(&ControlPacket::Pong).await?; - } - _ => {} - } + let packet = reader.recv_packet().await?; + let self_clone = self.clone(); + let session_clone = session.clone(); + let audio_socket = audio_socket.clone(); + let stop_tx = stop_tx.clone(); + + tokio::spawn(async move { + if let Err(e) = handlers::handle_packet( + self_clone, + session_clone, + packet, + audio_socket, + stop_tx, + server_addr, + ) + .await + { + eprintln!("处理控制包出错: {}", e); } - } - } - } -} - -async fn handle_recording( - config: AudioConfig, - socket: Arc, - server_addr: SocketAddr, - mut stop_rx: broadcast::Receiver<()>, -) -> Result<()> { - #[cfg(target_os = "linux")] - { - let recorder = AudioRecorder::new(&config)?; - let mut codec = OpusCodec::new(&config)?; - let mut pcm_buf = vec![0i16; config.frame_size]; - let mut opus_buf = vec![0u8; 4096]; - - loop { - if stop_rx.try_recv().is_ok() { - break; - } - let n = recorder.read(&mut pcm_buf)?; - if n > 0 { - let opus_len = codec.encode(&pcm_buf[..n], &mut opus_buf)?; - let packet = AudioPacket { - data: opus_buf[..opus_len].to_vec(), - }; - socket.send_packet(&packet, server_addr).await?; - } - } - } - #[cfg(not(target_os = "linux"))] - { - println!("当前系统不支持 ALSA 录音，模拟发送音频数据..."); - loop { - if stop_rx.try_recv().is_ok() { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - let packet = AudioPacket { - data: vec![0u8; 10], - }; - socket.send_packet(&packet, server_addr).await?; - } - } - Ok(()) -} - -async fn handle_playback( - config: AudioConfig, - socket: Arc, - mut stop_rx: broadcast::Receiver<()>, -) -> Result<()> { - #[cfg(target_os = "linux")] - { - let player = AudioPlayer::new(&config)?; - let mut codec = OpusCodec::new(&config)?; - let mut pcm_buf = vec![0i16; config.frame_size]; - let mut udp_buf = vec![0u8; 4096]; - - loop { - if stop_rx.try_recv().is_ok() { - break; - } - // 这里简单处理，UDP 接收可能阻塞。实际建议加超时或 select - let (packet, _) = socket.recv_packet(&mut udp_buf).await?; - let pcm_len = codec.decode(&packet.data, &mut pcm_buf)?; - player.write(&pcm_buf[..pcm_len])?; - } - } - #[cfg(not(target_os = "linux"))] - { - println!("当前系统不支持 ALSA 播放，模拟接收音频数据..."); - let mut udp_buf = vec![0u8; 4096]; - loop { - tokio::select! { - _ = stop_rx.recv() => break, - res = socket.recv_packet(&mut udp_buf) => { - let _ = res?; - } - } - } - } - Ok(()) -} - -async fn handle_rpc(method: &str, args: Vec) -> RpcResult { - if method == "shell" && !args.is_empty() { - let cmd = &args[0]; - println!("执行 Shell: {}", cmd); - - // 模拟执行 - #[cfg(target_os = "linux")] - { - use std::process::Command; - let output = Command::new("sh").arg("-c").arg(cmd).output(); - match output { - Ok(out) => RpcResult { - stdout: String::from_utf8_lossy(&out.stdout).to_string(), - stderr: String::from_utf8_lossy(&out.stderr).to_string(), - code: out.status.code().unwrap_or(0), - }, - Err(e) => RpcResult { - stdout: "".to_string(), - stderr: e.to_string(), - code: -1, - }, - } - } - #[cfg(not(target_os = "linux"))] - { - RpcResult { - stdout: format!("模拟执行 {} 成功", cmd), - stderr: "".to_string(), - code: 0, - } - } - } else { - RpcResult { - stdout: "".to_string(), - stderr: "未知方法或参数错误".to_string(), - code: -1, + }); } } + + pub async fn call_server(&self, method: &str, args: Vec) -> Result { + let session = self + .session + .lock() + .await + .clone() + .context("Session not established")?; + let (id, rx) = session.rpc.alloc_id(); + session + .writer + .lock() + .await + .send_packet(&ControlPacket::RpcRequest { + id, + method: method.to_string(), + args, + }) + .await?; + Ok(rx.await?) + } } diff --git a/packages/client-v2/src/app/client/entry.rs b/packages/client-v2/src/app/client/entry.rs index 79ee323..409c849 100644 --- a/packages/client-v2/src/app/client/entry.rs +++ b/packages/client-v2/src/app/client/entry.rs @@ -9,6 +9,6 @@ pub async fn run_client() -> Result<()> { let mac = "00:11:22:33:44:55"; let version = 1; - let client = Client::new(model, mac, version); + let client = std::sync::Arc::new(Client::new(model, mac, version)); client.run().await } diff --git a/packages/client-v2/src/app/client/handlers.rs b/packages/client-v2/src/app/client/handlers.rs new file mode 100644 index 0000000..58a82a3 --- /dev/null +++ b/packages/client-v2/src/app/client/handlers.rs @@ -0,0 +1,202 @@ +#![cfg(target_os = "linux")] + +use crate::app::client::core::Client; +use crate::app::client::session::ClientSession; +use crate::audio::codec::OpusCodec; +use crate::audio::config::AudioConfig; +use crate::audio::player::AudioPlayer; +use crate::audio::recorder::AudioRecorder; +use crate::net::network::AudioSocket; +use crate::net::protocol::{AudioPacket, ControlPacket, RpcResult}; +use anyhow::Result; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::sync::broadcast; + +pub async fn handle_packet( + client: Arc, + session: Arc, + packet: ControlPacket, + audio_socket: Arc, + stop_tx: broadcast::Sender<()>, + server_addr: SocketAddr, +) -> Result<()> { + match packet { + ControlPacket::RpcResponse { id, result } => { + session.rpc.fulfill(id, result); + } + ControlPacket::RpcRequest { id, method, args } => { + println!("收到 RPC 请求: {} {:?}", method, args); + let result = handle_rpc(&method, args).await; + let response = ControlPacket::RpcResponse { id, result }; + session.writer.lock().await.send_packet(&response).await?; + } + ControlPacket::StartRecording { config } => { + // ... + println!("开始录音: {:?}", config); + let mut stop_rx = stop_tx.subscribe(); + tokio::spawn(async move { + if let Err(e) = handle_recording(config, audio_socket, server_addr, stop_rx).await { + eprintln!("录音出错: {}", e); + } + }); + } + ControlPacket::StopRecording => { + println!("停止录音"); + let _ = stop_tx.send(()); + } + ControlPacket::StartPlayback { config } => { + println!("开始播放: {:?}", config); + let mut stop_rx = stop_tx.subscribe(); + tokio::spawn(async move { + if let Err(e) = handle_playback(config, audio_socket, stop_rx).await { + eprintln!("播放出错: {}", e); + } + }); + } + ControlPacket::StopPlayback => { + println!("停止播放"); + let _ = stop_tx.send(()); + } + ControlPacket::Ping => { + let _ = session + .writer + .lock() + .await + .send_packet(&ControlPacket::Pong) + .await; + } + _ => {} + } + Ok(()) +} + +pub async fn handle_rpc(method: &str, args: Vec) -> RpcResult { + match method { + "shell" => handle_shell(args).await, + _ => RpcResult { + stdout: "".to_string(), + stderr: format!("Unknown method: {}", method), + code: -1, + }, + } +} + +async fn handle_shell(args: Vec) -> RpcResult { + if args.is_empty() { + return RpcResult { + stdout: "".to_string(), + stderr: "Missing command argument".to_string(), + code: -1, + }; + } + let cmd = &args[0]; + println!("Executing Shell: {}", cmd); + + #[cfg(target_os = "linux")] + { + use std::process::Command; + let output = Command::new("sh").arg("-c").arg(cmd).output(); + match output { + Ok(out) => RpcResult { + stdout: String::from_utf8_lossy(&out.stdout).to_string(), + stderr: String::from_utf8_lossy(&out.stderr).to_string(), + code: out.status.code().unwrap_or(0), + }, + Err(e) => RpcResult { + stdout: "".to_string(), + stderr: e.to_string(), + code: -1, + }, + } + } + #[cfg(not(target_os = "linux"))] + { + RpcResult { + stdout: format!("Mock execution of {} successful", cmd), + stderr: "".to_string(), + code: 0, + } + } +} + +async fn handle_recording( + config: AudioConfig, + socket: Arc, + server_addr: SocketAddr, + mut stop_rx: broadcast::Receiver<()>, +) -> Result<()> { + #[cfg(target_os = "linux")] + { + let recorder = AudioRecorder::new(&config)?; + let mut codec = OpusCodec::new(&config)?; + let mut pcm_buf = vec![0i16; config.frame_size]; + let mut opus_buf = vec![0u8; 4096]; + + loop { + if stop_rx.try_recv().is_ok() { + break; + } + let n = recorder.read(&mut pcm_buf)?; + if n > 0 { + let opus_len = codec.encode(&pcm_buf[..n], &mut opus_buf)?; + let packet = AudioPacket { + data: opus_buf[..opus_len].to_vec(), + }; + socket.send_packet(&packet, server_addr).await?; + } + } + } + #[cfg(not(target_os = "linux"))] + { + println!("当前系统不支持 ALSA 录音，模拟发送音频数据..."); + loop { + if stop_rx.try_recv().is_ok() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + let packet = AudioPacket { + data: vec![0u8; 10], + }; + socket.send_packet(&packet, server_addr).await?; + } + } + Ok(()) +} + +async fn handle_playback( + config: AudioConfig, + socket: Arc, + mut stop_rx: broadcast::Receiver<()>, +) -> Result<()> { + #[cfg(target_os = "linux")] + { + let player = AudioPlayer::new(&config)?; + let mut codec = OpusCodec::new(&config)?; + let mut pcm_buf = vec![0i16; config.frame_size]; + let mut udp_buf = vec![0u8; 4096]; + + loop { + if stop_rx.try_recv().is_ok() { + break; + } + let (packet, _) = socket.recv_packet(&mut udp_buf).await?; + let pcm_len = codec.decode(&packet.data, &mut pcm_buf)?; + player.write(&pcm_buf[..pcm_len])?; + } + } + #[cfg(not(target_os = "linux"))] + { + println!("当前系统不支持 ALSA 播放，模拟接收音频数据..."); + let mut udp_buf = vec![0u8; 4096]; + loop { + tokio::select! { + _ = stop_rx.recv() => break, + res = socket.recv_packet(&mut udp_buf) => { + let _ = res?; + } + } + } + } + Ok(()) +} diff --git a/packages/client-v2/src/app/client/mod.rs b/packages/client-v2/src/app/client/mod.rs index 1ca5324..cc15fe5 100644 --- a/packages/client-v2/src/app/client/mod.rs +++ b/packages/client-v2/src/app/client/mod.rs @@ -1,2 +1,4 @@ pub mod core; pub mod entry; +pub mod handlers; +pub mod session; 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..ee85af6 --- /dev/null +++ b/packages/client-v2/src/app/client/session.rs @@ -0,0 +1,20 @@ +use crate::net::network::ControlWriter; +use crate::net::protocol::DeviceInfo; +use crate::net::rpc::RpcManager; +use std::sync::Arc; + +pub struct ClientSession { + pub info: DeviceInfo, + pub writer: Arc>, + pub rpc: Arc, +} + +impl ClientSession { + pub fn new(info: DeviceInfo, writer: ControlWriter) -> Self { + Self { + info, + writer: Arc::new(tokio::sync::Mutex::new(writer)), + rpc: Arc::new(RpcManager::new()), + } + } +} diff --git a/packages/client-v2/src/app/server/core.rs b/packages/client-v2/src/app/server/core.rs index 04674b3..2552c79 100644 --- a/packages/client-v2/src/app/server/core.rs +++ b/packages/client-v2/src/app/server/core.rs @@ -1,53 +1,19 @@ +use crate::app::server::handlers; +use crate::app::server::session::ServerSession; use crate::audio::codec::OpusCodec; use crate::audio::config::AudioConfig; use crate::audio::wav::{WavReader, WavWriter}; use crate::net::discovery::Discovery; use crate::net::network::{AudioSocket, ControlConnection, ServerNetwork}; -use crate::net::protocol::{AudioPacket, ControlPacket, DeviceInfo, RpcResult}; +use crate::net::protocol::{AudioPacket, ControlPacket, RpcResult}; use anyhow::{Context, Result}; use parking_lot::Mutex; use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; -use std::sync::atomic::{AtomicU32, Ordering}; -use tokio::sync::{broadcast, oneshot}; - -pub struct RpcManager { - next_id: AtomicU32, - pending: Mutex>>, -} - -impl RpcManager { - pub fn new() -> Self { - Self { - next_id: AtomicU32::new(1), - pending: Mutex::new(HashMap::new()), - } - } - - pub fn alloc_id(&self) -> (u32, oneshot::Receiver) { - let id = self.next_id.fetch_add(1, Ordering::SeqCst); - let (tx, rx) = oneshot::channel(); - self.pending.lock().insert(id, tx); - (id, rx) - } - - pub fn fulfill(&self, id: u32, result: RpcResult) { - if let Some(tx) = self.pending.lock().remove(&id) { - let _ = tx.send(result); - } - } -} - -pub struct Session { - pub info: DeviceInfo, - pub control: Arc>, - pub addr: SocketAddr, - pub rpc: Arc, -} pub struct Server { - sessions: Arc>>>, + sessions: Arc>>>, audio_socket: Arc, } @@ -98,36 +64,20 @@ impl Server { ); control.send_packet(&ControlPacket::IdentifyOk).await?; - let rpc = Arc::new(RpcManager::new()); - let control = Arc::new(tokio::sync::Mutex::new(control)); - let session = Arc::new(Session { - info, - control: control.clone(), - addr, - rpc: rpc.clone(), - }); + let (mut reader, writer) = control.split(); + let session = Arc::new(ServerSession::new(info, writer, addr)); self.sessions.lock().insert(addr, session.clone()); // 处理控制消息循环 loop { - let mut ctrl = control.lock().await; - let packet = ctrl.recv_packet().await?; - drop(ctrl); // 释放锁以允许发送 - - match packet { - ControlPacket::RpcResponse { id, result } => { - rpc.fulfill(id, result); + let packet = reader.recv_packet().await?; + let session = session.clone(); + tokio::spawn(async move { + if let Err(e) = handlers::handle_packet(session, packet).await { + eprintln!("处理来自 {} 的包时出错: {}", addr, e); } - ControlPacket::Ping => { - control - .lock() - .await - .send_packet(&ControlPacket::Pong) - .await?; - } - _ => {} - } + }); } } @@ -141,7 +91,7 @@ impl Server { .context("未找到 Session")?; let (id, rx) = session.rpc.alloc_id(); session - .control + .writer .lock() .await .send_packet(&ControlPacket::RpcRequest { @@ -163,7 +113,7 @@ impl Server { // 发送开始录音指令 session - .control + .writer .lock() .await .send_packet(&ControlPacket::StartRecording { @@ -189,7 +139,7 @@ impl Server { .cloned() .context("未找到 Session")?; session - .control + .writer .lock() .await .send_packet(&ControlPacket::StopRecording) @@ -206,7 +156,7 @@ impl Server { .context("未找到 Session")?; session - .control + .writer .lock() .await .send_packet(&ControlPacket::StartPlayback { @@ -233,7 +183,7 @@ impl Server { .cloned() .context("未找到 Session")?; session - .control + .writer .lock() .await .send_packet(&ControlPacket::StopPlayback) @@ -259,8 +209,6 @@ async fn save_audio_to_wav( println!("正在录制到 {}...", path); - // 这里需要一个停止机制，目前简单起见，如果一段时间没收到包就停止，或者通过全局状态 - // 为简单演示，我们录制 100 个包 for _ in 0..100 { let (packet, _) = socket.recv_packet(&mut udp_buf).await?; let pcm_len = codec.decode(&packet.data, &mut pcm_buf)?; @@ -296,7 +244,6 @@ async fn stream_wav_to_client( }; socket.send_packet(&packet, target).await?; - // 控制发送频率，约 20ms 一帧 tokio::time::sleep(std::time::Duration::from_millis(20)).await; } diff --git a/packages/client-v2/src/app/server/handlers.rs b/packages/client-v2/src/app/server/handlers.rs new file mode 100644 index 0000000..c84f3f3 --- /dev/null +++ b/packages/client-v2/src/app/server/handlers.rs @@ -0,0 +1,50 @@ +use crate::app::server::session::ServerSession; +use crate::net::protocol::{ControlPacket, RpcResult}; +use anyhow::Result; +use std::sync::Arc; + +pub async fn handle_packet(session: Arc, packet: ControlPacket) -> Result<()> { + match packet { + ControlPacket::RpcResponse { id, result } => { + session.rpc.fulfill(id, result); + } + ControlPacket::RpcRequest { id, method, args } => { + println!( + "收到来自客户端 {} 的 RPC 请求: {} {:?}", + session.addr, method, args + ); + let result = handle_server_rpc(&method, args).await; + session + .writer + .lock() + .await + .send_packet(&ControlPacket::RpcResponse { id, result }) + .await?; + } + ControlPacket::Ping => { + session + .writer + .lock() + .await + .send_packet(&ControlPacket::Pong) + .await?; + } + _ => {} + } + Ok(()) +} + +async fn handle_server_rpc(method: &str, _args: Vec) -> RpcResult { + match method { + "status" => RpcResult { + stdout: "Server is running normally".to_string(), + stderr: "".to_string(), + code: 0, + }, + _ => RpcResult { + stdout: "".to_string(), + stderr: format!("Server does not support method: {}", method), + code: -1, + }, + } +} diff --git a/packages/client-v2/src/app/server/mod.rs b/packages/client-v2/src/app/server/mod.rs index 1ca5324..cc15fe5 100644 --- a/packages/client-v2/src/app/server/mod.rs +++ b/packages/client-v2/src/app/server/mod.rs @@ -1,2 +1,4 @@ pub mod core; pub mod entry; +pub mod handlers; +pub mod session; 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..d3e498c --- /dev/null +++ b/packages/client-v2/src/app/server/session.rs @@ -0,0 +1,23 @@ +use crate::net::network::ControlWriter; +use crate::net::protocol::DeviceInfo; +use crate::net::rpc::RpcManager; +use std::net::SocketAddr; +use std::sync::Arc; + +pub struct ServerSession { + pub info: DeviceInfo, + pub writer: Arc>, + pub addr: SocketAddr, + pub rpc: Arc, +} + +impl ServerSession { + pub fn new(info: DeviceInfo, writer: ControlWriter, addr: SocketAddr) -> Self { + Self { + info, + writer: Arc::new(tokio::sync::Mutex::new(writer)), + addr, + rpc: Arc::new(RpcManager::new()), + } + } +} diff --git a/packages/client-v2/src/net/mod.rs b/packages/client-v2/src/net/mod.rs index 3bba08b..af1fd80 100644 --- a/packages/client-v2/src/net/mod.rs +++ b/packages/client-v2/src/net/mod.rs @@ -1,3 +1,4 @@ pub mod discovery; pub mod network; pub mod protocol; +pub mod rpc; diff --git a/packages/client-v2/src/net/network.rs b/packages/client-v2/src/net/network.rs index 3040ab4..758fe1d 100644 --- a/packages/client-v2/src/net/network.rs +++ b/packages/client-v2/src/net/network.rs @@ -3,9 +3,11 @@ use anyhow::{Context, Result}; use std::net::SocketAddr; use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; use tokio::net::{TcpListener, TcpStream, UdpSocket}; /// UDP 音频传输 +// ... (rest of AudioSocket is same) pub struct AudioSocket { socket: Arc, } @@ -39,6 +41,40 @@ impl AudioSocket { } } +/// TCP 控制连接读取端 +pub struct ControlReader { + reader: OwnedReadHalf, +} + +impl ControlReader { + pub async fn recv_packet(&mut self) -> Result { + let len = self.reader.read_u32().await? as usize; + if len > 10 * 1024 * 1024 { + return Err(anyhow::anyhow!("Packet too large: {}", len)); + } + let mut buf = vec![0u8; len]; + self.reader.read_exact(&mut buf).await?; + let packet = postcard::from_bytes(&buf)?; + Ok(packet) + } +} + +/// TCP 控制连接写入端 +pub struct ControlWriter { + writer: OwnedWriteHalf, +} + +impl ControlWriter { + pub async fn send_packet(&mut self, packet: &ControlPacket) -> Result<()> { + let bytes = postcard::to_allocvec(packet)?; + let len = bytes.len() as u32; + self.writer.write_u32(len).await?; + self.writer.write_all(&bytes).await?; + self.writer.flush().await?; + Ok(()) + } +} + /// TCP 控制连接 pub struct ControlConnection { stream: TcpStream, @@ -54,6 +90,7 @@ impl ControlConnection { let len = bytes.len() as u32; self.stream.write_u32(len).await?; self.stream.write_all(&bytes).await?; + self.stream.flush().await?; Ok(()) } @@ -68,13 +105,9 @@ impl ControlConnection { Ok(packet) } - pub fn split( - self, - ) -> ( - tokio::net::tcp::OwnedReadHalf, - tokio::net::tcp::OwnedWriteHalf, - ) { - self.stream.into_split() + pub fn split(self) -> (ControlReader, ControlWriter) { + let (r, w) = self.stream.into_split(); + (ControlReader { reader: r }, ControlWriter { writer: w }) } pub fn peer_addr(&self) -> Result { diff --git a/packages/client-v2/src/net/rpc.rs b/packages/client-v2/src/net/rpc.rs new file mode 100644 index 0000000..3b1c4fd --- /dev/null +++ b/packages/client-v2/src/net/rpc.rs @@ -0,0 +1,32 @@ +use crate::net::protocol::RpcResult; +use parking_lot::Mutex; +use std::collections::HashMap; +use std::sync::atomic::{AtomicU32, Ordering}; +use tokio::sync::oneshot; + +pub struct RpcManager { + next_id: AtomicU32, + pending: Mutex>>, +} + +impl RpcManager { + pub fn new() -> Self { + Self { + next_id: AtomicU32::new(1), + pending: Mutex::new(HashMap::new()), + } + } + + pub fn alloc_id(&self) -> (u32, oneshot::Receiver) { + let id = self.next_id.fetch_add(1, Ordering::SeqCst); + let (tx, rx) = oneshot::channel(); + self.pending.lock().insert(id, tx); + (id, rx) + } + + pub fn fulfill(&self, id: u32, result: RpcResult) { + if let Some(tx) = self.pending.lock().remove(&id) { + let _ = tx.send(result); + } + } +}