From ad6ba2c8e6d193e66d58a8f92d3d4d3b9a5d40af9bff041fb72ff9ad6de2962f Mon Sep 17 00:00:00 2001 From: raven <7156279+RavenX8@users.noreply.github.com> Date: Fri, 6 Jun 2025 17:52:29 -0400 Subject: [PATCH] More work. Added chat service Updated packet service to pass the tcp stream around in a Arc type. Updated character position data to not require multiplying the coords Added more debug logs Added an interceptor for gRPC comms with the chat server Updated build and push script for the chat server changes --- Cargo.toml | 1 + character-service/src/character_db_client.rs | 4 +- charts/osirose-new/values.yaml | 18 ++- chat-service/Cargo.toml | 22 +++ chat-service/Dockerfile | 25 ++++ chat-service/build.rs | 16 +++ chat-service/src/chat_channels/guild_chat.rs | 22 +++ chat-service/src/chat_channels/local_chat.rs | 22 +++ chat-service/src/chat_channels/mod.rs | 12 ++ chat-service/src/chat_channels/shout_chat.rs | 20 +++ chat-service/src/chat_service.rs | 130 +++++++++++++++++ chat-service/src/main.rs | 51 +++++++ database-service/Cargo.toml | 1 + database-service/src/sessions.rs | 9 +- packet-service/Cargo.toml | 2 + packet-service/build.rs | 1 + packet-service/src/connection_state.rs | 21 ++- packet-service/src/handlers/auth.rs | 100 ++++++++----- packet-service/src/handlers/character.rs | 60 +++++--- packet-service/src/handlers/chat.rs | 136 ++++++++++++++++++ packet-service/src/handlers/chat_client.rs | 91 ++++++++++++ packet-service/src/handlers/mod.rs | 2 + packet-service/src/handlers/world.rs | 12 +- .../src/interceptors/auth_interceptor.rs | 20 +++ packet-service/src/interceptors/mod.rs | 1 + packet-service/src/lib.rs | 4 + packet-service/src/main.rs | 10 +- packet-service/src/router.rs | 41 +++--- proto/chat.proto | 9 +- scripts/build_and_push.py | 6 +- utils/src/service_discovery.rs | 3 +- world-service/src/main.rs | 7 + 32 files changed, 787 insertions(+), 92 deletions(-) create mode 100644 chat-service/Cargo.toml create mode 100644 chat-service/Dockerfile create mode 100644 chat-service/build.rs create mode 100644 chat-service/src/chat_channels/guild_chat.rs create mode 100644 chat-service/src/chat_channels/local_chat.rs create mode 100644 chat-service/src/chat_channels/mod.rs create mode 100644 chat-service/src/chat_channels/shout_chat.rs create mode 100644 chat-service/src/chat_service.rs create mode 100644 chat-service/src/main.rs create mode 100644 packet-service/src/handlers/chat.rs create mode 100644 packet-service/src/handlers/chat_client.rs create mode 100644 packet-service/src/interceptors/auth_interceptor.rs create mode 100644 packet-service/src/interceptors/mod.rs diff --git a/Cargo.toml b/Cargo.toml index 37717d4..05ed338 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ resolver = "2" members = [ "auth-service", + "chat-service", "character-service", "database-service", "packet-service", diff --git a/character-service/src/character_db_client.rs b/character-service/src/character_db_client.rs index 3876dcf..88d92ca 100644 --- a/character-service/src/character_db_client.rs +++ b/character-service/src/character_db_client.rs @@ -209,8 +209,8 @@ impl CharacterDbClient { }; let position = Position { map_id: 20, - x: 5200.00, - y: 5200.00, + x: 520000.00, + y: 520000.00, z: 1.0, spawn_id: 1, }; diff --git a/charts/osirose-new/values.yaml b/charts/osirose-new/values.yaml index 5a8e8d5..c24d266 100644 --- a/charts/osirose-new/values.yaml +++ b/charts/osirose-new/values.yaml @@ -6,7 +6,7 @@ autoscaling: global: env: - LOG_LEVEL: "debug" + LOG_LEVEL: "info" APP_ENV: "dev" DATABASE_URL: "" # This is a placeholder. Will be dynamically constructed REDIS_URL: "redis://valkey:6379/0" @@ -26,6 +26,20 @@ services: targetPort: 50051 protocol: TCP + - name: chat-service + replicas: 1 + serviceAccount: azgstudio-serviceaccount + image: chat-service:latest + port: 50055 + env: + SERVICE_PORT: 50055 + LOG_LEVEL: "debug" + service: + portName: chat-service + port: 50055 + targetPort: 50055 + protocol: TCP + - name: character-service replicas: 1 serviceAccount: azgstudio-serviceaccount @@ -61,6 +75,7 @@ services: port: 29000 env: SERVICE_PORT: 29000 + LOG_LEVEL: "debug" service: type: LoadBalancer portName: game-packet-service @@ -75,6 +90,7 @@ services: port: 50054 env: SERVICE_PORT: 50054 + LOG_LEVEL: "debug" service: annotations: name: "Athena" diff --git a/chat-service/Cargo.toml b/chat-service/Cargo.toml new file mode 100644 index 0000000..f3a9aa2 --- /dev/null +++ b/chat-service/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "chat-service" +version = "0.1.0" +edition = "2021" + +[dependencies] +utils = { path = "../utils" } +dotenv = "0.15" +tokio = { version = "1.41.1", features = ["full"] } +serde = { version = "1.0", features = ["derive"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3.19", features = ["env-filter", "chrono"] } +tonic = "0.12.3" +prost = "0.13.4" +warp = "0.3.7" +tonic-health = "0.12.3" +futures = "0.3.31" +uuid = "1.15.1" +tokio-stream = "0.1.17" + +[build-dependencies] +tonic-build = "0.12.3" \ No newline at end of file diff --git a/chat-service/Dockerfile b/chat-service/Dockerfile new file mode 100644 index 0000000..8c17f5a --- /dev/null +++ b/chat-service/Dockerfile @@ -0,0 +1,25 @@ +FROM rust:alpine AS builder +LABEL authors="raven" + +RUN apk add --no-cache musl-dev libressl-dev protobuf-dev + +WORKDIR /usr/src/utils +COPY ./utils . + +WORKDIR /usr/src/proto +COPY ./proto . + +WORKDIR /usr/src/chat-service +COPY ./chat-service . + +RUN cargo build --release + +FROM alpine:3 + +RUN apk add --no-cache libssl3 libgcc + +COPY --from=builder /usr/src/chat-service/target/release/chat-service /usr/local/bin/chat-service + +EXPOSE 50054 + +CMD ["chat-service"] \ No newline at end of file diff --git a/chat-service/build.rs b/chat-service/build.rs new file mode 100644 index 0000000..9b6fa18 --- /dev/null +++ b/chat-service/build.rs @@ -0,0 +1,16 @@ +fn main() { + // gRPC Server code + tonic_build::configure() + .build_server(true) // Generate gRPC server code + .compile_well_known_types(true) + .type_attribute(".", "#[derive(serde::Serialize, serde::Deserialize)]") + .compile_protos(&["../proto/chat.proto"], &["../proto"]) + .unwrap_or_else(|e| panic!("Failed to compile protos {:?}", e)); + + // gRPC Client code + // tonic_build::configure() + // .build_server(false) // Generate gRPC client code + // .compile_well_known_types(true) + // .compile_protos(&["../proto/user_db_api.proto", "../proto/auth.proto"], &["../proto"]) + // .unwrap_or_else(|e| panic!("Failed to compile protos {:?}", e)); +} diff --git a/chat-service/src/chat_channels/guild_chat.rs b/chat-service/src/chat_channels/guild_chat.rs new file mode 100644 index 0000000..49332cc --- /dev/null +++ b/chat-service/src/chat_channels/guild_chat.rs @@ -0,0 +1,22 @@ +use crate::chat_channels::ChatChannel; +use crate::chat_service::Clients; +use crate::chat_service::chat::ChatMessage; + +/// A dedicated module for guild chat. +#[derive(Debug)] +pub struct GuildChat; + +impl ChatChannel for GuildChat { + fn handle_message(&self, message: ChatMessage, sender_id: &str, clients: &Clients) { + // This is a placeholder. In a real implementation, verify + // guild membership by consulting your Character or Guild service. + let clients_lock = clients.lock().unwrap(); + for (id, tx) in clients_lock.iter() { + // For demonstration, send only to clients whose IDs contain + // "guild". Replace this logic with your actual membership check. + if id != sender_id && id.contains("guild") { + let _ = tx.try_send(message.clone()); + } + } + } +} \ No newline at end of file diff --git a/chat-service/src/chat_channels/local_chat.rs b/chat-service/src/chat_channels/local_chat.rs new file mode 100644 index 0000000..778091c --- /dev/null +++ b/chat-service/src/chat_channels/local_chat.rs @@ -0,0 +1,22 @@ +use tracing::debug; +use crate::chat_channels::ChatChannel; +use crate::chat_service::Clients; +use crate::chat_service::chat::ChatMessage; + +/// A dedicated module for local chat. +#[derive(Debug)] +pub struct LocalChat; + +impl ChatChannel for LocalChat { + fn handle_message(&self, message: ChatMessage, sender_id: &str, clients: &Clients) { + // In a full implementation, you might query for nearby clients. + // For demo purposes, we simply broadcast to all clients except the sender. + debug!("LocalChat::handle_message: {:?}", message); + let clients_lock = clients.lock().unwrap(); + for (id, tx) in clients_lock.iter() { + // if id != sender_id { + let _ = tx.try_send(message.clone()); + // } + } + } +} \ No newline at end of file diff --git a/chat-service/src/chat_channels/mod.rs b/chat-service/src/chat_channels/mod.rs new file mode 100644 index 0000000..cff6da3 --- /dev/null +++ b/chat-service/src/chat_channels/mod.rs @@ -0,0 +1,12 @@ +pub mod local_chat; +pub mod shout_chat; +pub mod guild_chat; + +/// Define a common trait for all dedicated chat channels. +use crate::chat_service::Clients; +use crate::chat_service::chat::ChatMessage; + +pub trait ChatChannel: Send + Sync { + /// Process and route the chat message. + fn handle_message(&self, message: ChatMessage, sender_id: &str, clients: &Clients); +} \ No newline at end of file diff --git a/chat-service/src/chat_channels/shout_chat.rs b/chat-service/src/chat_channels/shout_chat.rs new file mode 100644 index 0000000..437e068 --- /dev/null +++ b/chat-service/src/chat_channels/shout_chat.rs @@ -0,0 +1,20 @@ +use crate::chat_channels::ChatChannel; +use crate::chat_service::Clients; +use crate::chat_service::chat::ChatMessage; + +/// A dedicated module for shout chat. +#[derive(Debug)] +pub struct ShoutChat; + +impl ChatChannel for ShoutChat { + fn handle_message(&self, message: ChatMessage, sender_id: &str, clients: &Clients) { + // For demo purposes, we simply broadcast to all clients except the sender. + // TODO: make sure the clients are on the same map + let clients_lock = clients.lock().unwrap(); + for (id, tx) in clients_lock.iter() { + if id != sender_id { + let _ = tx.try_send(message.clone()); + } + } + } +} \ No newline at end of file diff --git a/chat-service/src/chat_service.rs b/chat-service/src/chat_service.rs new file mode 100644 index 0000000..6e398d4 --- /dev/null +++ b/chat-service/src/chat_service.rs @@ -0,0 +1,130 @@ +use futures::{Stream, StreamExt}; +use std::collections::HashMap; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use tonic::{Request, Response, Status}; +use tracing::debug; + +pub mod common { + tonic::include_proto!("common"); +} +pub mod chat { + tonic::include_proto!("chat"); +} + +use chat::chat_service_server::{ChatService, ChatServiceServer}; +use chat::{ChatMessage, MessageType}; + +/// Type alias for storing client connections. +pub type Clients = +Arc>>>; + +use crate::chat_channels::ChatChannel; + +/// Our chat service struct. +#[derive(Clone)] +pub struct MyChatService { + pub clients: Clients, + pub local_channel: Arc, + pub shout_channel: Arc, + pub guild_channel: Arc, +} + +impl MyChatService { + /// Wrap our service as a gRPC service. + pub fn into_service(self) -> ChatServiceServer { + ChatServiceServer::new(self) + } +} + +#[tonic::async_trait] +impl ChatService for MyChatService { + type ChatStreamStream = + Pin> + Send + Sync + 'static>>; + + async fn chat_stream( + &self, + request: Request>, + ) -> Result, Status> { + debug!("New chat client connected"); + debug!("request: {:?}", request); + + let mut inbound = request.into_inner(); + + // Create a new client ID. In production, use authenticated IDs. + let client_id = format!("client-{}", uuid::Uuid::new_v4()); + + // Create a channel for sending outbound messages to this client. + let (tx, rx) = tokio::sync::mpsc::channel(32); + + { + let mut clients = self.clients.lock().unwrap(); + clients.insert(client_id.clone(), tx); + } + + // Clone shared resources for the spawned task. + let clients_clone = self.clients.clone(); + let local_channel = self.local_channel.clone(); + let shout_channel = self.shout_channel.clone(); + let guild_channel = self.guild_channel.clone(); + + tokio::spawn(async move { + while let Some(result) = inbound.next().await { + match result { + Ok(message) => { + debug!("message: {:?}", message); + // Dispatch based on the chat type. + match TryFrom::try_from(message.r#type) + .unwrap_or(MessageType::Normal) + { + MessageType::Normal => { + local_channel.handle_message( + message, + &client_id, + &clients_clone, + ); + } + MessageType::Shout => { + shout_channel.handle_message( + message, + &client_id, + &clients_clone, + ); + } + MessageType::Clan => { + guild_channel.handle_message( + message, + &client_id, + &clients_clone, + ); + } + // For other types, we simply broadcast as default. + _ => { + let clients_lock = clients_clone.lock().unwrap(); + for (id, tx) in clients_lock.iter() { + if id != &client_id { + let _ = tx.try_send(message.clone()); + } + } + } + } + } + Err(e) => { + eprintln!("Error receiving message from {}: {:?}", client_id, e); + break; + } + } + } + + // Remove the client when the stream ends. + let mut clients = clients_clone.lock().unwrap(); + clients.remove(&client_id); + println!("Client {} disconnected", client_id); + }); + + // Convert the rx half into a stream for the response. + let out_stream = tokio_stream::wrappers::ReceiverStream::new(rx) + .map(|msg| Ok(msg)); + Ok(Response::new(Box::pin(out_stream) as Self::ChatStreamStream)) + } +} \ No newline at end of file diff --git a/chat-service/src/main.rs b/chat-service/src/main.rs new file mode 100644 index 0000000..082d278 --- /dev/null +++ b/chat-service/src/main.rs @@ -0,0 +1,51 @@ +mod chat_service; +mod chat_channels; + +use dotenv::dotenv; +use std::env; +use utils::service_discovery::{get_kube_service_endpoints_by_dns, get_service_endpoints_by_dns}; +use utils::{health_check, logging}; +use chat_service::MyChatService; +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; +use tonic::transport::Server; +use crate::chat_service::chat::chat_service_server::ChatServiceServer; + +#[tokio::main] +async fn main() -> Result<(), Box> { + dotenv().ok(); + let app_name = env!("CARGO_PKG_NAME"); + logging::setup_logging(app_name, &["chat_service", "health_check"]); + + // Set the gRPC server address + let addr = env::var("LISTEN_ADDR").unwrap_or_else(|_| "0.0.0.0".to_string()); + let port = env::var("SERVICE_PORT").unwrap_or_else(|_| "50055".to_string()); + + let clients = Arc::new(Mutex::new(HashMap::new())); + + let chat_service = MyChatService { + clients: clients.clone(), + local_channel: Arc::new(chat_channels::local_chat::LocalChat), + shout_channel: Arc::new(chat_channels::shout_chat::ShoutChat), + guild_channel: Arc::new(chat_channels::guild_chat::GuildChat), + }; + + // Register service with Consul + let (mut health_reporter, health_service) = tonic_health::server::health_reporter(); + health_reporter + .set_serving::>() + .await; + let address = SocketAddr::new(addr.parse()?, port.parse()?); + tokio::spawn( + Server::builder() + .add_service(chat_service.into_service()) + .add_service(health_service) + .serve(address), + ); + + println!("Chat Service listening on {}", address); + + utils::signal_handler::wait_for_signal().await; + Ok(()) +} diff --git a/database-service/Cargo.toml b/database-service/Cargo.toml index 6b84728..2321360 100644 --- a/database-service/Cargo.toml +++ b/database-service/Cargo.toml @@ -21,6 +21,7 @@ serde_json = "1.0.133" async-trait = "0.1.83" utils = { path = "../utils" } tonic-health = "0.12.3" +log = "0.4.26" [build-dependencies] tonic-build = "0.12.3" diff --git a/database-service/src/sessions.rs b/database-service/src/sessions.rs index 673664e..5498063 100644 --- a/database-service/src/sessions.rs +++ b/database-service/src/sessions.rs @@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize}; use sqlx::{FromRow, Row}; use std::sync::Arc; use tokio::sync::Mutex; -use tracing::debug; +use log::debug; use utils::redis_cache::{Cache, RedisCache}; #[derive(Debug, FromRow, Serialize, Deserialize)] @@ -24,6 +24,8 @@ impl SessionRepository { pub async fn get_session(&self, session_id: &str) -> Result { let cache_key = format!("session:{}", session_id); + debug!("get_session: {:?}", session_id); + if let Some(session) = self .cache .lock() @@ -32,10 +34,13 @@ impl SessionRepository { .await .map_err(|_| sqlx::Error::RowNotFound)? { + debug!("Found session in cache: {:?}", session); return Ok(session); } - // Fetch from database + debug!("Session not found in cache, fetching from database"); + + // Fetch from the database let session = sqlx::query_as::<_, Session>("SELECT id, \"userId\" as user_id FROM session WHERE id = $1") .bind(session_id) .fetch_one(&self.pool) diff --git a/packet-service/Cargo.toml b/packet-service/Cargo.toml index 33edbd3..7ca8fa1 100644 --- a/packet-service/Cargo.toml +++ b/packet-service/Cargo.toml @@ -36,6 +36,8 @@ dashmap = "6.1.0" uuid = { version = "1.11.0", features = ["v4"] } chrono = "0.4.39" prometheus_exporter = "0.8.5" +futures = "0.3.31" +tokio-stream = "0.1.17" [build-dependencies] tonic-build = "0.12.3" diff --git a/packet-service/build.rs b/packet-service/build.rs index feec78b..6a69942 100644 --- a/packet-service/build.rs +++ b/packet-service/build.rs @@ -7,6 +7,7 @@ fn main() { .compile_protos( &[ "../proto/auth.proto", + "../proto/chat.proto", "../proto/character.proto", "../proto/character_common.proto", ], diff --git a/packet-service/src/connection_state.rs b/packet-service/src/connection_state.rs index a98306c..dd496cb 100644 --- a/packet-service/src/connection_state.rs +++ b/packet-service/src/connection_state.rs @@ -1,12 +1,17 @@ use std::collections::HashMap; +use std::sync::Arc; +use std::fmt; -#[derive(Clone, Debug)] +use crate::handlers::chat_client::ChatClientHandler; + +#[derive(Clone)] pub struct ConnectionState { pub user_id: Option, pub session_id: Option, pub character_id: Option, pub character_list: Option>, pub additional_data: HashMap, // Flexible data storage + pub chat_handler: Option>, } impl ConnectionState { @@ -17,6 +22,20 @@ impl ConnectionState { character_id: None, character_list: None, additional_data: HashMap::new(), + chat_handler: None, } } } + +impl fmt::Debug for ConnectionState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ConnectionState") + .field("user_id", &self.user_id) + .field("session_id", &self.session_id) + .field("character_id", &self.character_id) + .field("character_list", &self.character_list) + .field("additional_data", &self.additional_data) + .field("chat_handler", &self.chat_handler.as_ref().map(|_| "")) + .finish() + } +} \ No newline at end of file diff --git a/packet-service/src/handlers/auth.rs b/packet-service/src/handlers/auth.rs index fa9d6c2..b642798 100644 --- a/packet-service/src/handlers/auth.rs +++ b/packet-service/src/handlers/auth.rs @@ -2,16 +2,6 @@ use crate::auth_client::AuthClient; use crate::connection_service::ConnectionService; use crate::packet::{send_packet, Packet, PacketPayload}; use crate::packet_type::PacketType; -use crate::packets::cli_channel_list_req::CliChannelListReq; -use crate::packets::cli_join_server_token_req::CliJoinServerTokenReq; -use crate::packets::cli_login_token_req::CliLoginTokenReq; -use crate::packets::cli_srv_select_req::CliSrvSelectReq; -use crate::packets::srv_accept_reply::SrvAcceptReply; -use crate::packets::srv_channel_list_reply::{ChannelInfo, SrvChannelListReply}; -use crate::packets::srv_join_server_reply::SrvJoinServerReply; -use crate::packets::srv_login_reply::{ServerInfo, SrvLoginReply}; -use crate::packets::srv_logout_reply::SrvLogoutReply; -use crate::packets::srv_srv_select_reply::SrvSrvSelectReply; use crate::packets::*; use std::collections::HashMap; use std::env; @@ -26,9 +16,11 @@ use tracing::{debug, error, info, warn}; use utils::null_string::NullTerminatedString; use utils::service_discovery; use utils::service_discovery::{get_kube_service_endpoints_by_dns, get_service_info}; +use crate::handlers::chat::create_chat_client_handler; +use crate::handlers::chat_client::ChatClientHandler; pub(crate) async fn handle_alive_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, auth_client: Arc>, connection_service: Arc, @@ -50,26 +42,30 @@ pub(crate) async fn handle_alive_req( } pub(crate) async fn handle_accept_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, ) -> Result<(), Box> { + use crate::packets::srv_accept_reply::SrvAcceptReply; let data = SrvAcceptReply { result: srv_accept_reply::Result::Accepted, rand_value: 0, }; let response_packet = Packet::new(PacketType::PakssAcceptReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; Ok(()) } pub(crate) async fn handle_join_server_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, auth_client: Arc>, connection_service: Arc, connection_id: String, ) -> Result<(), Box> { + use crate::packets::cli_join_server_token_req::CliJoinServerTokenReq; + use crate::packets::srv_join_server_reply::SrvJoinServerReply; let request = CliJoinServerTokenReq::decode(packet.payload.as_slice()); debug!("{:?}", request); @@ -86,7 +82,8 @@ pub(crate) async fn handle_join_server_req( pay_flag: 0, }; let response_packet = Packet::new(PacketType::PakscJoinServerReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; return Err("Session not valid".into()); } @@ -96,7 +93,8 @@ pub(crate) async fn handle_join_server_req( pay_flag: 0, }; let response_packet = Packet::new(PacketType::PakscJoinServerReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; Ok(()) } else { Err("Unable to find connection state".into()) @@ -104,12 +102,13 @@ pub(crate) async fn handle_join_server_req( } pub(crate) async fn handle_logout_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, auth_client: Arc>, connection_service: Arc, connection_id: String, ) -> Result<(), Box> { + use crate::packets::srv_logout_reply::SrvLogoutReply; if let Some(mut state) = connection_service.get_connection(&connection_id) { let session_id = state.session_id.clone().unwrap(); let mut auth_client = auth_client.lock().await; @@ -117,8 +116,9 @@ pub(crate) async fn handle_logout_req( let data = SrvLogoutReply { wait_time: 1 }; let response_packet = Packet::new(PacketType::PakwcLogoutReply, &data)?; - send_packet(stream, &response_packet).await?; - stream.shutdown().await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; + locked_stream.shutdown().await?; Ok(()) } else { Err("Unable to find connection state".into()) @@ -126,13 +126,14 @@ pub(crate) async fn handle_logout_req( } pub(crate) async fn handle_login_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, auth_client: Arc>, connection_service: Arc, connection_id: String, - addr: SocketAddr, ) -> Result<(), Box> { + use crate::packets::cli_login_token_req::CliLoginTokenReq; + use crate::packets::srv_login_reply::{ServerInfo, SrvLoginReply}; debug!("decoding packet payload of size {}", packet.payload.as_slice().len()); let data = CliLoginTokenReq::decode(packet.payload.as_slice())?; debug!("{:?}", data); @@ -140,6 +141,7 @@ pub(crate) async fn handle_login_req( let mut auth_client = auth_client.lock().await; match auth_client.validate_session(&data.token.0).await { Ok(response) => { + debug!("Response: {:?}", response); if response.valid == false { info!("Login failed: Invalid credentials"); @@ -150,14 +152,32 @@ pub(crate) async fn handle_login_req( servers_info: Vec::new(), }; let response_packet = Packet::new(PacketType::PaklcLoginReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; } else { debug!("Successfully logged in"); if let Some(mut state) = connection_service.get_connection_mut(&connection_id) { - debug!("Response: {:?}", response); state.user_id = Some(response.user_id); - state.session_id = Some(response.session_id); + state.session_id = Some(response.session_id.clone()); + } + + let chat_url = format!( + "http://{}", + get_kube_service_endpoints_by_dns("chat-service", "tcp", "chat-service") + .await + .expect("Failed to get chat service endpoints") + .get(0) + .unwrap() + ); + + let handler = ChatClientHandler::new(chat_url, connection_id.clone(), response.session_id.clone()).await?; + let chat_handler = Arc::new(handler); + + create_chat_client_handler(stream.clone(), chat_handler.clone()).await?; + + if let Some(mut state) = connection_service.get_connection_mut(&connection_id) { + state.chat_handler = Some(chat_handler); } let mut id = 0; @@ -196,7 +216,8 @@ pub(crate) async fn handle_login_req( servers_info: server_info, }; let response_packet = Packet::new(PacketType::PaklcLoginReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; } } Err(err) => { @@ -208,7 +229,8 @@ pub(crate) async fn handle_login_req( servers_info: Vec::new(), }; let response_packet = Packet::new(PacketType::PaklcLoginReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; return Ok(()); } } @@ -227,7 +249,8 @@ pub(crate) async fn handle_login_req( servers_info: Vec::new(), }; let response_packet = Packet::new(PacketType::PaklcLoginReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; } Code::Unavailable => { warn!("Login failed: Service is unavailable"); @@ -238,7 +261,8 @@ pub(crate) async fn handle_login_req( servers_info: Vec::new(), }; let response_packet = Packet::new(PacketType::PaklcLoginReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; } _ => { error!("Unexpected error: {}", tonic_status.message()); @@ -249,7 +273,8 @@ pub(crate) async fn handle_login_req( servers_info: Vec::new(), }; let response_packet = Packet::new(PacketType::PaklcLoginReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; } } } @@ -260,11 +285,13 @@ pub(crate) async fn handle_login_req( } pub(crate) async fn handle_server_select_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, connection_service: Arc, connection_id: String, ) -> Result<(), Box> { + use crate::packets::cli_srv_select_req::CliSrvSelectReq; + use crate::packets::srv_srv_select_reply::SrvSrvSelectReply; let request = CliSrvSelectReq::decode(packet.payload.as_slice())?; debug!("{:?}", request); @@ -286,14 +313,17 @@ pub(crate) async fn handle_server_select_req( }; let response_packet = Packet::new(PacketType::PaklcSrvSelectReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; Ok(()) } pub(crate) async fn handle_channel_list_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, ) -> Result<(), Box> { + use crate::packets::cli_channel_list_req::CliChannelListReq; + use crate::packets::srv_channel_list_reply::{ChannelInfo, SrvChannelListReply}; let request = CliChannelListReq::decode(packet.payload.as_slice()); debug!("{:?}", request); @@ -326,7 +356,8 @@ pub(crate) async fn handle_channel_list_req( channels: channel_info, }; let response_packet = Packet::new(PacketType::PaklcChannelListReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; } } Err(err) => { @@ -335,7 +366,8 @@ pub(crate) async fn handle_channel_list_req( channels: Vec::new(), }; let response_packet = Packet::new(PacketType::PaklcChannelListReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; return Ok(()); } } diff --git a/packet-service/src/handlers/character.rs b/packet-service/src/handlers/character.rs index 4734c43..6658781 100644 --- a/packet-service/src/handlers/character.rs +++ b/packet-service/src/handlers/character.rs @@ -17,7 +17,7 @@ use tonic::{Code, Status}; use tracing::{debug, error, info, warn}; use utils::null_string::NullTerminatedString; -fn string_to_u32(s: &str) -> u32 { +pub(crate) fn string_to_u32(s: &str) -> u32 { let mut hasher = DefaultHasher::new(); s.hash(&mut hasher); // Convert the 64-bit hash to a 32-bit number. @@ -68,7 +68,7 @@ pub(crate) fn convert_type_to_body_part(slot: i32) -> ItemType { } pub(crate) async fn handle_char_list_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, character_client: Arc>, connection_service: Arc, @@ -91,6 +91,8 @@ pub(crate) async fn handle_char_list_req( let character_list = character_client.get_character_list(&user_id).await?; let mut characters = vec![]; let mut character_id_list: Vec = Vec::new(); + + // Build the visible inventory for character in character_list.characters { let mut item_list: [EquippedItem; (MAX_VISIBLE_ITEMS as usize)] = core::array::from_fn(|i| EquippedItem::default()); @@ -128,13 +130,14 @@ pub(crate) async fn handle_char_list_req( let data = SrvCharListReply { characters }; let response_packet = Packet::new(PacketType::PakccCharListReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; Ok(()) } pub(crate) async fn handle_create_char_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, character_client: Arc>, connection_service: Arc, @@ -176,13 +179,14 @@ pub(crate) async fn handle_create_char_req( let data = SrvCreateCharReply { result, platininum: 0 }; let response_packet = Packet::new(PacketType::PakccCreateCharReply, &data)?; - send_packet(stream, &response_packet).await?; - + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; + Ok(()) } pub(crate) async fn handle_delete_char_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, character_client: Arc>, connection_service: Arc, @@ -218,13 +222,14 @@ pub(crate) async fn handle_delete_char_req( name: character_name, }; let response_packet = Packet::new(PacketType::PakccDeleteCharReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; Ok(()) } pub(crate) async fn handle_select_char_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, character_client: Arc>, connection_service: Arc, @@ -255,7 +260,10 @@ pub(crate) async fn handle_select_char_req( ip: NullTerminatedString("".to_string()), }; let response_packet = Packet::new(PacketType::PakccSwitchServer, &data)?; - send_packet(stream, &response_packet).await?; + { + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; + } let mut character_client = character_client.lock().await; let character_data = character_client @@ -275,12 +283,14 @@ pub(crate) async fn handle_select_char_req( core::array::from_fn(|i| EquippedItem::default()); let mut inventory: [srv_inventory_data::Item; (MAX_ITEMS as usize)] = core::array::from_fn(|i| srv_inventory_data::Item::default()); + + // Build the character learned skill list let mut skill_list: [u16; (MAX_SKILL_COUNT as usize)] = [0u16; MAX_SKILL_COUNT as usize]; - for index in 0..skills.len() { skill_list[index] = skills[index].id as u16; } + // Build the character inventory list for item in items { if item.slot < MAX_VISIBLE_ITEMS as i32 { let slot = convert_type_to_body_part(item.slot) as isize - 2; @@ -318,8 +328,8 @@ pub(crate) async fn handle_select_char_req( let data = SrvSelectCharReply { race: looks.race as u8, map: position.map_id as u16, - x: position.x * 100.0, - y: position.y * 100.0, + x: position.x, + y: position.y, spawn: position.spawn_id as u16, body_face: looks.face as u32, body_hair: looks.hair as u32, @@ -362,7 +372,10 @@ pub(crate) async fn handle_select_char_req( name, }; let response_packet = Packet::new(PacketType::PakwcSelectCharReply, &data)?; - send_packet(stream, &response_packet).await?; + { + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; + } // here we build the inventory let data = SrvInventoryData { @@ -370,7 +383,10 @@ pub(crate) async fn handle_select_char_req( items: inventory, }; let response_packet = Packet::new(PacketType::PakwcInventoryData, &data)?; - send_packet(stream, &response_packet).await?; + { + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; + } // Now we need to build the Quest data let mut quests: [srv_quest_data::Quest; (MAX_QUESTS as usize)] = @@ -388,15 +404,21 @@ pub(crate) async fn handle_select_char_req( wishlist, }; let response_packet = Packet::new(PacketType::PakwcQuestData, &data)?; - send_packet(stream, &response_packet).await?; - - // Send the billing message (we don't actually use this so we just send the defaults to allow) + { + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; + } + + // Send the billing message (we don't use this, so we just send the defaults to allow) let data = SrvBillingMessage { function_type: 0x1001, pay_flag: 2, }; let response_packet = Packet::new(PacketType::PakwcBillingMessage, &data)?; - send_packet(stream, &response_packet).await?; + { + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; + } Ok(()) } diff --git a/packet-service/src/handlers/chat.rs b/packet-service/src/handlers/chat.rs new file mode 100644 index 0000000..8a1ef64 --- /dev/null +++ b/packet-service/src/handlers/chat.rs @@ -0,0 +1,136 @@ +use crate::character_client::CharacterClient; +use crate::connection_service::ConnectionService; +use crate::packet::{send_packet, Packet, PacketPayload}; +use crate::packet_type::PacketType; +use chrono::{Local, Timelike}; +use std::error::Error; +use std::sync::Arc; +use tokio::net::TcpStream; +use tokio::sync::Mutex; +use tonic::transport::Channel; +use tracing::{debug, error}; +use utils::null_string::NullTerminatedString; +use utils::service_discovery::get_kube_service_endpoints_by_dns; +use crate::handlers::chat_client::chat::{ChatMessage, MessageType}; +use crate::handlers::chat_client::ChatClientHandler; + +pub async fn create_chat_client_handler( + stream_for_task: Arc>, + task_chat_handler: Arc, +) -> Result<(), Box> { + use crate::packets::srv_normal_chat::SrvNormalChat; + use crate::packets::srv_shout_chat::SrvShoutChat; + use crate::packets::srv_party_chat::SrvPartyChat; + use crate::packets::srv_whisper_chat::SrvWhisperChat; + use crate::packets::srv_clan_chat::SrvClanChat; + use crate::packets::srv_allied_chat::SrvAlliedChat; + tokio::spawn({ + async move { + debug!("Spawning chat handler task"); + loop { + let mut rx = task_chat_handler.inbound_rx.lock().await; + while let Some(chat_msg) = rx.recv().await { + debug!("Packet-Service received chat message: {} (client_id: {}, type {})", chat_msg.message, chat_msg.client_id, chat_msg.r#type); + + debug!("Locking stream"); + let mut locked_stream = stream_for_task.lock().await; + debug!("Locked stream"); + match chat_msg.r#type { + 1 => { + // Normal Chat + let data = SrvNormalChat { + char_id: chat_msg.client_id.parse().unwrap_or(696), + message: NullTerminatedString(chat_msg.message), + }; + + // Send the packet to the client + let response_packet = Packet::new(PacketType::PakwcNormalChat, &data); + debug!("Attempting to send normal chat to client"); + if let Err(e) = send_packet(&mut locked_stream, &response_packet.unwrap()).await + { + error!("unable to send normal chat: {:?}", e); + } + } + 2 => { + // Shout Chat + let data = SrvShoutChat { + sender: Default::default(), + message: NullTerminatedString(chat_msg.message), + }; + let response_packet = Packet::new(PacketType::PakwcShoutChat, &data); + debug!("Attempting to send shout chat to client"); + if let Err(e) = send_packet(&mut locked_stream, &response_packet.unwrap()).await + { + error!("unable to send shout chat: {:?}", e); + } + } + 3 => { + // Party Chat + } + 4 => { + // Whisper Chat + } + 5 => { + // Clan Chat + } + 6 => { + // Allied Chat + } + _ => { + // Normal Chat + let data = SrvNormalChat { + char_id: 0, + message: NullTerminatedString(chat_msg.message), + }; + + // Send the packet to the client + let response_packet = Packet::new(PacketType::PakwcNormalChat, &data); + if let Err(e) = send_packet(&mut locked_stream, &response_packet.unwrap()).await + { + error!("unable to send normal chat: {:?}", e); + } + } + } + } + } + debug!("Chat handler task exiting"); + } + }); + + Ok(()) +} + +pub(crate) async fn handle_normal_chat( + stream: Arc>, + packet: Packet, + connection_service: Arc, + connection_id: String, +) -> Result<(), Box> { + use crate::packets::cli_normal_chat::*; + use crate::packets::srv_normal_chat::*; + let request = CliNormalChat::decode(packet.payload.as_slice())?; + debug!("{:?}", request); + + if let Some(mut state) = connection_service.get_connection(&connection_id) { + let user_id = state.user_id.clone().expect("Missing user id in connection state"); + let message = ChatMessage { + client_id: crate::handlers::character::string_to_u32(&user_id).to_string(), + r#type: MessageType::Normal as i32, + message: request.message.clone().0, + target_id: "".to_string(), + }; + state.chat_handler.unwrap().send_message(message).await; + } + + + // We're not sending here because we should get a message back from the chat service + // let data = SrvNormalChat { + // char_id: 0, + // message: request.message, + // }; + // let response_packet = Packet::new(PacketType::PakwcNormalChat, &data)?; + // let mut locked_stream = stream.lock().await; + // send_packet(&mut locked_stream, &response_packet).await?; + + Ok(()) +} diff --git a/packet-service/src/handlers/chat_client.rs b/packet-service/src/handlers/chat_client.rs new file mode 100644 index 0000000..8ec123f --- /dev/null +++ b/packet-service/src/handlers/chat_client.rs @@ -0,0 +1,91 @@ +use tonic::{Request, transport::Channel}; +use futures::StreamExt; +use tokio::sync::{mpsc, Mutex}; +use tokio_stream::wrappers::ReceiverStream; +use std::error::Error; + +pub mod chat { + tonic::include_proto!("chat"); +} + +use chat::chat_service_client::ChatServiceClient; +use chat::ChatMessage; +use crate::interceptors::auth_interceptor::AuthInterceptor; + +/// ChatClientHandler encapsulates the bidirectional chat stream. +/// In addition to providing an API to send messages, it also spawns a +/// background task which forwards incoming chat messages through an inbound channel. +pub struct ChatClientHandler { + outbound_tx: mpsc::Sender, + /// Inbound messages from the chat service are sent here. + pub inbound_rx: Mutex>, +} + +impl ChatClientHandler { + /// Creates and returns a new ChatClientHandler. + /// + /// * `chat_url` - Full URL of the Chat Service (for example, "http://127.0.0.1:50051") + /// * `client_id` - The authenticated client ID to be injected into each request. + /// * `session_id` - The authenticated session token to be injected into each request. + pub async fn new( + chat_url: String, + client_id: String, + session_id: String, + ) -> Result> { + // Create a channel to the Chat Service. + let channel = Channel::from_shared(chat_url)?.connect().await + .map_err(|e| Box::new(e) as Box)?; + let interceptor = AuthInterceptor { client_id, session_id }; + + // Create ChatService client with interceptor. + let mut chat_client = ChatServiceClient::with_interceptor(channel, interceptor); + + // Create an mpsc channel for outbound messages. + let (out_tx, out_rx) = mpsc::channel(32); + let outbound_stream = ReceiverStream::new(out_rx); + + // This channel will be used to forward inbound messages to the packet-service. + let (in_tx, in_rx) = mpsc::channel(32); + + // Establish the bidirectional chat stream. + let request = Request::new(outbound_stream); + let mut response = chat_client.chat_stream(request).await + .map_err(|e| Box::new(e) as Box)?.into_inner(); + + // Spawn a task to continuously receive messages from the Chat Service. + // Each received message is sent through the 'in_tx' channel. + tokio::spawn(async move { + while let Some(result) = response.next().await { + match result { + Ok(chat_msg) => { + // You might translate or process the chat_msg here, + // then forward it to your packet-service logic. + if let Err(e) = in_tx.send(chat_msg).await { + eprintln!("Failed to forward chat message: {:?}", e); + break; + } + } + Err(e) => { + eprintln!("Error receiving chat stream message: {:?}", e); + break; + } + } + } + println!("Chat inbound stream closed"); + }); + + Ok(Self { + outbound_tx: out_tx, + inbound_rx: Mutex::new(in_rx), + }) + } + + /// Sends a chat message to the Chat Service. + pub async fn send_message( + &self, + message: ChatMessage, + ) -> Result<(), Box> { + self.outbound_tx.send(message).await?; + Ok(()) + } +} \ No newline at end of file diff --git a/packet-service/src/handlers/mod.rs b/packet-service/src/handlers/mod.rs index fb98300..97c1a36 100644 --- a/packet-service/src/handlers/mod.rs +++ b/packet-service/src/handlers/mod.rs @@ -1,3 +1,5 @@ pub mod auth; pub mod character; pub mod world; +pub mod chat; +pub mod chat_client; diff --git a/packet-service/src/handlers/world.rs b/packet-service/src/handlers/world.rs index cdcdb8d..72db0b0 100644 --- a/packet-service/src/handlers/world.rs +++ b/packet-service/src/handlers/world.rs @@ -7,7 +7,9 @@ use std::error::Error; use std::sync::Arc; use tokio::net::TcpStream; use tokio::sync::Mutex; +use tonic::transport::Channel; use tracing::debug; +use utils::service_discovery::get_kube_service_endpoints_by_dns; fn distance(x1: f64, y1: f64, x2: f64, y2: f64) -> u16 { let dist = ((x2 - x1).powi(2) + (y2 - y1).powi(2)).sqrt(); @@ -15,7 +17,7 @@ fn distance(x1: f64, y1: f64, x2: f64, y2: f64) -> u16 { } pub(crate) async fn handle_change_map_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, character_client: Arc>, connection_service: Arc, @@ -64,13 +66,14 @@ pub(crate) async fn handle_change_map_req( team_number: 10, }; let response_packet = Packet::new(PacketType::PakwcChangeMapReply, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; Ok(()) } pub(crate) async fn handle_mouse_cmd_req( - stream: &mut TcpStream, + stream: Arc>, packet: Packet, connection_service: Arc, connection_id: String, @@ -96,6 +99,7 @@ pub(crate) async fn handle_mouse_cmd_req( z: request.z, }; let response_packet = Packet::new(PacketType::PakwcMouseCmd, &data)?; - send_packet(stream, &response_packet).await?; + let mut locked_stream = stream.lock().await; + send_packet(&mut locked_stream, &response_packet).await?; Ok(()) } diff --git a/packet-service/src/interceptors/auth_interceptor.rs b/packet-service/src/interceptors/auth_interceptor.rs new file mode 100644 index 0000000..51fa3fd --- /dev/null +++ b/packet-service/src/interceptors/auth_interceptor.rs @@ -0,0 +1,20 @@ +use tonic::{Request, Status, service::Interceptor}; + +#[derive(Clone, Debug)] +pub struct AuthInterceptor { + pub client_id: String, + pub session_id: String, +} + +impl Interceptor for AuthInterceptor { + fn call(&mut self, mut request: Request<()>) -> Result, Status> { + // Attach the authenticated client ID into the metadata. + request + .metadata_mut() + .insert("x-client-id", self.client_id.parse().unwrap()); + request + .metadata_mut() + .insert("x-session-id", self.session_id.parse().unwrap()); + Ok(request) + } +} \ No newline at end of file diff --git a/packet-service/src/interceptors/mod.rs b/packet-service/src/interceptors/mod.rs new file mode 100644 index 0000000..ca05488 --- /dev/null +++ b/packet-service/src/interceptors/mod.rs @@ -0,0 +1 @@ +pub mod auth_interceptor; \ No newline at end of file diff --git a/packet-service/src/lib.rs b/packet-service/src/lib.rs index 1f8322d..be0a3e1 100644 --- a/packet-service/src/lib.rs +++ b/packet-service/src/lib.rs @@ -5,3 +5,7 @@ pub mod connection_state; pub mod metrics; pub mod packet; pub mod packet_type; +pub mod handlers { + pub mod chat_client; +} +pub mod interceptors; diff --git a/packet-service/src/main.rs b/packet-service/src/main.rs index e426fe7..9dc187d 100644 --- a/packet-service/src/main.rs +++ b/packet-service/src/main.rs @@ -39,18 +39,19 @@ mod packet_type; mod packets; mod router; mod types; +mod interceptors; pub mod common { tonic::include_proto!("common"); } pub mod auth { - tonic::include_proto!("auth"); // Path matches the package name in auth.proto + tonic::include_proto!("auth"); } pub mod character_common { - tonic::include_proto!("character_common"); // Path matches the package name in auth.proto + tonic::include_proto!("character_common"); } pub mod character { - tonic::include_proto!("character"); // Path matches the package name in auth.proto + tonic::include_proto!("character"); } const BUFFER_POOL_SIZE: usize = 1000; @@ -110,13 +111,14 @@ async fn main() -> Result<(), Box> { let pool = buffer_pool.clone(); let permit = semaphore.clone().acquire_owned().await.unwrap(); + let stream = Arc::new(Mutex::new(socket)); // Spawn a new task for each connection tokio::spawn(async move { let _permit = permit; let connection_id = packet_router.connection_service.add_connection(); if let Err(e) = packet_router - .handle_connection(&mut socket, pool, connection_id.clone()) + .handle_connection(stream, pool, connection_id.clone()) .await { error!("Error handling connection: {}", e); diff --git a/packet-service/src/router.rs b/packet-service/src/router.rs index 6a3414e..4e16059 100644 --- a/packet-service/src/router.rs +++ b/packet-service/src/router.rs @@ -23,24 +23,30 @@ pub struct PacketRouter { impl PacketRouter { pub async fn handle_connection( &self, - stream: &mut TcpStream, + stream: Arc>, pool: Arc, connection_id: String, ) -> Result<(), Box> { ACTIVE_CONNECTIONS.inc(); while let Some(mut buffer) = pool.acquire().await { // Read data into the buffer - let mut header_handle = stream.take(6); - let n = header_handle.read(&mut buffer).await?; - if n == 0 { - break; // Connection closed - } - let packet_size = u16::from_le_bytes(buffer[0..2].try_into()?) as usize; - if packet_size > 6 { - let mut body_handle = stream.take((packet_size - 6) as u64); - let n = body_handle.read(&mut buffer[6..]).await?; - if n == 0 { - break; // Connection closed + 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 + // } + 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 + // } } } @@ -52,7 +58,7 @@ impl PacketRouter { Ok(packet) => { debug!("Parsed Packet: {:?}", packet); // Handle the parsed packet (route it, process it, etc.) - self.route_packet(stream, packet, connection_id.clone()).await?; + self.route_packet(stream.clone(), packet, connection_id.clone()).await?; } Err(e) => warn!("Failed to parse packet: {}", e), } @@ -67,7 +73,8 @@ impl PacketRouter { let mut auth_client = self.auth_client.lock().await; auth_client.logout(&session_id).await?; } else { - warn!("No session found for {}", stream.peer_addr()?); + let mut locked_stream = stream.lock().await; + warn!("No session found for {}", locked_stream.peer_addr()?); } } ACTIVE_CONNECTIONS.dec(); @@ -77,7 +84,7 @@ impl PacketRouter { #[rustfmt::skip] pub async fn route_packet( &self, - stream: &mut TcpStream, + stream: Arc>, packet: Packet, connection_id: String, ) -> Result<(), Box> { @@ -88,7 +95,7 @@ impl PacketRouter { 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::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, @@ -103,6 +110,8 @@ impl PacketRouter { 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, + // Chat Packets + PacketType::PakcsNormalChat => chat::handle_normal_chat(stream, packet, self.connection_service.clone(), connection_id).await, // 1 => chat::handle_chat(packet).await?, // 2 => movement::handle_movement(packet).await?, _ => { diff --git a/proto/chat.proto b/proto/chat.proto index 79465b3..c2bc00b 100644 --- a/proto/chat.proto +++ b/proto/chat.proto @@ -5,7 +5,7 @@ package chat; import "common.proto"; service ChatService { - rpc SendMessage(ChatMessage) returns (common.Empty); + rpc ChatStream(stream ChatMessage) returns (stream ChatMessage); } enum MessageType { @@ -19,7 +19,8 @@ enum MessageType { } message ChatMessage { - MessageType type = 1; - string message = 2; - string target = 3; + string client_id = 1; + MessageType type = 2; + string message = 3; + string target_id = 4; } diff --git a/scripts/build_and_push.py b/scripts/build_and_push.py index 9ae754b..1a49c3d 100644 --- a/scripts/build_and_push.py +++ b/scripts/build_and_push.py @@ -3,21 +3,19 @@ import os # Define your images, tags, and Dockerfile paths images = [ - # "api-service", "auth-service", + "chat-service", "character-service", "database-service", "packet-service", - # "session-service", "world-service" ] dockerfile_paths = [ - # "../api-service/Dockerfile", "../auth-service/Dockerfile", + "../chat-service/Dockerfile", "../character-service/Dockerfile", "../database-service/Dockerfile", "../packet-service/Dockerfile", - # "../session-service/Dockerfile", "../world-service/Dockerfile", ] diff --git a/utils/src/service_discovery.rs b/utils/src/service_discovery.rs index b8782ae..6d3bfbf 100644 --- a/utils/src/service_discovery.rs +++ b/utils/src/service_discovery.rs @@ -43,6 +43,7 @@ pub async fn get_kube_service_endpoints_by_dns( service_protocol: &str, service_name: &str, ) -> Result, Box> { + debug!("Looking up service '{}'", service_name); let (config, options) = read_system_conf()?; let resolver = TokioAsyncResolver::tokio(config, options); @@ -61,7 +62,7 @@ pub async fn get_kube_service_endpoints_by_dns( ))?); } } - + debug!("Got endpoints: {:?}", endpoints); Ok(endpoints) } diff --git a/world-service/src/main.rs b/world-service/src/main.rs index 03f713d..d0c73dd 100644 --- a/world-service/src/main.rs +++ b/world-service/src/main.rs @@ -19,6 +19,13 @@ async fn main() -> Result<(), Box> { .get(0) .unwrap() ); + let chat_service = format!( + "http://{}", + get_kube_service_endpoints_by_dns("chat-service", "tcp", "chat-service") + .await? + .get(0) + .unwrap() + ); // Register service with Consul health_check::start_health_check(addr.as_str()).await?;