Raven/osirose-new
SHA256
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects 6 Releases Wiki Activity

Refactor

Browse Source 
Updated handlers by spliting the TcpStream in half to allow reading and writing data at the same time.
This fixes an issue where you are unable to get chat messages until the client sends a packet to the server

Fixed client id's by adding the id manager
Added shout chat handling
This commit is contained in:
raven
2025-06-07 00:36:02 -04:00
parent d4dadf5170
commit aa2be43f4e
17 changed files with 480 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
packet-service/src/router.rs
View File

@@ -8,7 +8,7 @@ use crate::packet::Packet;
use crate::packet_type::PacketType;
use std::error::Error;
use std::sync::Arc;
use tokio::io::AsyncReadExt;
use tokio::io::{AsyncReadExt, ReadHalf};
use tokio::net::TcpStream;
use tokio::sync::Mutex;
use tracing::{debug, warn};
@@ -23,30 +23,20 @@ pub struct PacketRouter {
impl PacketRouter {
pub async fn handle_connection(
&self,
stream: Arc<Mutex<TcpStream>>,
mut stream: ReadHalf<TcpStream>,
pool: Arc<BufferPool>,
connection_id: String,
peer_addr: String,
) -> Result<(), Box<dyn Error + Send + Sync>> {
ACTIVE_CONNECTIONS.inc();
while let Some(mut buffer) = pool.acquire().await {
// Read data into the buffer
let packet_size: usize;
{
let mut locked_stream = stream.lock().await;
locked_stream.read_exact(&mut buffer[..6]).await?;
// let mut header_handle = locked_stream.take(6);
// let n = header_handle.read(&mut buffer).await?;
// if n == 0 {
// break; // Connection closed
// }
stream.read_exact(&mut buffer[..6]).await?;
packet_size = u16::from_le_bytes(buffer[0..2].try_into()?) as usize;
if packet_size > 6 {
locked_stream.read_exact(&mut buffer[6..packet_size]).await?;
// let mut body_handle = locked_stream.take((packet_size - 6) as u64);
// let n = body_handle.read(&mut buffer[6..]).await?;
// if n == 0 {
// break; // Connection closed
// }
stream.read_exact(&mut buffer[6..packet_size]).await?;
}
}
@@ -57,8 +47,7 @@ impl PacketRouter {
match Packet::from_raw(&buffer[..packet_size]) {
Ok(packet) => {
debug!("Parsed Packet: {:?}", packet);
// Handle the parsed packet (route it, process it, etc.)
self.route_packet(stream.clone(), packet, connection_id.clone()).await?;
self.route_packet(packet, connection_id.clone()).await?;
}
Err(e) => warn!("Failed to parse packet: {}", e),
}
@@ -73,8 +62,7 @@ impl PacketRouter {
let mut auth_client = self.auth_client.lock().await;
auth_client.logout(&session_id).await?;
} else {
let mut locked_stream = stream.lock().await;
warn!("No session found for {}", locked_stream.peer_addr()?);
warn!("No session found for {}", peer_addr);
}
}
ACTIVE_CONNECTIONS.dec();
@@ -84,34 +72,36 @@ impl PacketRouter {
#[rustfmt::skip]
pub async fn route_packet(
&self,
stream: Arc<Mutex<TcpStream>>,
packet: Packet,
connection_id: String,
) -> Result<(), Box<dyn Error + Send + Sync>> {
debug!("Routing packet: {:?}", packet);
match packet.packet_type {
// Generic Server Packets
PacketType::PakcsAlive => auth::handle_alive_req(stream, packet, self.auth_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsAcceptReq => auth::handle_accept_req(stream, packet).await,
PacketType::PakcsJoinServerTokenReq => auth::handle_join_server_req(stream, packet, self.auth_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsAlive => auth::handle_alive_req( packet, self.auth_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsAcceptReq => auth::handle_accept_req(packet, self.connection_service.clone(), connection_id).await,
PacketType::PakcsJoinServerTokenReq => auth::handle_join_server_req(packet, self.auth_client.clone(), self.connection_service.clone(), connection_id).await,
// Login Packets
PacketType::PakcsLoginTokenReq => auth::handle_login_req(stream, packet, self.auth_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsLogoutReq => auth::handle_logout_req(stream, packet, self.auth_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsSrvSelectReq => auth::handle_server_select_req(stream, packet, self.connection_service.clone(), connection_id).await,
PacketType::PakcsChannelListReq => auth::handle_channel_list_req(stream, packet).await,
PacketType::PakcsLoginTokenReq => auth::handle_login_req(packet, self.auth_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsLogoutReq => auth::handle_logout_req(packet, self.auth_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsSrvSelectReq => auth::handle_server_select_req(packet, self.connection_service.clone(), connection_id).await,
PacketType::PakcsChannelListReq => auth::handle_channel_list_req(packet, self.connection_service.clone(), connection_id).await,
// Character Packets
PacketType::PakcsCharListReq => character::handle_char_list_req(stream, packet, self.character_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsCreateCharReq => character::handle_create_char_req(stream, packet, self.character_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsDeleteCharReq => character::handle_delete_char_req(stream, packet, self.character_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsSelectCharReq => character::handle_select_char_req(stream, packet, self.character_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsCharListReq => character::handle_char_list_req(packet, self.character_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsCreateCharReq => character::handle_create_char_req(packet, self.character_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsDeleteCharReq => character::handle_delete_char_req(packet, self.character_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsSelectCharReq => character::handle_select_char_req(packet, self.character_client.clone(), self.connection_service.clone(), connection_id).await,
// World Packets
PacketType::PakcsChangeMapReq => world::handle_change_map_req(stream, packet, self.character_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsMouseCmd => world::handle_mouse_cmd_req(stream, packet, self.connection_service.clone(), connection_id).await,
PacketType::PakcsChangeMapReq => world::handle_change_map_req(packet, self.character_client.clone(), self.connection_service.clone(), connection_id).await,
PacketType::PakcsMouseCmd => world::handle_mouse_cmd_req(packet, self.connection_service.clone(), connection_id).await,
PacketType::PakcsToggleMove => world::handle_togggle_move_req(packet, self.connection_service.clone(), connection_id).await,
PacketType::PakcsSetAnimation => world::handle_set_animation_req(packet, self.connection_service.clone(), connection_id).await,
// Chat Packets
PacketType::PakcsNormalChat => chat::handle_normal_chat(stream, packet, self.connection_service.clone(), connection_id).await,
PacketType::PakcsNormalChat => chat::handle_normal_chat(packet, self.connection_service.clone(), connection_id).await,
PacketType::PakcsShoutChat => chat::handle_shout_chat(packet, self.connection_service.clone(), connection_id).await,
// 1 => chat::handle_chat(packet).await?,
// 2 => movement::handle_movement(packet).await?,
_ => {