- add: logout grpc function

- add: logout packet handler
- add: connection state and service for storing connection data
- add: session service calls to auth-service
- fix: compile error on database service due to moved redis cache

packet-service/src/main.rs

@@ -18,6 +18,7 @@ use tracing::{debug, error, info, warn};
 use utils::consul_registration;
 use utils::service_discovery::get_service_address;
 use warp::Filter;
+use crate::connection_service::ConnectionService;
 
 mod packet_type;
 mod packet;
@@ -30,6 +31,9 @@ mod handlers;
 mod bufferpool;
 mod metrics;
 mod auth_client;
+mod connection_state;
+mod connection_service;
+
 pub mod auth {
     tonic::include_proto!("auth"); // Path matches the package name in auth.proto
 }
@@ -38,7 +42,7 @@ const BUFFER_POOL_SIZE: usize = 1000;
 const MAX_CONCURRENT_CONNECTIONS: usize = 100;
 
 
-async fn handle_connection(stream: &mut TcpStream, pool: Arc<BufferPool>, auth_client: Arc<Mutex<AuthClient>>) -> Result<(), Box<dyn Error + Send + Sync>> {
+async fn handle_connection(stream: &mut TcpStream, pool: Arc<BufferPool>, auth_client: Arc<Mutex<AuthClient>>, connection_service: Arc<ConnectionService>, connection_id: String) -> Result<(), Box<dyn Error + Send + Sync>> {
     ACTIVE_CONNECTIONS.inc();
     while let Some(mut buffer) = pool.acquire().await {
         // Read data into the buffer
@@ -53,13 +57,19 @@ async fn handle_connection(stream: &mut TcpStream, pool: Arc<BufferPool>, auth_c
             Ok(packet) => {
                 debug!("Parsed Packet: {:?}", packet);
                 // Handle the parsed packet (route it, process it, etc.)
-                router::route_packet(stream, packet, auth_client.clone()).await?;
+                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();
+        let mut auth_client = auth_client.lock().await;
+        auth_client.logout(&session_id).await?;
+    }
     ACTIVE_CONNECTIONS.dec();
     Ok(())
 }
@@ -113,12 +123,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         let semaphore = Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTIONS));
         let listener = TcpListener::bind(full_addr.clone()).await.unwrap();
         let buffer_pool = BufferPool::new(BUFFER_POOL_SIZE);
+        let connection_service = Arc::new(ConnectionService::new());
 
         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();
             info!("New connection from {}", addr);
 
             let pool = buffer_pool.clone();
@@ -127,9 +139,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             // Spawn a new task for each connection
             tokio::spawn(async move {
                 let _permit = permit;
-                if let Err(e) = handle_connection(&mut socket, pool, auth_client).await {
+                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 {
                     error!("Error handling connection: {}", e);
                 }
+                connection_service.remove_connection(&connection_id);
             });
         }
     });