edison/re-teamspeak
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: handle_data now returns HandleResult with parsed events

Browse Source 
- Add CommandEvent enum (InitServer, ChannelList, ClientList, TextMessage, etc.)
- Add HandleResult struct with responses and events
- handle_data returns HandleResult instead of Vec<Vec<u8>>
- Session uses HandleResult for real event emission
- Update socket.rs and tests for new return type
This commit is contained in:
ReTeamSpeak
2026-05-12 19:53:22 +09:00
parent ace2f5deed
commit d34f5dd952
3 changed files with 252 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tscore/src/connection/client.rs
+200 -16
View File
@@ -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<ChannelEntry>),
ChannelListFinished,
ClientList(Vec<ClientEntry>),
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<Vec<u8>>,
pub events: Vec<CommandEvent>,
}
/// 客户端配置
#[derive(Debug, Clone)]
pub struct ClientConfig {
@@ -128,8 +191,9 @@ impl Client {
}
/// 处理接收到的数据
pub fn handle_data(&mut self, data: &[u8]) -> Result<Vec<Vec<u8>>, ProtocolError> {
pub fn handle_data(&mut self, data: &[u8]) -> Result<HandleResult, ProtocolError> {
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<ChannelEntry> {
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<ClientEntry> {
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);
src/tscore/src/connection/session.rs
+50 -262
View File
@@ -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<SessionCommand>,
event_rx: mpsc::Receiver<SessionEvent>,
@@ -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<Vec<u8>> {
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<SessionEvent>) {
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<ChannelEntry> {
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<u16> {
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<Vec<u8>> {
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<ClientEntry> {
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> {
src/tscore/src/network/socket.rs
+2 -2
View File
@@ -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?;
}