From 4a826d2a46ead6cc3505cd5fd88fda31cbf85689ea19835fa2767206e850dede Mon Sep 17 00:00:00 2001 From: raven <7156279+RavenX8@users.noreply.github.com> Date: Sat, 4 Jan 2025 17:45:54 -0500 Subject: [PATCH] - update: packet router to have the various services needed for the packets to be local to it. - add: character service grpc client calls --- character-service/src/character_service.rs | 4 +- packet-service/src/auth_client.rs | 1 + packet-service/src/character_client.rs | 51 +++++++++++ packet-service/src/connection_service.rs | 1 + packet-service/src/handlers/character.rs | 48 +++++++++-- packet-service/src/main.rs | 63 +++++--------- packet-service/src/router.rs | 99 ++++++++++++++++------ 7 files changed, 189 insertions(+), 78 deletions(-) create mode 100644 packet-service/src/character_client.rs diff --git a/character-service/src/character_service.rs b/character-service/src/character_service.rs index 77c3b67..2eae83b 100644 --- a/character-service/src/character_service.rs +++ b/character-service/src/character_service.rs @@ -29,7 +29,7 @@ impl CharacterService for MyCharacterService { job: 111, last_played: 1633017600, // Example timestamp delete_time: 0, - face: 0, + face: 1, hair: 0, items: vec![], }, @@ -41,7 +41,7 @@ impl CharacterService for MyCharacterService { job: 211, last_played: 1633017600, // Example timestamp delete_time: 0, - face: 0, + face: 1, hair: 0, items: vec![], }, diff --git a/packet-service/src/auth_client.rs b/packet-service/src/auth_client.rs index e9b077c..829be6a 100644 --- a/packet-service/src/auth_client.rs +++ b/packet-service/src/auth_client.rs @@ -2,6 +2,7 @@ use crate::auth::auth_service_client::AuthServiceClient; use crate::auth::{Empty, LoginRequest, LoginResponse, LogoutRequest, ValidateSessionRequest, ValidateSessionResponse, ValidateTokenRequest, ValidateTokenResponse}; use tonic::transport::Channel; +#[derive(Clone, Debug)] pub struct AuthClient { client: AuthServiceClient, } diff --git a/packet-service/src/character_client.rs b/packet-service/src/character_client.rs new file mode 100644 index 0000000..d8daebd --- /dev/null +++ b/packet-service/src/character_client.rs @@ -0,0 +1,51 @@ +use crate::char::character_service_client::CharacterServiceClient; +use crate::char::{GetCharacterListRequest, GetCharacterListResponse, Character, EquippedItem}; +use tonic::transport::Channel; + +#[derive(Clone, Debug)] +pub struct CharacterClient { + client: CharacterServiceClient, +} + +impl CharacterClient { + pub async fn connect(endpoint: &str) -> Result> { + let client = CharacterServiceClient::connect(endpoint.to_string()).await?; + Ok(CharacterClient { client }) + } + + pub async fn get_character_list(&mut self, user_id: &str) -> Result> { + let request = GetCharacterListRequest { + user_id: user_id.to_string(), + }; + + let response = self.client.get_character_list(request).await?; + Ok(response.into_inner()) + } + + pub async fn create_character(&mut self, user_id: &str) -> Result> { + let request = GetCharacterListRequest { + user_id: user_id.to_string(), + }; + + let response = self.client.get_character_list(request).await?; + Ok(response.into_inner()) + } + + pub async fn delete_character(&mut self, user_id: &str) -> Result> { + let request = GetCharacterListRequest { + user_id: user_id.to_string(), + }; + + let response = self.client.get_character_list(request).await?; + Ok(response.into_inner()) + } + + pub async fn get_character(&mut self, char_id: u8) -> Result> { + let request = GetCharacterListRequest { + user_id: char_id.to_string(), + }; + + let response = self.client.get_character_list(request).await?; + Ok(response.into_inner()) + } +} diff --git a/packet-service/src/connection_service.rs b/packet-service/src/connection_service.rs index e5c2704..b4a9eab 100644 --- a/packet-service/src/connection_service.rs +++ b/packet-service/src/connection_service.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use uuid::Uuid; use crate::connection_state::ConnectionState; +#[derive(Clone, Debug)] pub struct ConnectionService { pub connections: Arc>, // Map connection ID to state } diff --git a/packet-service/src/handlers/character.rs b/packet-service/src/handlers/character.rs index eefe14a..1e457bd 100644 --- a/packet-service/src/handlers/character.rs +++ b/packet-service/src/handlers/character.rs @@ -12,16 +12,18 @@ use tokio::net::TcpStream; use tokio::sync::Mutex; use tonic::{Code, Status}; use tracing::{debug, error, info, warn}; +use crate::auth_client::AuthClient; +use crate::character_client::CharacterClient; use crate::connection_service::ConnectionService; use crate::packets::cli_create_char_req::CliCreateCharReq; use crate::packets::cli_delete_char_req::CliDeleteCharReq; use crate::packets::cli_select_char_req::CliSelectCharReq; -use crate::packets::srv_char_list_reply::SrvCharListReply; +use crate::packets::srv_char_list_reply::{CharInfo, EquippedItem, SrvCharListReply}; use crate::packets::srv_create_char_reply::SrvCreateCharReply; use crate::packets::srv_delete_char_reply::SrvDeleteCharReply; use crate::packets::srv_select_char_reply::SrvSelectCharReply; -pub(crate) async fn handle_char_list_req(stream: &mut TcpStream, packet: Packet, connection_service: Arc, connection_id: String) -> Result<(), Box> { +pub(crate) async fn handle_char_list_req(stream: &mut TcpStream, packet: Packet, character_client: Arc>, connection_service: Arc, connection_id: String) -> Result<(), Box> { let request = CliCharListReq::decode(packet.payload.as_slice()); debug!("{:?}", request); @@ -31,17 +33,40 @@ pub(crate) async fn handle_char_list_req(stream: &mut TcpStream, packet: Packet, user_id = state.user_id.expect("Missing user id in connection state"); session_id = state.session_id.expect("Missing session id in connection state"); } - + // query the character service for the character list for this user - - let data = SrvCharListReply { characters: vec![] }; + let mut character_client = character_client.lock().await; + let character_list = character_client.get_character_list(&user_id.to_string()).await?; + debug!("{:?}", character_list.characters); + let mut characters = vec![]; + for character in character_list.characters { + // let item_list = character.items.iter().map(|item| { + // + // }); + + let character_info = CharInfo { + name: NullTerminatedString(character.name), + race: character.race as u8, + level: character.level as u16, + job: character.job as u16, + remain_secs_until_delete: character.delete_time as u32, + platinium: 0, + face: character.face as u32, + hair: character.hair as u32, + items: vec![], + }; + characters.push(character_info); + } + + debug!("{:?}", characters); + let data = SrvCharListReply { characters }; let response_packet = Packet::new(PacketType::PakccCharListReply, &data)?; send_packet(stream, &response_packet).await?; Ok(()) } -pub(crate) async fn handle_create_char_req(stream: &mut TcpStream, packet: Packet, connection_service: Arc, connection_id: String) -> Result<(), Box> { +pub(crate) async fn handle_create_char_req(stream: &mut TcpStream, packet: Packet, character_client: Arc>, connection_service: Arc, connection_id: String) -> Result<(), Box> { let request = CliCreateCharReq::decode(packet.payload.as_slice())?; debug!("{:?}", request); @@ -51,6 +76,8 @@ pub(crate) async fn handle_create_char_req(stream: &mut TcpStream, packet: Packe } // send the data to the character service to create the character + let mut character_client = character_client.lock().await; + // character_client.create_character(&session_id).await?; let data = SrvCreateCharReply { result: srv_create_char_reply::Result::Ok, platininum: 0 }; let response_packet = Packet::new(PacketType::PakccCreateCharReply, &data)?; @@ -59,7 +86,7 @@ pub(crate) async fn handle_create_char_req(stream: &mut TcpStream, packet: Packe Ok(()) } -pub(crate) async fn handle_delete_char_req(stream: &mut TcpStream, packet: Packet, connection_service: Arc, connection_id: String) -> Result<(), Box> { +pub(crate) async fn handle_delete_char_req(stream: &mut TcpStream, packet: Packet, character_client: Arc>, connection_service: Arc, connection_id: String) -> Result<(), Box> { let request = CliDeleteCharReq::decode(packet.payload.as_slice())?; debug!("{:?}", request); @@ -67,7 +94,8 @@ pub(crate) async fn handle_delete_char_req(stream: &mut TcpStream, packet: Packe if let Some(mut state) = connection_service.get_connection(&connection_id) { session_id = state.session_id.expect("Missing session id in connection state"); } - + + let mut character_client = character_client.lock().await; // character_client.set_delete_character(request.is_delete, request.char_id, request.name); let character_name = request.name; @@ -78,7 +106,7 @@ pub(crate) async fn handle_delete_char_req(stream: &mut TcpStream, packet: Packe Ok(()) } -pub(crate) async fn handle_select_char_req(stream: &mut TcpStream, packet: Packet, connection_service: Arc, connection_id: String) -> Result<(), Box> { +pub(crate) async fn handle_select_char_req(stream: &mut TcpStream, packet: Packet, character_client: Arc>, connection_service: Arc, connection_id: String) -> Result<(), Box> { let request = CliSelectCharReq::decode(packet.payload.as_slice())?; debug!("{:?}", request); @@ -86,6 +114,8 @@ pub(crate) async fn handle_select_char_req(stream: &mut TcpStream, packet: Packe state.character_id = Some(request.char_id as i8); } + let mut character_client = character_client.lock().await; + // character_client.get_character(request.char_id); let data = SrvSelectCharReply { race: 0, map: 0, diff --git a/packet-service/src/main.rs b/packet-service/src/main.rs index 8bf4f46..34bce3b 100644 --- a/packet-service/src/main.rs +++ b/packet-service/src/main.rs @@ -18,7 +18,9 @@ use tracing::{debug, error, info, warn}; use utils::consul_registration; use utils::service_discovery::get_service_address; use warp::Filter; +use crate::character_client::CharacterClient; use crate::connection_service::ConnectionService; +use crate::router::PacketRouter; mod packet_type; mod packet; @@ -31,53 +33,20 @@ mod handlers; mod bufferpool; mod metrics; mod auth_client; +mod character_client; mod connection_state; mod connection_service; pub mod auth { tonic::include_proto!("auth"); // Path matches the package name in auth.proto } +pub mod char { + tonic::include_proto!("character"); // Path matches the package name in auth.proto +} const BUFFER_POOL_SIZE: usize = 1000; const MAX_CONCURRENT_CONNECTIONS: usize = 100; - -async fn handle_connection(stream: &mut TcpStream, pool: Arc, auth_client: Arc>, connection_service: Arc, connection_id: String) -> Result<(), Box> { - ACTIVE_CONNECTIONS.inc(); - while let Some(mut buffer) = pool.acquire().await { - // Read data into the buffer - let n = stream.read(&mut buffer).await?; - if n == 0 { - break; // Connection closed - } - PACKETS_RECEIVED.inc(); - - // Process the packet - match Packet::from_raw(&buffer[..n]) { - Ok(packet) => { - debug!("Parsed Packet: {:?}", packet); - // Handle the parsed packet (route it, process it, etc.) - router::route_packet(stream, packet, auth_client.clone(), connection_service.clone(), connection_id.clone()).await?; - } - Err(e) => warn!("Failed to parse packet: {}", e), - } - - pool.release(buffer).await; - } - - if let Some(state) = connection_service.get_connection(&connection_id) { - let session_id = state.session_id.unwrap_or_default(); - if !session_id.is_empty() { - let mut auth_client = auth_client.lock().await; - auth_client.logout(&session_id).await?; - } else { - warn!("No session found for {}", stream.peer_addr()?); - } - } - ACTIVE_CONNECTIONS.dec(); - Ok(()) -} - #[tokio::main] async fn main() -> Result<(), Box> { dotenv().ok(); @@ -97,6 +66,7 @@ async fn main() -> Result<(), Box> { let health_check_url = format!("http://{}:{}/health", service_address, health_port); let health_check_endpoint_addr = format!("{}:{}", service_address, health_port); let auth_node = get_service_address(&consul_url, "auth-service").await?; + let character_node = get_service_address(&consul_url, "character-service").await?; // Register service with Consul let service_id = consul_registration::get_or_generate_service_id(env!("CARGO_PKG_NAME")); @@ -121,6 +91,10 @@ async fn main() -> Result<(), Box> { let auth_url = format!("http://{}:{}", auth_address.ServiceAddress, auth_address.ServicePort); let auth_client = Arc::new(Mutex::new(AuthClient::connect(&auth_url).await?)); + let character_address = character_node.get(0).unwrap(); + let character_url = format!("http://{}:{}", character_address.ServiceAddress, character_address.ServicePort); + let character_client = Arc::new(Mutex::new(CharacterClient::connect(&character_url).await?)); + let full_addr = format!("{}:{}", &addr, port); tokio::spawn(async move { @@ -129,12 +103,17 @@ async fn main() -> Result<(), Box> { let buffer_pool = BufferPool::new(BUFFER_POOL_SIZE); let connection_service = Arc::new(ConnectionService::new()); + let packet_router = PacketRouter { + auth_client, + character_client, + connection_service, + }; + info!("Packet service listening on {}", full_addr); loop { let (mut socket, addr) = listener.accept().await.unwrap(); - let auth_client = auth_client.clone(); - let connection_service = connection_service.clone(); + let packet_router = packet_router.clone(); info!("New connection from {}", addr); let pool = buffer_pool.clone(); @@ -143,11 +122,11 @@ async fn main() -> Result<(), Box> { // Spawn a new task for each connection tokio::spawn(async move { let _permit = permit; - let connection_id = connection_service.add_connection(); - if let Err(e) = handle_connection(&mut socket, pool, auth_client, connection_service.clone(), connection_id.clone()).await { + let connection_id = packet_router.connection_service.add_connection(); + if let Err(e) = packet_router.handle_connection(&mut socket, pool, connection_id.clone()).await { error!("Error handling connection: {}", e); } - connection_service.remove_connection(&connection_id); + packet_router.connection_service.remove_connection(&connection_id); }); } }); diff --git a/packet-service/src/router.rs b/packet-service/src/router.rs index ba35042..14bd001 100644 --- a/packet-service/src/router.rs +++ b/packet-service/src/router.rs @@ -4,35 +4,84 @@ use crate::packet::Packet; use crate::packet_type::PacketType; use std::error::Error; use std::sync::Arc; +use tokio::io::AsyncReadExt; use tokio::net::TcpStream; use tokio::sync::Mutex; use tracing::{debug, warn}; +use crate::bufferpool::BufferPool; +use crate::character_client::CharacterClient; use crate::connection_service::ConnectionService; +use crate::metrics::{ACTIVE_CONNECTIONS, PACKETS_RECEIVED}; -pub async fn route_packet(stream: &mut TcpStream, packet: Packet, auth_client: Arc>, connection_service: Arc, connection_id: String) -> Result<(), Box> { - debug!("Routing packet: {:?}", packet); - match packet.packet_type { - PacketType::PakcsAlive => Ok(()), - // Generic Server Packets - PacketType::PakcsAcceptReq => auth::handle_accept_req(stream, packet).await, - PacketType::PakcsJoinServerTokenReq => auth::handle_join_server_req(stream, packet, auth_client, connection_service, connection_id).await, - // Login Packets - PacketType::PakcsLoginTokenReq => auth::handle_login_req(stream, packet, auth_client, connection_service, connection_id, stream.peer_addr()?).await, - PacketType::PakcsLogoutReq => auth::handle_logout_req(stream, packet, auth_client, connection_service, connection_id).await, - PacketType::PakcsSrvSelectReq => auth::handle_server_select_req(stream, packet, connection_service, connection_id).await, - PacketType::PakcsChannelListReq => auth::handle_channel_list_req(stream, packet).await, - - // Character Packets - PacketType::PakcsCharListReq => character::handle_char_list_req(stream, packet, connection_service, connection_id).await, - PacketType::PakcsCreateCharReq => character::handle_create_char_req(stream, packet, connection_service, connection_id).await, - PacketType::PakcsDeleteCharReq => character::handle_delete_char_req(stream, packet, connection_service, connection_id).await, - PacketType::PakcsSelectCharReq => character::handle_select_char_req(stream, packet, connection_service, connection_id).await, - - // 1 => chat::handle_chat(packet).await?, - // 2 => movement::handle_movement(packet).await?, - _ => { - warn!("Unhandled packet type: {:?}", packet.packet_type); - Ok(()) - }, +#[derive(Clone, Debug)] +pub struct PacketRouter { + pub auth_client: Arc>, + pub character_client: Arc>, + pub connection_service: Arc, +} + +impl PacketRouter { + pub async fn handle_connection(&self, stream: &mut TcpStream, pool: Arc, connection_id: String) -> Result<(), Box> { + ACTIVE_CONNECTIONS.inc(); + while let Some(mut buffer) = pool.acquire().await { + // Read data into the buffer + let n = stream.read(&mut buffer).await?; + if n == 0 { + break; // Connection closed + } + PACKETS_RECEIVED.inc(); + + // Process the packet + match Packet::from_raw(&buffer[..n]) { + Ok(packet) => { + debug!("Parsed Packet: {:?}", packet); + // Handle the parsed packet (route it, process it, etc.) + self.route_packet(stream, packet, connection_id.clone()).await?; + } + Err(e) => warn!("Failed to parse packet: {}", e), + } + + pool.release(buffer).await; + } + + if let Some(state) = self.connection_service.get_connection(&connection_id) { + let session_id = state.session_id.unwrap_or_default(); + if !session_id.is_empty() { + let mut auth_client = self.auth_client.lock().await; + auth_client.logout(&session_id).await?; + } else { + warn!("No session found for {}", stream.peer_addr()?); + } + } + ACTIVE_CONNECTIONS.dec(); + Ok(()) + } + + pub async fn route_packet(&self, stream: &mut TcpStream, packet: Packet, connection_id: String) -> Result<(), Box> { + debug!("Routing packet: {:?}", packet); + match packet.packet_type { + PacketType::PakcsAlive => Ok(()), + // Generic Server Packets + 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, + // Login Packets + PacketType::PakcsLoginTokenReq => auth::handle_login_req(stream, packet, self.auth_client.clone(), self.connection_service.clone(), connection_id, stream.peer_addr()?).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, + + // 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, + + // 1 => chat::handle_chat(packet).await?, + // 2 => movement::handle_movement(packet).await?, + _ => { + warn!("Unhandled packet type: {:?}", packet.packet_type); + Ok(()) + }, + } } }