From 309342b730a74728b26903aff0239ad0c3416577 Mon Sep 17 00:00:00 2001 From: Alex-Emmet Date: Tue, 20 Jan 2026 15:19:47 +0100 Subject: [PATCH] Call Kicking and Timeouting, basics of anonymous calls --- src/calls/call_group.rs | 22 ++++++++ src/calls/call_manager.rs | 21 ++++---- src/calls/caller.rs | 18 +++++++ src/rho/client_connection.rs | 100 ++++++++++++++++++++++++++++++++--- src/rho/iota_connection.rs | 20 +++++-- 5 files changed, 159 insertions(+), 22 deletions(-) diff --git a/src/calls/call_group.rs b/src/calls/call_group.rs index 3880046..0f10fbe 100644 --- a/src/calls/call_group.rs +++ b/src/calls/call_group.rs @@ -9,6 +9,7 @@ pub struct CallGroup { pub call_id: Uuid, pub members: RwLock>>, pub show: RwLock, + pub anonymous_joining: RwLock, } impl CallGroup { @@ -17,6 +18,27 @@ impl CallGroup { call_id, members: RwLock::new(vec![user]), show: RwLock::new(true), + anonymous_joining: RwLock::new(false), } } + + pub async fn get_caller(&self, user_id: i64) -> Option> { + self.members + .read() + .await + .iter() + .find(|caller| caller.user_id == user_id) + .cloned() + } + + pub async fn set_anonymous_joining(&self, enable: bool) { + *self.anonymous_joining.write().await = enable; + } + + pub async fn remove_caller(&self, user_id: i64) { + self.members + .write() + .await + .retain(|caller| caller.user_id != user_id); + } } diff --git a/src/calls/call_manager.rs b/src/calls/call_manager.rs index 5902cb0..b407d50 100644 --- a/src/calls/call_manager.rs +++ b/src/calls/call_manager.rs @@ -1,6 +1,6 @@ use livekit_api::services::room::RoomClient; use once_cell::sync::Lazy; -use std::{env, str::FromStr, sync::Arc, time::Duration}; +use std::{str::FromStr, sync::Arc, time::Duration}; use tokio::sync::RwLock; use uuid::Uuid; @@ -28,6 +28,11 @@ pub async fn get_call_groups(user_id: i64) -> Vec> { call_groups } +pub async fn get_call(call_id: Uuid) -> Option> { + let call_groups = CALL_GROUPS.read().await; + call_groups.iter().find(|g| g.call_id == call_id).cloned() +} + pub async fn get_call_token(user_id: i64, call_id: Uuid) -> Option { let existing_group = { let call_groups = CALL_GROUPS.read().await; @@ -36,21 +41,16 @@ pub async fn get_call_token(user_id: i64, call_id: Uuid) -> Option { // if the group exists if let Some(cg) = existing_group { - let members = cg.members.write().await; + let members = cg.members.read().await; // if the user is already a member if let Some(member) = members.iter().find(|m| m.user_id == user_id) { + if member.is_timeout().await { + return None; + } return Some(member.create_token()); } return None; - /* - let new_caller = Arc::new(Caller::new(user_id, call_id, user_id)); - let token = new_caller.create_token(); - - members.push(new_caller); - - return Some(token); - */ } let mut call_groups = CALL_GROUPS.write().await; @@ -85,6 +85,7 @@ pub async fn add_invite(call_id: Uuid, inviter_id: i64, invitee_id: i64) -> bool } false } + pub fn garbage_collect_calls() { tokio::spawn(async move { loop { diff --git a/src/calls/caller.rs b/src/calls/caller.rs index 4816f2d..f812e3f 100644 --- a/src/calls/caller.rs +++ b/src/calls/caller.rs @@ -1,3 +1,9 @@ +use std::{ + sync::Arc, + time::{SystemTime, UNIX_EPOCH}, +}; + +use tokio::sync::RwLock; use uuid::Uuid; use crate::calls::call_util; @@ -6,6 +12,7 @@ pub struct Caller { pub user_id: i64, pub call_id: Uuid, pub has_admin: bool, + pub timeout: RwLock, } impl Caller { @@ -14,6 +21,7 @@ impl Caller { user_id, call_id, has_admin, + timeout: RwLock::new(0), } } pub fn set_admin(&mut self, has_admin: bool) { @@ -22,6 +30,16 @@ impl Caller { pub fn has_admin(&self) -> bool { self.has_admin } + pub async fn is_timeout(&self) -> bool { + *self.timeout.read().await + > SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as i64 + } + pub async fn set_timeout(&self, timeout: i64) { + *self.timeout.write().await = timeout; + } pub fn create_token(&self) -> String { if let Ok(token) = call_util::create_token(self.user_id, self.call_id, self.has_admin()) { token diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index a72247c..470bc1a 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -4,6 +4,7 @@ use json::JsonValue; use json::number::Number; use rand::Rng; use rand::distributions::Alphanumeric; +use std::str::FromStr; use std::sync::Arc; use std::time::Duration; use tokio::sync::RwLock; @@ -296,6 +297,21 @@ impl ClientConnection { return; } + if cv.is_type(CommunicationType::call_disconnect_user) { + self.handle_call_disconnect_user(cv).await; + return; + } + + if cv.is_type(CommunicationType::call_timeout_user) { + self.handle_call_timeout_user(cv).await; + return; + } + + if cv.is_type(CommunicationType::call_set_anonymous_joining) { + self.handle_call_set_anonymous_joining(cv).await; + return; + } + if cv.is_type(CommunicationType::change_user_data) || cv.is_type(CommunicationType::get_user_data) || cv.is_type(CommunicationType::get_iota_data) @@ -462,17 +478,87 @@ impl ClientConnection { } } async fn handle_call_timeout_user(self: Arc, cv: CommunicationValue) { - let user_id = cv.get_data(DataTypes::call_id).unwrap(); - let call_id = cv.get_data(DataTypes::user_id).unwrap(); // JA man braucht CALL_ID + let call_id = Uuid::from_str( + cv.get_data(DataTypes::call_id) + .unwrap_or(&JsonValue::Null) + .as_str() + .unwrap_or(""), + ) + .unwrap(); + let user_id = cv + .get_data(DataTypes::user_id) + .unwrap_or(&JsonValue::Null) + .as_i64() + .unwrap_or(0); + let untill = cv + .get_data(DataTypes::untill) + .unwrap_or(&JsonValue::Null) + .as_i64() + .unwrap_or(0); + + let call = call_manager::get_call(call_id).await; + if let Some(call) = call { + if call + .get_caller(self.get_user_id().await) + .await + .unwrap() + .has_admin() + { + call.get_caller(user_id).await.unwrap().set_timeout(untill); + } + } } async fn handle_call_disconnect_user(self: Arc, cv: CommunicationValue) { - let user_id = cv.get_data(DataTypes::call_id).unwrap(); - let call_id = cv.get_data(DataTypes::user_id).unwrap(); // JA man braucht CALL_ID - let untill = cv.get_data(DataTypes::untill).unwrap(); + let call_id = Uuid::from_str( + cv.get_data(DataTypes::call_id) + .unwrap_or(&JsonValue::Null) + .as_str() + .unwrap_or(""), + ) + .unwrap(); + let user_id = cv + .get_data(DataTypes::user_id) + .unwrap_or(&JsonValue::Null) + .as_i64() + .unwrap_or(0); + + let call = call_manager::get_call(call_id).await; + if let Some(call) = call { + if call + .get_caller(self.get_user_id().await) + .await + .unwrap() + .has_admin() + { + call.remove_caller(user_id).await; + } + } } async fn handle_call_set_anonymous_joining(self: Arc, cv: CommunicationValue) { - let call_id = cv.get_data(DataTypes::user_id).unwrap(); - let enable = cv.get_data(DataTypes::enable).unwrap(); + let call_id = Uuid::from_str( + cv.get_data(DataTypes::call_id) + .unwrap_or(&JsonValue::Null) + .as_str() + .unwrap_or(""), + ) + .unwrap(); + let enable = cv + .get_data(DataTypes::enable) + .unwrap_or(&JsonValue::Null) + .as_bool() + .unwrap_or(false); + + let call = call_manager::get_call(call_id).await; + if let Some(call) = call { + if call + .get_caller(self.get_user_id().await) + .await + .unwrap() + .has_admin() + { + call.set_anonymous_joining(enable).await; + } + } } /// Forward message to Iota diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index d42a426..06f0704 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -513,14 +513,24 @@ impl IotaConnection { let empty = &calls.is_empty(); for call in calls { for inviter in call.members.read().await.iter() { + let call_self = call.get_caller(receiver_id).await.unwrap(); let inviter_id = inviter.user_id; if let Some(call_ids) = invites.get_mut(&inviter_id) { - call_ids.push(JsonValue::String(call.call_id.to_string())); + let mut call_obj = JsonValue::new_object(); + let _ = call_obj.insert("call_id", JsonValue::String(call.call_id.to_string())); + let timeout = *call_self.timeout.read().await; + if timeout > 0 { + let _ = call_obj.insert("timeout", JsonValue::from(timeout)); + } + call_ids.push(call_obj); } else { - invites.insert( - inviter_id, - vec![JsonValue::String(call.call_id.to_string())], - ); + let mut call_obj = JsonValue::new_object(); + let _ = call_obj.insert("call_id", JsonValue::String(call.call_id.to_string())); + let timeout = *call_self.timeout.read().await; + if timeout > 0 { + let _ = call_obj.insert("timeout", JsonValue::from(timeout)); + } + invites.insert(inviter_id, vec![call_obj]); } } }