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

New Calling structure using https://livekit.io/ (: Incomplete, Untested

Browse source 
:)
This commit is contained in:
Alex Emmet 2025-11-24 22:42:02 +00:00
commit 48ead50947
13 changed files with 1167 additions and 568 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -8,7 +8,7 @@ target
# MSVC Windows builds of rustc generate these, which store debugging information
*.pdb
*.env
# Generated by cargo mutants
# Contains mutation testing data
**/mutants.out*/W

988
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

4
Cargo.toml
View file

@ -45,3 +45,7 @@ walkdir = "2.5.0"
warp = "*"
x448 = { version = "*" }
x509 = "*"
log = "0.4"
livekit = "0.7.25"
livekit-api = "0.4.10"
dotenv = "0.15.0"

283
src/calls/call_connection.rs
View file

@ -1,283 +0,0 @@
use async_tungstenite::{WebSocketReceiver, WebSocketSender, tungstenite::Message};
use std::{any::Any, sync::Arc};
use tokio::sync::{
RwLock,
mpsc::{UnboundedSender, unbounded_channel},
};
use tokio_util::compat::Compat;
use tungstenite::Utf8Bytes;
use uuid::Uuid;
use crate::{
calls::call_manager,
data::communication::{CommunicationType, CommunicationValue, DataTypes},
util::print::{PrintType, line, line_err},
};
use json::JsonValue;
pub struct CallConnection {
pub sender: Arc<RwLock<WebSocketSender<Compat<tokio::net::TcpStream>>>>,
pub receiver: Arc<RwLock<WebSocketReceiver<Compat<tokio::net::TcpStream>>>>,
pub tx: UnboundedSender<Utf8Bytes>,
pub user_id: Arc<RwLock<Option<Uuid>>>,
pub call_id: Arc<RwLock<Option<Uuid>>>,
}
impl CallConnection {
pub async fn new(
sender: WebSocketSender<Compat<tokio::net::TcpStream>>,
receiver: WebSocketReceiver<Compat<tokio::net::TcpStream>>,
) -> Arc<Self> {
let (tx, mut rx) = unbounded_channel::<Utf8Bytes>();
let conn = Arc::new(Self {
sender: Arc::new(RwLock::new(sender)),
receiver: Arc::new(RwLock::new(receiver)),
tx,
user_id: Arc::new(RwLock::new(None)),
call_id: Arc::new(RwLock::new(None)),
});
// Spawn sender task to forward tx → session
let sender_clone = Arc::clone(&conn.sender);
tokio::spawn(async move {
while let Some(msg) = rx.recv().await {
let mut sess = sender_clone.write().await;
let _ = sess.send(Message::Text(msg)).await;
}
});
conn
}
/// Handle a single incoming message
pub async fn handle_message(self: Arc<Self>, msg: Utf8Bytes) {
let mut cv = CommunicationValue::from_json(&msg);
if let Some(sender) = *self.user_id.read().await {
cv = cv.with_sender(sender);
}
if cv.is_type(CommunicationType::webrtc_sdp)
| cv.is_type(CommunicationType::webrtc_ice)
| cv.is_type(CommunicationType::watch_stream)
{
line(
PrintType::CallIn,
&format!(
"{:?} : {} -> {} ",
&cv.type_id(),
&cv.get_sender(),
&cv.get_data(DataTypes::receiver_id).unwrap()
),
);
} else if !cv.is_type(CommunicationType::ping) {
line(PrintType::CallIn, &cv.to_json().to_string());
}
match cv.comm_type {
CommunicationType::identification => self.handle_identification(cv).await,
CommunicationType::ping => self.handle_ping(cv).await,
CommunicationType::client_changed => self.handle_client_changed(cv).await,
CommunicationType::start_stream | CommunicationType::end_stream => {
self.handle_stream_toggle(cv).await
}
CommunicationType::webrtc_sdp
| CommunicationType::webrtc_ice
| CommunicationType::watch_stream => self.handle_direct_relay(cv).await,
_ => {}
}
}
async fn handle_identification(&self, cv: CommunicationValue) {
let user_str = cv
.get_data(DataTypes::user_id)
.and_then(|v| v.as_str())
.unwrap_or("");
let call_str = cv
.get_data(DataTypes::call_id)
.and_then(|v| v.as_str())
.unwrap_or("");
let secret_sha = cv
.get_data(DataTypes::call_secret_sha)
.and_then(|v| v.as_str())
.unwrap_or("");
let (Ok(uid), Ok(cid)) = (Uuid::parse_str(user_str), Uuid::parse_str(call_str)) else {
return;
};
{
*self.user_id.write().await = Some(uid);
*self.call_id.write().await = Some(cid);
}
let group = call_manager::get_or_create_group(cid, secret_sha).await;
if let None = group {
self.send_message(&CommunicationValue::new(
CommunicationType::error_invalid_secret,
))
.await;
return;
}
let group = group.unwrap();
// Build broadcast
let broadcast = CommunicationValue::new(CommunicationType::client_connected)
.with_id(cv.get_id().clone())
.add_data_str(DataTypes::user_id, uid.to_string())
.add_data_str(DataTypes::call_state, "muted".to_string());
{
group
.lock()
.await
.broadcast(&broadcast.to_json().to_string())
.await;
}
{
group.lock().await.add_member(uid, self.tx.clone());
}
// Build response
let mut response = CommunicationValue::new(CommunicationType::identification_response)
.with_id(cv.get_id().clone());
let mut users = JsonValue::new_object();
{
for caller in group.lock().await.callers.values() {
let mut user_info = JsonValue::new_object();
let _ = user_info.insert("state", JsonValue::from("muted"));
let _ = user_info.insert("streaming", JsonValue::from(false));
let _ = users.insert(&caller.user_id.read().await.to_string(), user_info);
}
}
response = response.add_data(DataTypes::about, users);
self.send_message(&response).await;
}
async fn handle_ping(&self, cv: CommunicationValue) {
let resp = CommunicationValue::new(CommunicationType::pong).with_id(cv.get_id().clone());
self.send_message(&resp).await;
}
async fn handle_client_changed(&self, cv: CommunicationValue) {
let uid = match *self.user_id.read().await {
Some(id) => id,
None => return,
};
let cid = match *self.call_id.read().await {
Some(id) => id,
None => return,
};
if let Some(JsonValue::String(_state_str)) = cv.get_data(DataTypes::call_state) {
let Some(group_lock) = call_manager::get_group(cid).await else {
return;
};
let mut group = group_lock.lock().await;
if let Some(_caller) = group.caller_state_mut(&uid) {
// caller_state_change(caller, &state_str);
}
let mut bc = cv.clone();
bc = bc.add_data(DataTypes::sender_id, JsonValue::String(uid.to_string()));
group.broadcast(&bc.to_json().to_string()).await;
}
}
async fn handle_stream_toggle(&self, cv: CommunicationValue) {
let uid = match *self.user_id.read().await {
Some(id) => id,
None => return,
};
let cid = match *self.call_id.read().await {
Some(id) => id,
None => return,
};
if let Some(group_lock) = call_manager::get_group(cid).await {
let group = group_lock.lock().await;
let _streaming = cv.comm_type == CommunicationType::start_stream;
let mut bc = cv.clone();
bc = bc.add_data(DataTypes::sender_id, JsonValue::String(uid.to_string()));
group.broadcast(&bc.to_json().to_string()).await;
}
}
async fn handle_direct_relay(&self, cv: CommunicationValue) {
let uid = match *self.user_id.read().await {
Some(id) => id,
None => return,
};
let cid = match *self.call_id.read().await {
Some(id) => id,
None => return,
};
if let Some(JsonValue::String(receiver_str)) = cv.get_data(DataTypes::receiver_id) {
if let Ok(receiver_id) = Uuid::parse_str(&receiver_str) {
if let Some(group) = call_manager::get_group(cid).await {
let mut bc = cv.clone();
bc = bc.add_data(DataTypes::sender_id, JsonValue::String(uid.to_string()));
group
.lock()
.await
.send_to(&receiver_id, &bc.to_json().to_string())
.await;
line(
PrintType::CallOut,
&format!("Forwarded WebRTC message to receiver: {}", receiver_str),
);
} else {
line_err(PrintType::CallOut, "Failed to find the group for the call.");
}
} else {
line_err(PrintType::CallOut, "Invalid receiver_id in WebRTC message.");
}
} else {
line_err(PrintType::CallOut, "Missing receiver_id in WebRTC message.");
}
}
pub async fn send_message(&self, cv: &CommunicationValue) {
if !cv.is_type(CommunicationType::pong) {
line(PrintType::CallOut, &cv.to_json().to_string());
}
let mut session = self.sender.write().await;
if let Err(e) = session
.send(Message::Text(Utf8Bytes::from(cv.to_json().to_string())))
.await
{
line_err(
PrintType::CallOut,
&format!("Failed to send message to client: {}", e),
);
}
}
pub async fn close(&self) {
let (uid, cid) = { (*self.user_id.read().await, *self.call_id.read().await) };
if let (Some(uid), Some(cid)) = (uid, cid) {
if let Some(group) = call_manager::get_group(cid).await {
group
.lock()
.await
.broadcast(
&CommunicationValue::new(CommunicationType::client_disconnected)
.add_data_str(DataTypes::user_id, uid.to_string())
.to_json()
.to_string(),
)
.await;
group.lock().await.disconnect_member(uid).await;
}
call_manager::remove_inactive().await;
}
let mut session = self.sender.write().await;
let _ = session.close(None).await;
}
pub async fn handle_close(&self) {
self.close().await;
}
}

