Hass/open-xiaoai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 事件收发与处理

Browse Source
This commit is contained in:
Del Wang
2026-01-06 18:57:14 +08:00
parent 487a5b8c14
commit fb6c1fd9f8
8 changed files with 97 additions and 400 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/client-v2/src/app/client/mod.rs
+20 -57
View File
@@ -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<Option<Arc<Session>>>,
/// 全局取消令牌
cancel: CancellationToken,
/// 服务端事件广播
server_events: broadcast::Sender<ServerEvent>,
/// 事件总线
event_bus: Arc<EventBus>,
}
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<ServerEvent> {
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<Session>, 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()
packages/client-v2/src/app/client/session.rs
+5
View File
@@ -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()
packages/client-v2/src/app/server/mod.rs
+20 -116
View File
@@ -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<AudioBus>,
/// 服务端事件总线
event_bus: Arc<ServerEventBus>,
event_bus: Arc<EventBus>,
/// 服务器取消令牌
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<ServerEventBus> {
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<Session>, 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(())
}
packages/client-v2/src/bin/client.rs
+14 -7
View File
@@ -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═══════════════════════════════════════════════════════");
packages/client-v2/src/bin/server.rs
+18 -17
View File
@@ -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)...");
packages/client-v2/src/net/event.rs
+15 -185
View File
@@ -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<u8> },
}
/// 客户端事件（Client → Server）
#[derive(Serialize, Deserialize, Debug, Clone)]
pub enum ClientEvent {
/// 状态更新
StatusUpdate {
cpu_usage: f32,
memory_usage: f32,
temperature: Option<f32>,
},
/// 音频电平
AudioLevel { level_db: f32, is_silent: bool },
/// 按键事件
KeyPress { key: String, action: KeyAction },
/// 警告/错误
Alert {
level: NotificationLevel,
message: String,
},
/// 自定义事件
Custom { name: String, payload: Vec<u8> },
}
/// 通知级别
#[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<E: Clone> EventSubscription<E> {
}
}
/// 服务端事件总线
pub struct ServerEventBus {
tx: broadcast::Sender<(Option<SocketAddr>, 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<SocketAddr>, 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<E>: Send + Sync {
async fn handle(&self, event: E);
}
/// 运行事件处理循环
pub async fn run_event_loop<E, H>(mut subscription: EventSubscription<E>, handler: Arc<H>)
where
E: Clone + Send + 'static,
H: EventHandler<E> + 'static,
{
while let Some(event) = subscription.recv().await {
handler.handle(event).await;
Self::new(128)
}
}
packages/client-v2/src/net/protocol.rs
+5 -16
View File
@@ -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<String>,
},
/// 取消订阅
Unsubscribe {
event_types: Vec<String>,
Event {
/// 微秒时间戳
timestamp: u128,
data: EventData,
},
// ========== 音频控制 ==========
packages/client-v2/src/net/sync.rs
-2
View File
@@ -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 {