From d34f5dd9521c03335e2e9ad6fbecff284933b15b Mon Sep 17 00:00:00 2001 From: ReTeamSpeak Date: Tue, 12 May 2026 19:53:22 +0900 Subject: [PATCH] refactor: handle_data now returns HandleResult with parsed events - Add CommandEvent enum (InitServer, ChannelList, ClientList, TextMessage, etc.) - Add HandleResult struct with responses and events - handle_data returns HandleResult instead of Vec> - Session uses HandleResult for real event emission - Update socket.rs and tests for new return type --- src/tscore/src/connection/client.rs | 216 +++++++++++++++++-- src/tscore/src/connection/session.rs | 312 +++++---------------------- src/tscore/src/network/socket.rs | 4 +- 3 files changed, 252 insertions(+), 280 deletions(-) diff --git a/src/tscore/src/connection/client.rs b/src/tscore/src/connection/client.rs index acb3de7..1da8c13 100644 --- a/src/tscore/src/connection/client.rs +++ b/src/tscore/src/connection/client.rs @@ -11,6 +11,69 @@ use crate::protocol::{ }; use crate::ProtocolError; +#[derive(Debug, Clone, serde::Serialize)] +pub struct ChannelEntry { + pub cid: u64, + pub pid: u64, + pub channel_order: u64, + pub channel_name: String, + pub total_clients: u16, + pub channel_needed_subscribe_power: i32, +} + +#[derive(Debug, Clone, serde::Serialize)] +pub struct ClientEntry { + pub clid: u16, + pub cid: u64, + pub client_database_id: u64, + pub client_nickname: String, + pub client_type: u8, +} + +#[derive(Debug, Clone, serde::Serialize)] +pub enum CommandEvent { + InitServer { + client_id: u16, + name: String, + platform: String, + version: String, + max_clients: u16, + clients_online: u16, + channels_online: u16, + }, + ChannelList(Vec), + ChannelListFinished, + ClientList(Vec), + ClientEntered { + clid: u16, + cid: u64, + client_nickname: String, + }, + ClientLeft { + clid: u16, + reason: String, + }, + ClientMoved { + clid: u16, + cid: u64, + }, + TextMessage { + invoker_id: u16, + invoker_name: String, + message: String, + target_mode: u8, + }, + Error { + id: u32, + message: String, + }, +} + +pub struct HandleResult { + pub responses: Vec>, + pub events: Vec, +} + /// 客户端配置 #[derive(Debug, Clone)] pub struct ClientConfig { @@ -128,8 +191,9 @@ impl Client { } /// 处理接收到的数据 - pub fn handle_data(&mut self, data: &[u8]) -> Result>, ProtocolError> { + pub fn handle_data(&mut self, data: &[u8]) -> Result { let mut responses = Vec::new(); + let mut events = Vec::new(); match self.state() { ConnectionState::Connecting => { @@ -191,7 +255,7 @@ impl Client { responses.push(self.build_clientinit_packet()?); } } - return Ok(responses); + return Ok(HandleResult { responses, events }); } if matches!(packet_type, PacketType::Command | PacketType::CommandLow) { @@ -210,6 +274,34 @@ impl Client { self.state_machine .transition(ConnectionState::ChannelListFinished) .map_err(ProtocolError::PacketParse)?; + + events.push(CommandEvent::InitServer { + client_id: self.client_id.unwrap_or(0), + name: cmd + .get("virtualserver_name") + .unwrap_or_default() + .to_string(), + platform: cmd + .get("virtualserver_platform") + .unwrap_or_default() + .to_string(), + version: cmd + .get("virtualserver_version") + .unwrap_or_default() + .to_string(), + max_clients: cmd + .get("virtualserver_maxclients") + .and_then(|v| v.parse().ok()) + .unwrap_or(0), + clients_online: cmd + .get("virtualserver_clientsonline") + .and_then(|v| v.parse().ok()) + .unwrap_or(0), + channels_online: cmd + .get("virtualserver_channelsonline") + .and_then(|v| v.parse().ok()) + .unwrap_or(0), + }); } "initivexpand" => { // 旧协议密钥交换 @@ -220,23 +312,64 @@ impl Client { responses.extend(self.handle_initivexpand2(&cmd)?); } "channellist" => { - // 频道列表 + events + .push(CommandEvent::ChannelList(Self::parse_channel_entries(&cmd))); } "channellistfinished" => { self.state_machine .transition(ConnectionState::ChannelListFinished) .map_err(ProtocolError::PacketParse)?; + events.push(CommandEvent::ChannelListFinished); + } + "clientlist" => { + events.push(CommandEvent::ClientList(Self::parse_client_entries(&cmd))); } "notifycliententerview" => { - // 客户端进入视图 + events.push(CommandEvent::ClientEntered { + clid: cmd.get("clid").and_then(|v| v.parse().ok()).unwrap_or(0), + cid: cmd.get("cid").and_then(|v| v.parse().ok()).unwrap_or(0), + client_nickname: cmd + .get("client_nickname") + .unwrap_or_default() + .to_string(), + }); + } + "notifyclientleftview" => { + events.push(CommandEvent::ClientLeft { + clid: cmd.get("clid").and_then(|v| v.parse().ok()).unwrap_or(0), + reason: cmd.get("reasonmsg").unwrap_or_default().to_string(), + }); + } + "notifyclientmoved" => { + events.push(CommandEvent::ClientMoved { + clid: cmd.get("clid").and_then(|v| v.parse().ok()).unwrap_or(0), + cid: cmd.get("ctid").and_then(|v| v.parse().ok()).unwrap_or(0), + }); + } + "notifytextmessage" => { + events.push(CommandEvent::TextMessage { + invoker_id: cmd + .get("invokerid") + .and_then(|v| v.parse().ok()) + .unwrap_or(0), + invoker_name: cmd + .get("invokername") + .unwrap_or_default() + .to_string(), + message: cmd.get("msg").unwrap_or_default().to_string(), + target_mode: cmd + .get("targetmode") + .and_then(|v| v.parse().ok()) + .unwrap_or(0), + }); } "error" => { if let Some(id) = cmd.get("id") { if id != "0" { - return Err(ProtocolError::PacketParse(format!( - "服务器错误: {}", - cmd.get("msg").unwrap_or("未知") - ))); + events.push(CommandEvent::Error { + id: id.parse().unwrap_or(0), + message: cmd.get("msg").unwrap_or("unknown").to_string(), + }); } } } @@ -247,7 +380,58 @@ impl Client { _ => {} } - Ok(responses) + Ok(HandleResult { responses, events }) + } + + fn parse_channel_entries(cmd: &Command) -> Vec { + let mut entries = Vec::new(); + let cid = cmd.get("cid").and_then(|v| v.parse().ok()).unwrap_or(0); + let pid = cmd.get("pid").and_then(|v| v.parse().ok()).unwrap_or(0); + let channel_order = cmd + .get("channel_order") + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + let channel_name = cmd.get("channel_name").unwrap_or_default().to_string(); + let total_clients = cmd + .get("total_clients") + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + let channel_needed_subscribe_power = cmd + .get("channel_needed_subscribe_power") + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + + entries.push(ChannelEntry { + cid, + pid, + channel_order, + channel_name, + total_clients, + channel_needed_subscribe_power, + }); + entries + } + + fn parse_client_entries(cmd: &Command) -> Vec { + let clid = cmd.get("clid").and_then(|v| v.parse().ok()).unwrap_or(0); + let cid = cmd.get("cid").and_then(|v| v.parse().ok()).unwrap_or(0); + let client_database_id = cmd + .get("client_database_id") + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + let client_nickname = cmd.get("client_nickname").unwrap_or_default().to_string(); + let client_type = cmd + .get("client_type") + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + + vec![ClientEntry { + clid, + cid, + client_database_id, + client_nickname, + client_type, + }] } /// 构建 Init2 数据包 @@ -696,16 +880,16 @@ mod tests { server_packet.set_packet_id(0); crypto::encrypt_fake(&mut server_packet).unwrap(); - let responses = client.handle_data(&server_packet.to_bytes()).unwrap(); - assert_eq!(responses.len(), 2); + let result = client.handle_data(&server_packet.to_bytes()).unwrap(); + assert_eq!(result.responses.len(), 2); - let ack = InPacket::parse(Direction::C2S, &responses[0]).unwrap(); + let ack = InPacket::parse(Direction::C2S, &result.responses[0]).unwrap(); assert_eq!(ack.header.packet_id, 0); assert_eq!(ack.header.flags.packet_type(), PacketType::Ack); let ack_content = crypto::decrypt_fake(&ack).unwrap(); assert_eq!(ack_content, 0u16.to_be_bytes()); - let clientek = InPacket::parse(Direction::C2S, &responses[1]).unwrap(); + let clientek = InPacket::parse(Direction::C2S, &result.responses[1]).unwrap(); assert_eq!(clientek.header.packet_id, 1); assert_eq!(clientek.header.flags.packet_type(), PacketType::Command); assert!(clientek.header.flags.is_newprotocol()); @@ -732,10 +916,10 @@ mod tests { ack.set_packet_id(0); crypto::encrypt_fake(&mut ack).unwrap(); - let responses = client.handle_data(&ack.to_bytes()).unwrap(); - assert_eq!(responses.len(), 1); + let result = client.handle_data(&ack.to_bytes()).unwrap(); + assert_eq!(result.responses.len(), 1); - let clientinit = InPacket::parse(Direction::C2S, &responses[0]).unwrap(); + let clientinit = InPacket::parse(Direction::C2S, &result.responses[0]).unwrap(); assert_eq!(clientinit.header.packet_id, 2); assert_eq!(clientinit.header.flags.packet_type(), PacketType::Command); diff --git a/src/tscore/src/connection/session.rs b/src/tscore/src/connection/session.rs index 5c6a0dd..e6b6d11 100644 --- a/src/tscore/src/connection/session.rs +++ b/src/tscore/src/connection/session.rs @@ -2,9 +2,9 @@ use std::time::Duration; use tokio::net::UdpSocket; use tokio::sync::mpsc; -use super::client::{Client, ClientConfig}; +use super::client::{ChannelEntry, Client, ClientConfig, ClientEntry, CommandEvent, HandleResult}; use super::state::ConnectionState; -use crate::protocol::{Command, Direction, InPacket, PacketType}; +use crate::protocol::Command; use crate::ProtocolError; pub enum SessionCommand { @@ -71,25 +71,6 @@ pub enum SessionEvent { Disconnected, } -#[derive(Debug, Clone, serde::Serialize)] -pub struct ChannelEntry { - pub cid: u64, - pub pid: u64, - pub channel_order: u64, - pub channel_name: String, - pub total_clients: u16, - pub channel_needed_subscribe_power: i32, -} - -#[derive(Debug, Clone, serde::Serialize)] -pub struct ClientEntry { - pub clid: u16, - pub cid: u64, - pub client_database_id: u64, - pub client_nickname: String, - pub client_type: u8, -} - pub struct SessionHandle { command_tx: mpsc::Sender, event_rx: mpsc::Receiver, @@ -228,20 +209,11 @@ impl Session { let mut buf = [0u8; 2048]; loop { let len = socket.recv(&mut buf).await?; - let responses = client.handle_data(&buf[..len])?; - for response in &responses { + let result = client.handle_data(&buf[..len])?; + for response in &result.responses { socket.send(response).await?; } - - if let Ok(packet) = InPacket::parse(Direction::S2C, &buf[..len]) { - if let Some(content) = Self::decrypt_packet_static(&client, &packet) { - if let Ok(content_str) = std::str::from_utf8(&content) { - for cmd in Command::parse_many(content_str).unwrap_or_default() { - Self::collect_init_event(&cmd, &mut init_events); - } - } - } - } + init_events.extend(result.events); if client.state() == ConnectionState::ChannelListFinished { return Ok::<(), ProtocolError>(()); @@ -252,7 +224,7 @@ impl Session { .map_err(|_| ProtocolError::Timeout("handshake timed out".to_string()))??; let (command_tx, command_rx) = mpsc::channel(32); - let (event_tx, mut event_rx) = mpsc::channel(32); + let (event_tx, event_rx) = mpsc::channel(32); let session = Self { client, @@ -269,96 +241,6 @@ impl Session { Ok((session, handle)) } - fn decrypt_packet_static(client: &Client, packet: &InPacket) -> Option> { - let packet_type = packet.header.flags.packet_type(); - if packet.header.flags.is_unencrypted() { - return Some(packet.data.clone()); - } - if packet_type == PacketType::Ack || packet_type == PacketType::AckLow { - return None; - } - None - } - - fn collect_init_event(cmd: &Command, events: &mut Vec) { - match cmd.name.as_str() { - "initserver" => { - let client_id = cmd - .get("client_id") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - let name = cmd - .get("virtualserver_name") - .unwrap_or_default() - .to_string(); - let platform = cmd - .get("virtualserver_platform") - .unwrap_or_default() - .to_string(); - let version = cmd - .get("virtualserver_version") - .unwrap_or_default() - .to_string(); - let max_clients = cmd - .get("virtualserver_maxclients") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - let clients_online = cmd - .get("virtualserver_clientsonline") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - let channels_online = cmd - .get("virtualserver_channelsonline") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - - events.push(SessionEvent::ServerInfo { - name, - platform, - version, - max_clients, - clients_online, - channels_online, - }); - events.push(SessionEvent::Connected { client_id }); - } - "channellist" => { - let channels = Self::parse_channel_list(cmd); - events.push(SessionEvent::ChannelList(channels)); - } - _ => {} - } - } - - fn parse_channel_list(cmd: &Command) -> Vec { - let mut channels = Vec::new(); - let cid = cmd.get("cid").and_then(|v| v.parse().ok()).unwrap_or(0); - let pid = cmd.get("pid").and_then(|v| v.parse().ok()).unwrap_or(0); - let channel_order = cmd - .get("channel_order") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - let channel_name = cmd.get("channel_name").unwrap_or_default().to_string(); - let total_clients = cmd - .get("total_clients") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - let channel_needed_subscribe_power = cmd - .get("channel_needed_subscribe_power") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - - channels.push(ChannelEntry { - cid, - pid, - channel_order, - channel_name, - total_clients, - channel_needed_subscribe_power, - }); - channels - } - pub fn client_id(&self) -> Option { self.client.client_id() } @@ -374,22 +256,12 @@ impl Session { tokio::select! { result = self.socket.recv(&mut buf) => { let len = result?; - let responses = self.client.handle_data(&buf[..len])?; - for response in &responses { + let handle_result = self.client.handle_data(&buf[..len])?; + for response in &handle_result.responses { self.socket.send(response).await?; } - - if let Ok(packet) = InPacket::parse(Direction::S2C, &buf[..len]) { - let packet_type = packet.header.flags.packet_type(); - if packet_type == PacketType::Command || packet_type == PacketType::CommandLow { - if let Some(content) = self.decrypt_packet(&packet) { - if let Ok(content_str) = std::str::from_utf8(&content) { - for cmd in Command::parse_many(content_str).unwrap_or_default() { - self.emit_command_event(&cmd).await; - } - } - } - } + for event in handle_result.events { + self.emit_session_event(event).await; } } Some(command) = self.command_rx.recv() => { @@ -408,49 +280,17 @@ impl Session { } } - fn decrypt_packet(&self, packet: &InPacket) -> Option> { - let packet_type = packet.header.flags.packet_type(); - if packet.header.flags.is_unencrypted() { - return Some(packet.data.clone()); - } - if packet_type == PacketType::Ack || packet_type == PacketType::AckLow { - return None; - } - None - } - - async fn emit_command_event(&self, cmd: &Command) { - let event = match cmd.name.as_str() { - "initserver" => { - let client_id = cmd - .get("client_id") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - let name = cmd - .get("virtualserver_name") - .unwrap_or_default() - .to_string(); - let platform = cmd - .get("virtualserver_platform") - .unwrap_or_default() - .to_string(); - let version = cmd - .get("virtualserver_version") - .unwrap_or_default() - .to_string(); - let max_clients = cmd - .get("virtualserver_maxclients") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - let clients_online = cmd - .get("virtualserver_clientsonline") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - let channels_online = cmd - .get("virtualserver_channelsonline") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - + async fn emit_session_event(&self, event: CommandEvent) { + let session_event = match event { + CommandEvent::InitServer { + client_id, + name, + platform, + version, + max_clients, + clients_online, + channels_online, + } => { let _ = self .event_tx .send(SessionEvent::ServerInfo { @@ -462,91 +302,39 @@ impl Session { channels_online, }) .await; - Some(SessionEvent::Connected { client_id }) + SessionEvent::Connected { client_id } } - "channellist" => { - let channels = Self::parse_channel_list(cmd); - Some(SessionEvent::ChannelList(channels)) + CommandEvent::ChannelList(channels) => SessionEvent::ChannelList(channels), + CommandEvent::ChannelListFinished => return, + CommandEvent::ClientList(clients) => SessionEvent::ClientList(clients), + CommandEvent::ClientEntered { + clid, + cid, + client_nickname, + } => SessionEvent::ClientEntered { + clid, + cid, + client_nickname, + }, + CommandEvent::ClientLeft { clid, reason } => SessionEvent::ClientLeft { clid, reason }, + CommandEvent::ClientMoved { clid, cid } => SessionEvent::ClientMoved { clid, cid }, + CommandEvent::TextMessage { + invoker_id, + invoker_name, + message, + target_mode, + } => SessionEvent::TextMessage { + invoker_id, + invoker_name, + message, + target_mode, + }, + CommandEvent::Error { id, message } => { + SessionEvent::Error(format!("server error {id}: {message}")) } - "clientlist" => { - let clients = Self::parse_client_list(cmd); - Some(SessionEvent::ClientList(clients)) - } - "notifycliententerview" => { - let clid = cmd.get("clid").and_then(|v| v.parse().ok()).unwrap_or(0); - let cid = cmd.get("cid").and_then(|v| v.parse().ok()).unwrap_or(0); - let client_nickname = cmd.get("client_nickname").unwrap_or_default().to_string(); - Some(SessionEvent::ClientEntered { - clid, - cid, - client_nickname, - }) - } - "notifyclientleftview" => { - let clid = cmd.get("clid").and_then(|v| v.parse().ok()).unwrap_or(0); - let reason = cmd.get("reasonmsg").unwrap_or_default().to_string(); - Some(SessionEvent::ClientLeft { clid, reason }) - } - "notifyclientmoved" => { - let clid = cmd.get("clid").and_then(|v| v.parse().ok()).unwrap_or(0); - let cid = cmd.get("ctid").and_then(|v| v.parse().ok()).unwrap_or(0); - Some(SessionEvent::ClientMoved { clid, cid }) - } - "notifytextmessage" => { - let invoker_id = cmd - .get("invokerid") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - let invoker_name = cmd.get("invokername").unwrap_or_default().to_string(); - let message = cmd.get("msg").unwrap_or_default().to_string(); - let target_mode = cmd - .get("targetmode") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - Some(SessionEvent::TextMessage { - invoker_id, - invoker_name, - message, - target_mode, - }) - } - "error" => { - let id = cmd.get("id").unwrap_or("0"); - if id != "0" { - let msg = cmd.get("msg").unwrap_or("unknown error").to_string(); - Some(SessionEvent::Error(msg)) - } else { - None - } - } - _ => None, }; - if let Some(event) = event { - let _ = self.event_tx.send(event).await; - } - } - - fn parse_client_list(cmd: &Command) -> Vec { - let clid = cmd.get("clid").and_then(|v| v.parse().ok()).unwrap_or(0); - let cid = cmd.get("cid").and_then(|v| v.parse().ok()).unwrap_or(0); - let client_database_id = cmd - .get("client_database_id") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - let client_nickname = cmd.get("client_nickname").unwrap_or_default().to_string(); - let client_type = cmd - .get("client_type") - .and_then(|v| v.parse().ok()) - .unwrap_or(0); - - vec![ClientEntry { - clid, - cid, - client_database_id, - client_nickname, - client_type, - }] + let _ = self.event_tx.send(session_event).await; } async fn handle_command(&mut self, command: SessionCommand) -> Result<(), ProtocolError> { diff --git a/src/tscore/src/network/socket.rs b/src/tscore/src/network/socket.rs index 7aaf7f4..6ca7179 100644 --- a/src/tscore/src/network/socket.rs +++ b/src/tscore/src/network/socket.rs @@ -122,8 +122,8 @@ async fn perform_handshake_until( let mut buf = [0u8; 2048]; loop { let len = socket.recv(&mut buf).await?; - let responses = client.handle_data(&buf[..len])?; - for response in responses { + let result = client.handle_data(&buf[..len])?; + for response in result.responses { socket.send(&response).await?; }