tensamin/omikron
Watch
6
0
Fork
You've already forked omikron
 0
Code Issues Pull requests 3 Projects Releases Packages Wiki Activity Actions

Call Kicking and Timeouting, basics of anonymous calls

Browse source
This commit is contained in:
Alex-Emmet 2026-01-20 15:19:47 +01:00
commit 309342b730
5 changed files with 159 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

22
src/calls/call_group.rs
View file

@ -9,6 +9,7 @@ pub struct CallGroup {
pub call_id: Uuid,
pub members: RwLock<Vec<Arc<Caller>>>,
pub show: RwLock<bool>,
pub anonymous_joining: RwLock<bool>,
}
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<Arc<Caller>> {
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);
}
}

21
src/calls/call_manager.rs
View file

@ -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<Arc<CallGroup>> {
call_groups
}
pub async fn get_call(call_id: Uuid) -> Option<Arc<CallGroup>> {
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<String> {
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<String> {
// 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 {

18
src/calls/caller.rs
View file

@ -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<i64>,
}
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

100
src/rho/client_connection.rs
View file

@ -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<Self>, 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<Self>, 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<Self>, 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

20
src/rho/iota_connection.rs
View file

@ -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]);
}
}
}