diff --git a/packages/client-v2/src/app/client/mod.rs b/packages/client-v2/src/app/client/mod.rs index d08f621..5587dd8 100644 --- a/packages/client-v2/src/app/client/mod.rs +++ b/packages/client-v2/src/app/client/mod.rs @@ -36,7 +36,7 @@ use crate::net::command::{ ShellResponse, }; use crate::net::discovery::Discovery; -use crate::net::event::{ClientEvent, NotificationLevel, ServerEvent}; +use crate::net::event::{EventBus, EventBusSubscription, EventData}; use crate::net::network::{AudioSocket, Connection}; use crate::net::protocol::ControlPacket; use crate::net::sync::now_us; @@ -44,7 +44,7 @@ use anyhow::{Result, anyhow}; use session::handshake; use std::net::SocketAddr; use std::sync::Arc; -use tokio::sync::{RwLock, broadcast}; +use tokio::sync::RwLock; use tokio_util::sync::CancellationToken; /// 客户端配置 @@ -91,19 +91,18 @@ pub struct Client { session: RwLock>>, /// 全局取消令牌 cancel: CancellationToken, - /// 服务端事件广播 - server_events: broadcast::Sender, + /// 事件总线 + event_bus: Arc, } impl Client { /// 创建新客户端 pub fn new(config: ClientConfig) -> Self { - let (server_events, _) = broadcast::channel(64); Self { config, session: RwLock::new(None), cancel: CancellationToken::new(), - server_events, + event_bus: Arc::new(EventBus::default()), } } @@ -112,9 +111,9 @@ impl Client { Self::new(ClientConfig::default()) } - /// 订阅服务端事件 - pub fn subscribe_events(&self) -> broadcast::Receiver { - self.server_events.subscribe() + /// 订阅事件 + pub fn subscribe_events(&self) -> EventBusSubscription { + self.event_bus.subscribe() } /// 运行客户端(自动发现服务器并连接) @@ -249,18 +248,19 @@ impl Client { let t4 = now_us(); session.update_clock(client_ts, server_ts, t4); } + ControlPacket::Event { timestamp, data } => { + self.event_bus.receive(data, timestamp, session.id()); + } ControlPacket::RpcResponse { id, result } => { session.resolve_rpc(id, result); } + // todo 耗时任务,异步响应 ControlPacket::RpcRequest { id, command } => { let result = self.handle_command(session, command).await; session .send(&ControlPacket::RpcResponse { id, result }) .await?; } - ControlPacket::ServerEvent(event) => { - self.handle_server_event(session, event).await; - } ControlPacket::StartRecording { config } => { println!("[Client] Starting recording..."); let handle = RecordPipeline::spawn( @@ -360,39 +360,6 @@ impl Client { }) } - /// 处理服务端事件 - 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; @@ -424,25 +391,21 @@ impl Client { } } - /// 发送客户端事件 - pub async fn send_event(&self, event: ClientEvent) -> Result<()> { + /// 发送事件 + pub async fn send_event(&self, event: EventData) -> Result<()> { let session_guard = self.session.read().await; if let Some(session) = session_guard.as_ref() { - session.send(&ControlPacket::ClientEvent(event)).await + session + .send(&ControlPacket::Event { + timestamp: now_us(), + data: 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() diff --git a/packages/client-v2/src/app/client/session.rs b/packages/client-v2/src/app/client/session.rs index 2e937c5..040712c 100644 --- a/packages/client-v2/src/app/client/session.rs +++ b/packages/client-v2/src/app/client/session.rs @@ -129,6 +129,11 @@ impl Session { } } + /// 获取会话 ID(使用 TCP 地址) + pub fn id(&self) -> SocketAddr { + self.conn.peer_addr() + } + /// 检查会话是否仍然有效 pub fn is_alive(&self) -> bool { !self.cancel.is_cancelled() diff --git a/packages/client-v2/src/app/server/mod.rs b/packages/client-v2/src/app/server/mod.rs index aa6a584..6456b4d 100644 --- a/packages/client-v2/src/app/server/mod.rs +++ b/packages/client-v2/src/app/server/mod.rs @@ -48,7 +48,7 @@ use crate::net::command::{ AudioState, Command, CommandError, CommandResult, DeviceInfo, ShellResponse, }; use crate::net::discovery::Discovery; -use crate::net::event::{ClientEvent, ServerEvent, ServerEventBus}; +use crate::net::event::{EventBus, EventBusSubscription, EventData}; use crate::net::network::Connection; use crate::net::protocol::ControlPacket; use crate::net::sync::now_us; @@ -92,7 +92,7 @@ pub struct Server { /// 音频总线 audio_bus: Arc, /// 服务端事件总线 - event_bus: Arc, + event_bus: Arc, /// 服务器取消令牌 cancel: CancellationToken, /// 服务器启动时间 @@ -108,15 +108,15 @@ impl Server { config, sessions: Arc::new(SessionManager::new()), audio_bus, - event_bus: Arc::new(ServerEventBus::default()), + event_bus: Arc::new(EventBus::default()), cancel: CancellationToken::new(), started_at: std::time::Instant::now(), }) } - /// 获取事件总线(用于外部订阅) - pub fn event_bus(&self) -> Arc { - self.event_bus.clone() + /// 订阅事件 + pub fn subscribe_events(&self) -> EventBusSubscription { + self.event_bus.subscribe() } /// 启动服务器 @@ -199,23 +199,6 @@ impl Server { 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; @@ -223,20 +206,6 @@ impl Server { 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 } @@ -314,18 +283,20 @@ impl Server { }; session.send(&pong).await?; } + ControlPacket::Event { timestamp, data } => { + self.event_bus.receive(data, timestamp, session.id()); + } ControlPacket::RpcResponse { id, result } => { session.resolve_rpc(id, result); } + // todo 耗时任务,异步响应 ControlPacket::RpcRequest { id, command } => { let result = self.handle_command(session, command).await; session .send(&ControlPacket::RpcResponse { id, result }) .await?; } - ControlPacket::ClientEvent(event) => { - self.handle_client_event(session, event).await; - } + _ => {} } Ok(()) @@ -376,30 +347,6 @@ impl Server { } } - /// 处理客户端事件 - 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 ==================== /// 获取所有已连接的客户端地址 @@ -433,19 +380,16 @@ impl Server { } } - /// 推送事件给所有客户端 - 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<()> { + /// 发送事件 + pub async fn send_event(&self, addr: SocketAddr, event: EventData) -> 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 + session + .send(&ControlPacket::Event { + timestamp: now_us(), + data: event, + }) + .await?; + Ok(()) } /// 开始录音 @@ -473,16 +417,6 @@ impl Server { .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(()) } @@ -493,16 +427,6 @@ impl Server { session.stop_recording(); session.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(()) } @@ -530,16 +454,6 @@ impl Server { session.start_playback(handle); - // 发布事件 - session - .send(&ControlPacket::ServerEvent( - ServerEvent::AudioStatusChanged { - is_recording: session.is_recording(), - is_playing: true, - }, - )) - .await?; - println!("[Server] Playback started for {} from {}", addr, file_path); Ok(()) } @@ -550,16 +464,6 @@ impl Server { session.stop_playback(); session.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(()) } diff --git a/packages/client-v2/src/bin/client.rs b/packages/client-v2/src/bin/client.rs index 3dcb27f..f1471d9 100644 --- a/packages/client-v2/src/bin/client.rs +++ b/packages/client-v2/src/bin/client.rs @@ -9,7 +9,7 @@ use std::sync::Arc; use xiao::app::client::{Client, ClientConfig}; use xiao::net::command::Command; -use xiao::net::event::NotificationLevel; +use xiao::net::event::EventData; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -29,8 +29,13 @@ async fn main() -> anyhow::Result<()> { let event_client = client.clone(); tokio::spawn(async move { let mut rx = event_client.subscribe_events(); - while let Ok(event) = rx.recv().await { - println!("📨 [ServerEvent] {:?}", event); + while let Some((event, ts, addr)) = rx.recv().await { + match event { + EventData::Hello { message, .. } => { + println!("[Event] Hello: {} ts:{} addr:{}", message, ts, addr); + } + _ => {} + } } }); @@ -71,13 +76,15 @@ async fn main() -> anyhow::Result<()> { } // 3. 发送客户端事件 - println!("\n3️⃣ Sending alert event to server..."); + println!("\n3️⃣ Sending event to server..."); match client - .send_alert(NotificationLevel::Info, "Client started successfully!") + .send_event(EventData::Hello { + message: "from client!".to_string(), + }) .await { - Ok(_) => println!(" ✅ Alert sent"), - Err(e) => println!(" ❌ Failed to send alert: {}", e), + Ok(_) => println!(" ✅ Event sent"), + Err(e) => println!(" ❌ Failed to send event: {}", e), } println!("\n═══════════════════════════════════════════════════════"); diff --git a/packages/client-v2/src/bin/server.rs b/packages/client-v2/src/bin/server.rs index d0a9cfe..d02aee3 100644 --- a/packages/client-v2/src/bin/server.rs +++ b/packages/client-v2/src/bin/server.rs @@ -10,7 +10,7 @@ use std::sync::Arc; use xiao::app::server::{Server, ServerConfig}; use xiao::audio::config::AudioConfig; use xiao::net::command::Command; -use xiao::net::event::{NotificationLevel, ServerEvent}; +use xiao::net::event::EventData; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -37,14 +37,11 @@ async fn main() -> anyhow::Result<()> { // 启动事件监听器 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 { + let mut rx = event_server.subscribe_events(); + while let Some((event, ts, addr)) = rx.recv().await { match event { - ServerEvent::ClientJoined { model, .. } => { - println!("📱 [Event] Client joined: {} ({:?})", model, addr); - } - ServerEvent::ClientLeft { model, .. } => { - println!("📴 [Event] Client left: {} ({:?})", model, addr); + EventData::Hello { message, .. } => { + println!("[Event] Hello: {} ts:{} addr:{}", message, ts, addr); } _ => {} } @@ -92,16 +89,20 @@ async fn main() -> anyhow::Result<()> { Err(e) => println!(" ❌ Shell failed: {}", e), } - // 4. 测试事件广播 + // 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"); + match server + .send_event( + addr, + EventData::Hello { + message: "from server!".to_string(), + }, + ) + .await + { + Ok(_) => println!(" ✅ Event sent"), + Err(e) => println!(" ❌ Failed to send event: {}", e), + } // 5. 测试音频录制 println!("\n5️⃣ Testing Audio Recording (10 seconds)..."); diff --git a/packages/client-v2/src/net/event.rs b/packages/client-v2/src/net/event.rs index 1cc5cec..0111a2a 100644 --- a/packages/client-v2/src/net/event.rs +++ b/packages/client-v2/src/net/event.rs @@ -30,94 +30,15 @@ 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 enum EventData { + /// 测试事件 + Hello { message: String }, } // ==================== 事件总线 ==================== @@ -141,125 +62,34 @@ impl EventSubscription { } } -/// 服务端事件总线 -pub struct ServerEventBus { - tx: broadcast::Sender<(Option, ServerEvent)>, +/// 事件总线(单向,只接收转发) +pub struct EventBus { + tx: broadcast::Sender<(EventData, u128, SocketAddr)>, } -impl ServerEventBus { +pub type EventBusSubscription = EventSubscription<(EventData, u128, SocketAddr)>; + +impl EventBus { 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 receive(&self, event: EventData, timestamp: u128, addr: SocketAddr) { + let _ = self.tx.send((event, timestamp, addr)); } /// 订阅事件 - pub fn subscribe(&self) -> EventSubscription<(Option, ServerEvent)> { + pub fn subscribe(&self) -> EventBusSubscription { EventSubscription { receiver: self.tx.subscribe(), } } } -impl Default for ServerEventBus { +impl Default for EventBus { 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; + Self::new(128) } } diff --git a/packages/client-v2/src/net/protocol.rs b/packages/client-v2/src/net/protocol.rs index d1c9268..749818d 100644 --- a/packages/client-v2/src/net/protocol.rs +++ b/packages/client-v2/src/net/protocol.rs @@ -4,7 +4,7 @@ use crate::audio::config::AudioConfig; use crate::net::command::{Command, CommandResult}; -use crate::net::event::{ClientEvent, ServerEvent}; +use crate::net::event::EventData; use serde::{Deserialize, Serialize}; // ==================== 基础类型 ==================== @@ -63,31 +63,20 @@ pub enum ControlPacket { }, // ========== RPC ========== - /// RPC 请求(新版,使用 Command 类型) RpcRequest { id: u32, command: Command, }, - /// RPC 响应(新版,使用 CommandResult 类型) RpcResponse { id: u32, result: CommandResult, }, // ========== 事件 ========== - /// 服务端事件推送 - ServerEvent(ServerEvent), - /// 客户端事件推送 - ClientEvent(ClientEvent), - - // ========== 订阅管理 ========== - /// 订阅事件类型 - Subscribe { - event_types: Vec, - }, - /// 取消订阅 - Unsubscribe { - event_types: Vec, + Event { + /// 微秒时间戳 + timestamp: u128, + data: EventData, }, // ========== 音频控制 ========== diff --git a/packages/client-v2/src/net/sync.rs b/packages/client-v2/src/net/sync.rs index 4f33cc0..5d8bdfc 100644 --- a/packages/client-v2/src/net/sync.rs +++ b/packages/client-v2/src/net/sync.rs @@ -35,7 +35,6 @@ pub struct ClockSync { struct OffsetSample { offset: i128, rtt: i128, - timestamp: u128, } #[derive(Clone, Copy)] @@ -96,7 +95,6 @@ impl ClockSync { let sample = OffsetSample { offset, rtt, - timestamp: client_recv_ts, }; self.offsets.push_back(sample); if self.offsets.len() > self.window_size {