65
src/calls/call_group.rs
View file

@ -1,60 +1,27 @@
use crate::calls::caller::Caller;
use std::{collections::HashMap, sync::Arc};
use tokio::sync::mpsc::UnboundedSender;
use tungstenite::Utf8Bytes;
use std::sync::Arc;
use tokio::sync::RwLock;
use uuid::Uuid;
use crate::calls::caller::Caller;
pub struct CallGroup {
pub callers: HashMap<Uuid, Arc<Caller>>,
pub secret_hash: String,
pub call_id: Uuid,
pub members: RwLock<Vec<Arc<Caller>>>,
}
impl CallGroup {
pub fn new(secret_hash: String) -> Self {
Self {
callers: HashMap::new(),
secret_hash: secret_hash,
pub fn new(call_id: Uuid, user: Arc<Caller>) -> Self {
CallGroup {
call_id,
members: RwLock::new(vec![user]),
}
}
pub fn add_member(&mut self, user_id: Uuid, tx: UnboundedSender<Utf8Bytes>) {
self.callers
.insert(user_id, Arc::new(Caller::new(user_id, tx)));
}
pub fn remove_member(&mut self, user_id: Uuid) {
self.callers.remove(&user_id);
}
pub fn get_member(&self, user_id: &Uuid) -> Option<&Arc<Caller>> {
self.callers.get(user_id)
}
pub async fn disconnect_member(&mut self, user_id: Uuid) {
if let Some(caller) = self.callers.get(&user_id) {
caller.disconnect().await;
}
}
pub async fn is_empty(&self) -> bool {
for caller in self.callers.values() {
if caller.is_connected().await {
return false;
}
}
true
}
pub async fn send_to(&self, user_id: &Uuid, message: &str) {
if let Some(caller) = self.callers.get(user_id) {
caller.send(Utf8Bytes::from(message.to_string())).await;
}
}
pub async fn broadcast(&self, message: &str) {
for caller in self.callers.values() {
caller.send(Utf8Bytes::from(message.to_string())).await;
}
}
pub fn caller_state_mut(&mut self, user_id: &Uuid) -> Option<&Arc<Caller>> {
self.callers.get(user_id)
pub async fn add_member(self: Arc<Self>, member: Uuid, inviter: Uuid) {
self.members
.write()
.await
.push(Arc::new(Caller::new(member, inviter, self.call_id)));
}
}

126
src/calls/call_manager.rs
View file

@ -1,87 +1,81 @@
use crate::calls::call_group::CallGroup;
use futures::lock::Mutex; // Used only to protect CallGroup contents
use once_cell::sync::Lazy;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock; // Used to protect the global HashMap
use once_cell::sync::Lazy;
use tokio::sync::RwLock;
use uuid::Uuid;
// FIX 1: Simplify the global state. Only use RwLock to protect the HashMap.
// The inner Arc<Mutex<CallGroup>> protects the contents of each group.
pub static CALL_GROUPS: Lazy<RwLock<HashMap<Uuid, Arc<Mutex<CallGroup>>>>> =
Lazy::new(|| RwLock::new(HashMap::new()));
use crate::calls::{call_group::CallGroup, caller::Caller};
/// Retrieves a CallGroup by ID, if it exists.
pub async fn get_group(call_id: Uuid) -> Option<Arc<Mutex<CallGroup>>> {
// Acquire read lock (very fast, non-blocking for other readers)
CALL_GROUPS
.read()
.await
.get(&call_id) // Use get(), as we only need to clone the Arc
.cloned()
}
static CALL_GROUPS: Lazy<RwLock<Vec<Arc<CallGroup>>>> = Lazy::new(|| RwLock::new(Vec::new()));
/// Retrieves an existing group or creates a new one, performing a secret check.
pub async fn get_or_create_group(call_id: Uuid, secret: &str) -> Option<Arc<Mutex<CallGroup>>> {
// Phase 1: Check existence under a READ lock.
{
let groups = CALL_GROUPS.read().await;
if let Some(group_arc) = groups.get(&call_id) {
// Found it. Release the global READ lock and check the secret.
// This await on the group's lock only blocks access to that *specific* group.
let group_lock = group_arc.lock().await;
if group_lock.secret_hash.eq(secret) {
return Some(group_arc.clone());
} else {
return None; // Secret mismatch
pub async fn get_call_invites(user_id: Uuid) -> Vec<Arc<Caller>> {
let mut callers = Vec::new();
for cg in CALL_GROUPS.read().await.iter() {
for member in cg.members.read().await.iter() {
if member.user_id == user_id {
callers.push(member.clone());
}
}
} // READ lock automatically released here.
}
callers
}
// Phase 2: Not found, acquire a WRITE lock to create (only held for insertion).
let mut groups = CALL_GROUPS.write().await;
pub async fn get_call_groups(user_id: Uuid) -> Vec<Arc<CallGroup>> {
let mut call_groups = Vec::new();
for cg in CALL_GROUPS.read().await.iter() {
for member in cg.members.read().await.iter() {
if member.user_id == user_id {
call_groups.push(cg.clone());
}
}
}
call_groups
}
// FIX 2: Double-check (race condition prevention) - A group might have been created
// between the read lock release and the write lock acquisition.
if let Some(group_arc) = groups.get(&call_id) {
// Already created by another task, check secret again.
let group_lock = group_arc.lock().await;
if group_lock.secret_hash.eq(secret) {
return Some(group_arc.clone());
} else {
pub async fn get_call_token(user_id: Uuid, call_id: Uuid) -> Option<String> {
let call_groups = CALL_GROUPS.read().await;
for cg in call_groups.iter() {
if cg.call_id == call_id {
for member in cg.members.read().await.iter() {
if member.user_id == user_id {
return Some(member.create_token());
}
}
return None;
}
}
// Still not found, proceed with creation.
let new_group = Arc::new(Mutex::new(CallGroup::new(secret.to_string())));
groups.insert(call_id, new_group.clone());
// WRITE lock automatically released here.
Some(new_group)
let caller = Arc::new(Caller::new(user_id, user_id, call_id));
let call_group = CallGroup::new(call_id, caller.clone());
{
CALL_GROUPS.write().await.push(Arc::new(call_group));
}
Some(caller.create_token())
}
/// Removes inactive groups.
pub async fn remove_inactive() {
// Collect keys to remove under a read lock first.
let groups_to_check: Vec<Uuid> = CALL_GROUPS.read().await.keys().cloned().collect();
let mut rem = Vec::new();
for call_id in groups_to_check {
// Retrieve the group Arc outside the global lock
if let Some(group_arc) = get_group(call_id).await {
// FIX 3: Check for emptiness outside the global lock
if group_arc.lock().await.is_empty().await {
rem.push(call_id);
pub async fn add_invite(call_id: Uuid, inviter_id: Uuid, invitee_id: Uuid) -> bool {
let call_groups = CALL_GROUPS.read().await;
for cg in call_groups.iter() {
if cg.call_id == call_id {
for member in cg.members.read().await.iter() {
if member.user_id == inviter_id {
cg.clone().add_member(inviter_id, invitee_id).await;
return true;
}
}
return false;
}
}
false
}
// Acquire write lock only to perform the removals.
if !rem.is_empty() {
let mut groups = CALL_GROUPS.write().await;
for call_id in rem {
groups.remove(&call_id);
pub async fn get_call_group_by_user(user_id: Uuid) -> Option<Arc<CallGroup>> {
let call_groups = CALL_GROUPS.read().await;
for cg in call_groups.iter() {
for member in cg.members.read().await.iter() {
if member.user_id == user_id {
return Some(cg.clone());
}
}
}
None
}

21
src/calls/call_util.rs Normal file
View file

@ -0,0 +1,21 @@
use livekit_api::access_token;
use std::env;
use uuid::Uuid;
pub fn create_token(
user_id: Uuid,
call_id: Uuid,
) -> Result<String, access_token::AccessTokenError> {
let api_key = env::var("LIVEKIT_API_KEY").expect("LIVEKIT_API_KEY is not set");
let api_secret = env::var("LIVEKIT_API_SECRET").expect("LIVEKIT_API_SECRET is not set");
let token = access_token::AccessToken::with_api_key(&api_key, &api_secret)
.with_identity(&user_id.to_string())
.with_grants(access_token::VideoGrants {
room_join: true,
room: call_id.to_string(),
..Default::default()
})
.to_jwt();
return token;
}

53
src/calls/caller.rs
View file

@ -1,49 +1,26 @@
use std::sync::Arc;
use futures::lock::Mutex;
use tokio::sync::{RwLock, mpsc::UnboundedSender};
use tungstenite::Utf8Bytes;
use uuid::Uuid;
use crate::calls::call_util;
pub struct Caller {
pub user_id: RwLock<Uuid>,
pub tx: Mutex<Option<UnboundedSender<Utf8Bytes>>>,
pub user_state: RwLock<CallUserState>,
pub streaming: RwLock<bool>,
}
#[derive(Clone)]
pub enum CallUserState {
Active,
Muted,
Deafed,
Disconnected,
pub user_id: Uuid,
pub call_id: Uuid,
pub inviters: Vec<Uuid>,
}
impl Caller {
pub fn new(user_id: Uuid, tx: UnboundedSender<Utf8Bytes>) -> Self {
Self {
user_id: RwLock::new(user_id),
tx: Mutex::new(Some(tx)),
user_state: RwLock::new(CallUserState::Active),
streaming: RwLock::new(false),
pub fn new(user_id: Uuid, call_id: Uuid, inviter_id: Uuid) -> Self {
Caller {
user_id,
call_id,
inviters: vec![inviter_id],
}
}
pub async fn send(&self, msg: impl Into<Utf8Bytes>) {
if let Some(tx) = &*self.tx.lock().await {
let _ = tx.send(msg.into());
}
}
pub async fn disconnect(self: &Arc<Self>) {
*self.user_state.write().await = CallUserState::Disconnected;
*self.tx.lock().await = None;
}
pub async fn is_connected(&self) -> bool {
if let CallUserState::Disconnected = *self.user_state.read().await {
false
pub fn create_token(&self) -> String {
if let Ok(token) = call_util::create_token(self.user_id, self.call_id) {
token
} else {
true
String::new()
}
}
}

2
src/calls/mod.rs
View file

@ -1,4 +1,4 @@
pub mod call_connection;
pub mod call_group;
pub mod call_manager;
pub mod call_util;
pub mod caller;

20
src/data/communication.rs
View file

@ -20,7 +20,6 @@ pub enum DataTypes {
user_state,
user_states,
user_pings,
call_state,
screen_share,
private_key_hash,
accepted,
@ -36,10 +35,7 @@ pub enum DataTypes {
shared_secret_sign,
shared_secret,
call_id,
call_name,
call_secret_sha,
call_secret,
shared_call_secret,
call_token,
start_date,
end_date,
receiver_id,
@ -96,7 +92,6 @@ impl DataTypes {
"userstate" => DataTypes::user_state,
"userstates" => DataTypes::user_states,
"userpings" => DataTypes::user_pings,
"callstate" => DataTypes::call_state,
"screenshare" => DataTypes::screen_share,
"privatekeyhash" => DataTypes::private_key_hash,
"accepted" => DataTypes::accepted,
@ -112,10 +107,7 @@ impl DataTypes {
"sharedsecretsign" => DataTypes::shared_secret_sign,
"sharedsecret" => DataTypes::shared_secret,
"callid" => DataTypes::call_id,
"callname" => DataTypes::call_name,
"callsecretsha" => DataTypes::call_secret_sha,
"callsecret" => DataTypes::call_secret,
"sharedcallsecret" => DataTypes::shared_call_secret,
"calltoken" => DataTypes::call_token,
"startdate" => DataTypes::start_date,
"enddate" => DataTypes::end_date,
"receiverid" => DataTypes::receiver_id,
@ -209,10 +201,8 @@ pub enum CommunicationType {
start_stream,
end_stream,
watch_stream,
get_call,
new_call,
call_get,
call_invite,
end_call,
function,
update,
}
@ -263,10 +253,8 @@ impl CommunicationType {
"startstream" => CommunicationType::start_stream,
"endstream" => CommunicationType::end_stream,
"watchstream" => CommunicationType::watch_stream,
"getcall" => CommunicationType::get_call,
"newcall" => CommunicationType::new_call,
"callget" => CommunicationType::call_get,
"callinvite" => CommunicationType::call_invite,
"endcall" => CommunicationType::end_call,
"function" => CommunicationType::function,
"update" => CommunicationType::update,
_ => CommunicationType::error,

44
src/main.rs
View file

@ -7,23 +7,24 @@ mod util;
use async_tungstenite::accept_hdr_async;
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_connection::CallConnection,
calls::call_util,
omega::omega_connection::OmegaConnection,
rho::{client_connection::ClientConnection, iota_connection::IotaConnection},
util::config_util::CONFIG,
util::print::{PrintType, line, line_err, print_start_message},
util::{
config_util::CONFIG,
print::{PrintType, line, line_err},
},
};
#[tokio::main]
async fn main() {
print_start_message();
tokio::spawn(async move {
OmegaConnection::new().connect().await;
});
@ -121,39 +122,6 @@ async fn main() {
}
}
}
} else if path == "/ws/call/" {
line(PrintType::CallIn, "New Call connection");
let call_conn: Arc<CallConnection> =
Arc::from(CallConnection::new(sender, receiver).await);
loop {
let msg_result = {
let mut session_lock = call_conn.receiver.write().await;
session_lock.next().await
};
match msg_result {
Some(Ok(msg)) => {
if msg.is_text() {
let text = msg.into_text().unwrap();
call_conn.clone().handle_message(text).await;
} else if msg.is_close() {
line(PrintType::CallIn, "Call disconnected");
call_conn.handle_close().await;
return;
}
}
Some(Err(e)) => {
line_err(PrintType::CallIn, &format!("WebSocket error: {}", e));
call_conn.handle_close().await;
return;
}
None => {
line(PrintType::CallIn, "Call stream ended");
call_conn.handle_close().await;
return;
}
}
}
}
});
}

54
src/rho/client_connection.rs
View file

@ -141,7 +141,7 @@ impl ClientConnection {
}
// Handle get call requests
if cv.is_type(CommunicationType::get_call) {
if cv.is_type(CommunicationType::call_get) {
self.handle_get_call(cv).await;
return;
}
@ -303,13 +303,14 @@ impl ClientConnection {
}
};
// Get call group - placeholder implementation
// let call_group = CallManager::get_call_group(call_id, &call_secret_sha.unwrap(), true).await;
// if call_group.is_none() {
// self.send_error_response(&cv.get_id(), CommunicationType::error)
// .await;
// return;
// }
let invited =
call_manager::add_invite(call_id, self.user_id.read().await.unwrap(), receiver_id)
.await;
if invited {
self.send_error_response(&cv.get_id(), CommunicationType::error)
.await;
return;
}
// Find target RhoConnection
let target_rho = match rho_manager::get_rho_con_for_user(receiver_id).await {
@ -328,20 +329,16 @@ impl ClientConnection {
};
// Create and send call distribution message
let distribute = CommunicationValue::new(CommunicationType::new_call)
let forward = CommunicationValue::new(CommunicationType::call_invite)
.with_receiver(receiver_id)
.with_sender(sender_id)
.add_data_str(DataTypes::call_id, call_id.to_string())
.add_data_str(DataTypes::receiver_id, receiver_id.to_string())
.add_data_str(DataTypes::sender_id, sender_id.to_string());
target_rho.message_to_client(distribute).await;
target_rho.message_to_client(forward).await;
// Handle call group invitation logic here
// This would require implementing CallGroup::Caller and related functionality
// Send success response
let response = CommunicationValue::new(CommunicationType::call_invite).with_id(cv.get_id());
let response = CommunicationValue::new(CommunicationType::success).with_id(cv.get_id());
self.send_message(&response).await;
}
@ -368,26 +365,17 @@ impl ClientConnection {
}
};
// Get call group
let mut response = CommunicationValue::new(CommunicationType::get_call)
if let Some(token) = call_manager::get_call_token(user_id, call_id).await {
let response = CommunicationValue::new(CommunicationType::call_get)
.with_id(cv.get_id())
.with_receiver(user_id);
if let Some(_call_group) = call_manager::get_group(call_id).await {
/*response = response
.add_data_str(DataTypes::call_state, call_group.call_state.to_string())
.add_data_str(DataTypes::start_date, call_group.started_at.to_string());
if call_group.lock(). != 0 {
response =
response.add_data_str(DataTypes::end_date, call_group.ended_at.to_string());
}
*/
} else {
response = response.add_data_str(DataTypes::call_state, "DESTROYED".to_string());
}
.with_receiver(user_id)
.add_data_str(DataTypes::call_token, token);
self.send_message(&response).await;
} else {
self.send_error_response(&cv.get_id(), CommunicationType::error)
.await;
return;
}
}
/// Forward message to Iota

55
src/rho/iota_connection.rs
View file

@ -1,3 +1,5 @@
use crate::calls::call_group::CallGroup;
use crate::calls::call_manager;
use crate::util::print::PrintType;
use crate::util::print::line;
use crate::util::print::line_err;
@ -5,12 +7,14 @@ use async_tungstenite::WebSocketReceiver;
use async_tungstenite::WebSocketSender;
use async_tungstenite::tungstenite::Message;
use json::JsonValue;
use json::parse;
use std::{
collections::HashMap,
sync::{Arc, Weak},
};
use tokio::sync::RwLock;
use tokio_util::compat::Compat;
use tower::retry::backoff::InvalidBackoff;
use tungstenite::Utf8Bytes;
use uuid::Uuid;
@ -278,22 +282,44 @@ impl IotaConnection {
/// Handle GET_CHATS message
async fn handle_get_chats(&self, cv: CommunicationValue) {
let receiver_id = cv.get_receiver();
let interested_ids: Vec<Uuid> = Vec::new();
let mut interested_ids: Vec<Uuid> = Vec::new();
// Process contacts and add call information
if let Some(contacts_data) = cv.get_data(DataTypes::user_ids) {
/*let contacts: Vec<(Uuid, String)> = parse_contacts(contacts_data);
for contact in &contacts {
interested_ids.push(contact.user_id);
let calls: Vec<Arc<CallGroup>> = call_manager::get_call_groups(receiver_id).await;
let mut invites: HashMap<Uuid, Vec<Uuid>> = HashMap::new();
for call in calls {
for inviter in call.members.read().await.iter() {
let inviter_id = inviter.user_id;
if let Some(call_ids) = invites.get_mut(&inviter_id) {
call_ids.push(call.call_id);
} else {
invites.insert(inviter_id, vec![call.call_id]);
}
}
}
// Get call invites for receiver
let invites = CallManager::get_call_invites(receiver_id).await;
// Enrich contacts with call information
let enriched_contacts = enrich_with_calls(contacts, invites);
cv = cv.add_data(DataTypes::user_ids, enriched_contacts.into());*/
// Process contacts and add call information
// Enriched Contacts data:
// [{"user_id": "uuid", "calls": ["call_id"]}, {"user_id": "uuid"}]
let mut enriched_contacts = JsonValue::new_array();
// Contacts data:
// [{"user_id": "uuid"}, {"user_id": "uuid"}]
if let Some(contacts_data) = cv.get_data(DataTypes::user_ids) {
if let JsonValue::Array(user_ids) = contacts_data {
for user_id in user_ids {
if let JsonValue::String(user_id_str) = user_id {
if let Ok(user_id) = Uuid::parse_str(user_id_str) {
interested_ids.push(user_id);
let mut enriched_contact = JsonValue::new_object();
let _ = enriched_contact.insert("user_id", user_id.to_string());
let _ = enriched_contact.insert("calls", JsonValue::new_array());
let _ = enriched_contacts.push(enriched_contact);
}
}
}
} else {
enriched_contacts = contacts_data.clone();
}
}
// Notify OmegaConnection about user states
@ -305,7 +331,8 @@ impl IotaConnection {
}
// Forward to client
self.forward_to_client(cv).await;
self.forward_to_client(cv.add_data(DataTypes::user_ids, enriched_contacts))
.await;
}
/// Forward message to client