From 30cb6c5fa5e9f836f39f8f0ba0c0eb4b9fa44ae1 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Tue, 25 Nov 2025 22:33:29 +0000 Subject: [PATCH] Async Handling --- src/calls/call_manager.rs | 3 +- src/main.rs | 4 +- src/rho/client_connection.rs | 74 ++++++++++++++++++------------------ src/rho/rho_manager.rs | 1 - 4 files changed, 41 insertions(+), 41 deletions(-) diff --git a/src/calls/call_manager.rs b/src/calls/call_manager.rs index 04a5d40..9e0334c 100644 --- a/src/calls/call_manager.rs +++ b/src/calls/call_manager.rs @@ -33,7 +33,8 @@ pub async fn get_call_groups(user_id: Uuid) -> Vec> { } pub async fn get_call_token(user_id: Uuid, call_id: Uuid) -> Option { - let call_groups = CALL_GROUPS.read().await; + let call_groups = { CALL_GROUPS.read().await.clone() }; + for cg in call_groups.iter() { if cg.call_id == call_id { for member in cg.members.read().await.iter() { diff --git a/src/main.rs b/src/main.rs index 87e3f33..4fa79da 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,15 +6,14 @@ mod rho; mod util; use async_tungstenite::accept_hdr_async; +use dotenv::dotenv; use futures::StreamExt; -use livekit_api::services::room::{CreateRoomOptions, RoomClient}; use std::sync::Arc; use tokio::net::TcpListener; use tokio_util::compat::TokioAsyncReadCompatExt; use tungstenite::handshake::server::{Request, Response}; use crate::{ - calls::call_util, omega::omega_connection::OmegaConnection, rho::{client_connection::ClientConnection, iota_connection::IotaConnection}, util::{ @@ -25,6 +24,7 @@ use crate::{ #[tokio::main] async fn main() { + dotenv().ok(); tokio::spawn(async move { OmegaConnection::new().connect().await; }); diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 6f0e2ef..479ba0f 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -110,44 +110,44 @@ impl ClientConnection { /// Handle incoming message from client pub async fn handle_message(self: Arc, message: Utf8Bytes) { - let cv = CommunicationValue::from_json(&message); - - // Handle identification - if cv.is_type(CommunicationType::identification) && !self.is_identified().await { - self.handle_identification(Arc::clone(&self), cv).await; - return; - } - - if !self.is_identified().await { - return; - } - - // Handle ping - if cv.is_type(CommunicationType::ping) { - self.handle_ping(cv).await; - return; - } - line(PrintType::ClientIn, &cv.to_json().to_string()); - // Handle client status changes - if cv.is_type(CommunicationType::client_changed) { - self.handle_client_changed(cv).await; - return; - } - - // Handle call invites - if cv.is_type(CommunicationType::call_invite) { - self.handle_call_invite(cv).await; - return; - } - - // Handle get call requests - if cv.is_type(CommunicationType::call_token) { - self.handle_get_call(cv).await; - return; - } - - // Forward other messages to Iota tokio::spawn(async move { + let cv = CommunicationValue::from_json(&message); + + // Handle identification + if cv.is_type(CommunicationType::identification) && !self.is_identified().await { + self.handle_identification(Arc::clone(&self), cv).await; + return; + } + + if !self.is_identified().await { + return; + } + + // Handle ping + if cv.is_type(CommunicationType::ping) { + self.handle_ping(cv).await; + return; + } + line(PrintType::ClientIn, &cv.to_json().to_string()); + // Handle client status changes + if cv.is_type(CommunicationType::client_changed) { + self.handle_client_changed(cv).await; + return; + } + + // Handle call invites + if cv.is_type(CommunicationType::call_invite) { + self.handle_call_invite(cv).await; + return; + } + + // Handle get call requests + if cv.is_type(CommunicationType::call_token) { + self.handle_get_call(cv).await; + return; + } + + // Forward other messages to Iota self.forward_to_iota(cv).await; }); } diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index df9df92..9478943 100644 --- a/src/rho/rho_manager.rs +++ b/src/rho/rho_manager.rs @@ -1,7 +1,6 @@ use super::rho_connection::RhoConnection; use crate::util::print::PrintType; use crate::util::print::line; -use crate::util::print::line_err; use std::{ collections::HashMap, sync::{Arc, LazyLock},