From 3cb2f0ec5e56fde339087508739b53114063a121 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Thu, 13 Nov 2025 23:13:55 +0000 Subject: [PATCH] Calling Logging fixing --- src/auth/auth_connector.rs | 2 +- src/calls/call_connection.rs | 18 ++++++++++----- src/calls/call_group.rs | 29 +++++++++++++++-------- src/calls/call_manager.rs | 17 +++++++------- src/calls/caller.rs | 43 +++++++++++++++++++++++++++++++---- src/data/communication.rs | 14 +++++------- src/main.rs | 7 ++---- src/omega/omega_connection.rs | 6 ++--- src/rho/client_connection.rs | 42 +++++++++++++++++----------------- src/rho/iota_connection.rs | 16 +++++-------- src/rho/rho_connection.rs | 3 --- src/rho/rho_manager.rs | 33 --------------------------- 12 files changed, 118 insertions(+), 112 deletions(-) diff --git a/src/auth/auth_connector.rs b/src/auth/auth_connector.rs index 2c40098..2dd91f8 100644 --- a/src/auth/auth_connector.rs +++ b/src/auth/auth_connector.rs @@ -132,7 +132,7 @@ pub async fn get_public_key(user_id: Uuid) -> Option { .ok()?; let body = res.text().await.ok()?; - let mut cv = CommunicationValue::from_json(&body); + let cv = CommunicationValue::from_json(&body); if cv.comm_type != CommunicationType::message_send { return None; diff --git a/src/calls/call_connection.rs b/src/calls/call_connection.rs index 9ce475b..f360727 100644 --- a/src/calls/call_connection.rs +++ b/src/calls/call_connection.rs @@ -1,10 +1,7 @@ -use async_tungstenite::{ - WebSocketReceiver, WebSocketSender, WebSocketStream, tungstenite::Message, -}; -use futures::SinkExt; +use async_tungstenite::{WebSocketReceiver, WebSocketSender, tungstenite::Message}; use std::sync::Arc; use tokio::sync::{ - Mutex, RwLock, + RwLock, mpsc::{UnboundedSender, unbounded_channel}, }; use tokio_util::compat::Compat; @@ -101,6 +98,15 @@ impl CallConnection { } 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()) @@ -113,7 +119,6 @@ impl CallConnection { .broadcast(&broadcast.to_json().to_string()); } - // Add member to group { group.lock().await.add_member(uid, self.tx.clone()); } @@ -244,6 +249,7 @@ impl CallConnection { .to_json() .to_string(), ); + group.lock().await.get_member(uid).await; group.lock().await.remove_member(uid); } call_manager::remove_inactive().await; diff --git a/src/calls/call_group.rs b/src/calls/call_group.rs index 3d196de..cf1d32c 100644 --- a/src/calls/call_group.rs +++ b/src/calls/call_group.rs @@ -1,24 +1,30 @@ use crate::calls::caller::Caller; -use std::collections::HashMap; +use std::{collections::HashMap, sync::Arc}; use tokio::sync::mpsc::UnboundedSender; use tungstenite::Utf8Bytes; use uuid::Uuid; pub struct CallGroup { - pub call_id: Uuid, - pub callers: HashMap, + pub callers: HashMap>, + pub secret_hash: String, } impl CallGroup { - pub fn new(call_id: Uuid) -> Self { + pub fn new(secret_hash: String) -> Self { Self { - call_id, callers: HashMap::new(), + secret_hash: secret_hash, } } pub fn add_member(&mut self, user_id: Uuid, tx: UnboundedSender) { - self.callers.insert(user_id, Caller { user_id, tx }); + self.callers + .insert(user_id, Arc::new(Caller::new(user_id, tx))); + } + pub fn disconnect_member(&mut self, user_id: Uuid) { + if let Some(caller) = self.callers.get(&user_id) { + caller.disconnect(); + } } pub fn remove_member(&mut self, user_id: Uuid) { @@ -26,7 +32,12 @@ impl CallGroup { } pub fn is_empty(&self) -> bool { - self.callers.is_empty() + for caller in self.callers.values() { + if !caller.is_connected() { + return false; + } + } + true } pub fn send_to(&self, user_id: &Uuid, message: &str) { @@ -41,7 +52,7 @@ impl CallGroup { } } - pub fn caller_state_mut(&mut self, user_id: &Uuid) -> Option<&mut Caller> { - self.callers.get_mut(user_id) + pub fn caller_state_mut(&mut self, user_id: &Uuid) -> Option<&Arc> { + self.callers.get(user_id) } } diff --git a/src/calls/call_manager.rs b/src/calls/call_manager.rs index 3b6eb5d..ea2c27a 100644 --- a/src/calls/call_manager.rs +++ b/src/calls/call_manager.rs @@ -4,12 +4,8 @@ use once_cell::sync::Lazy; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; -use tokio::sync::mpsc::UnboundedSender; -use tungstenite::Utf8Bytes; use uuid::Uuid; -pub type Tx = UnboundedSender; - pub static CALL_GROUPS: Lazy>>>>> = Lazy::new(|| RwLock::new(Mutex::new(HashMap::new()))); @@ -22,14 +18,18 @@ pub async fn get_group(call_id: Uuid) -> Option>> { .get_mut(&call_id) .cloned() } -pub async fn get_or_create_group(call_id: Uuid, secret: &str) -> Arc> { +pub async fn get_or_create_group(call_id: Uuid, secret: &str) -> Option>> { let g = CALL_GROUPS.write().await; if let Some(group) = g.lock().await.get_mut(&call_id) { - group.clone() + if group.lock().await.secret_hash.eq(secret) { + Some(group.clone()) + } else { + None + } } else { - let cg = Arc::new(Mutex::new(CallGroup::new(call_id))); + let cg = Arc::new(Mutex::new(CallGroup::new(secret.to_string()))); g.lock().await.insert(call_id, cg.clone()); - cg + Some(cg) } } @@ -45,7 +45,6 @@ pub async fn remove_inactive() { .unwrap() .lock() .await - .callers .is_empty() { rem.push(cg.clone()); diff --git a/src/calls/caller.rs b/src/calls/caller.rs index 83a43d5..8fb0fda 100644 --- a/src/calls/caller.rs +++ b/src/calls/caller.rs @@ -1,14 +1,49 @@ +use std::sync::Arc; + +use futures::lock::Mutex; use tokio::sync::mpsc::UnboundedSender; use tungstenite::Utf8Bytes; use uuid::Uuid; -#[derive(Debug, Clone)] pub struct Caller { - pub user_id: Uuid, - pub tx: UnboundedSender, + pub user_id: Mutex, + pub tx: Mutex>>, + pub user_state: Mutex, + pub streaming: Mutex, +} +#[derive(Clone)] +pub enum CallUserState { + Active, + Muted, + Deafed, + Disconnected, } impl Caller { + pub fn new(user_id: Uuid, tx: UnboundedSender) -> Self { + Self { + user_id: Mutex::new(user_id), + tx: Mutex::new(Some(tx)), + user_state: Mutex::new(CallUserState::Active), + streaming: Mutex::new(false), + } + } + pub fn send(&self, msg: impl Into) { - let _ = self.tx.send(msg.into()); + if let Some(tx) = &self.tx.lock().await { + let _ = tx.send(msg.into()); + } + } + + pub fn disconnect(self: Arc) { + self.user_state = CallUserState::Disconnected; + self.tx = None; + } + + pub fn is_connected(&self) -> bool { + if let CallUserState::Disconnected = self.user_state { + false + } else { + true + } } } diff --git a/src/data/communication.rs b/src/data/communication.rs index 44ef8e1..c396125 100644 --- a/src/data/communication.rs +++ b/src/data/communication.rs @@ -6,6 +6,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use uuid::Uuid; #[derive(Eq, Hash, PartialEq, Clone, Debug)] +#[allow(non_camel_case_types, dead_code)] pub enum DataTypes { error_type, accepted_ids, @@ -153,10 +154,14 @@ impl DataTypes { } #[derive(PartialEq, Clone, Debug)] +#[allow(non_camel_case_types, dead_code)] pub enum CommunicationType { error, error_invalid_user_id, + error_not_found, error_no_iota, + error_invalid_challenge, + error_invalid_secret, error_invalid_private_key, success, message, @@ -267,6 +272,7 @@ pub struct CommunicationValue { pub data: HashMap, } +#[allow(dead_code)] impl CommunicationValue { pub fn new(comm_type: CommunicationType) -> Self { Self { @@ -366,14 +372,6 @@ impl CommunicationValue { data, } } - pub fn ack_message(message_id: Uuid, sender: Uuid) -> CommunicationValue { - let mut cv = CommunicationValue::new(CommunicationType::message).with_id(message_id); - - let s = sender; - cv = cv.add_data(DataTypes::send_time, JsonValue::String(s.to_string())); - - cv - } pub fn forward_to_other_iota(original: &mut CommunicationValue) -> CommunicationValue { let receiver = Uuid::from_str( &*original diff --git a/src/main.rs b/src/main.rs index 035a6be..c7abfc1 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,13 +7,10 @@ mod util; use async_tungstenite::accept_hdr_async; use futures::StreamExt; -use std::{sync::Arc, time::Duration}; +use std::sync::Arc; use tokio::net::TcpListener; use tokio_util::compat::TokioAsyncReadCompatExt; -use tungstenite::{ - Message, Utf8Bytes, - handshake::server::{Request, Response}, -}; +use tungstenite::handshake::server::{Request, Response}; use crate::{ calls::call_connection::CallConnection, diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index 0285eaf..b7037c5 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -3,7 +3,7 @@ use std::time::Duration; use crate::data::communication::{CommunicationType, CommunicationValue}; use crate::util::print::PrintType; -use crate::util::print::line_err; +use crate::util::print::{line, line_err}; use crate::{ data::{ communication::DataTypes, @@ -14,10 +14,9 @@ use crate::{ }; use async_tungstenite::tungstenite::protocol::Message; use dashmap::DashMap; -use futures::{SinkExt, StreamExt}; +use futures::StreamExt; use json::JsonValue; use once_cell::sync::Lazy; -use tokio::io::unix::AsyncFd; use tokio::sync::Mutex; use tokio::time::sleep; use tokio_util::compat::Compat; @@ -141,6 +140,7 @@ impl OmegaConnection { pub async fn send_message(&self, cv: &CommunicationValue) { let mut guard = self.ws_stream.lock().await; if let Some(ws) = guard.as_mut() { + line(PrintType::OmegaOut, &cv.to_json().to_string()); let _ = ws .send(Message::Text(Utf8Bytes::from(cv.to_json().to_string()))) .await; diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 2b1e07e..e853c0d 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -1,14 +1,13 @@ +use async_tungstenite::tungstenite::Message; use async_tungstenite::{WebSocketReceiver, WebSocketSender}; -use async_tungstenite::{WebSocketStream, tungstenite::Message}; -use futures::SinkExt; use std::sync::{Arc, Weak}; -use tokio::sync::Mutex; use tokio::sync::RwLock; use tokio_util::compat::Compat; use tungstenite::Utf8Bytes; use uuid::Uuid; use super::{rho_connection::RhoConnection, rho_manager}; +use crate::calls::call_manager; use crate::util::print::PrintType; use crate::util::print::line; use crate::util::print::line_err; @@ -154,7 +153,7 @@ impl ClientConnection { } /// Handle identification message - async fn handle_identification(&self, sarc: Arc, mut cv: CommunicationValue) { + async fn handle_identification(&self, sarc: Arc, cv: CommunicationValue) { // Extract user ID let user_id = match cv.get_data(DataTypes::user_id) { Some(id_str) => match Uuid::parse_str(&id_str.to_string()) { @@ -228,7 +227,7 @@ impl ClientConnection { } /// Handle ping message - async fn handle_ping(&self, mut cv: CommunicationValue) { + async fn handle_ping(&self, cv: CommunicationValue) { // Update our ping if provided if let Some(last_ping) = cv.get_data(DataTypes::last_ping) { if let Ok(ping_val) = last_ping.to_string().parse::() { @@ -253,9 +252,9 @@ impl ClientConnection { } /// Handle client status change - async fn handle_client_changed(&self, mut cv: CommunicationValue) { + async fn handle_client_changed(&self, cv: CommunicationValue) { if let Some(user_id) = self.get_user_id().await { - if let Some(status_str) = cv.get_data(DataTypes::user_state) { + if let Some(_status_str) = cv.get_data(DataTypes::user_state) { // Parse user status - this would need to be implemented properly let user_status = UserStatus::online; // placeholder if let Some(rho_conn) = self.get_rho_connection().await { @@ -359,7 +358,7 @@ impl ClientConnection { } /// Handle get call request - async fn handle_get_call(&self, mut cv: CommunicationValue) { + async fn handle_get_call(&self, cv: CommunicationValue) { let user_id = match self.get_user_id().await { Some(id) => id, None => return, @@ -396,24 +395,25 @@ impl ClientConnection { .with_id(cv.get_id()) .with_receiver(user_id); - // Placeholder call group logic - // if let Some(call_group) = CallManager::get_call_group(call_id, &call_secret_sha.unwrap(), true).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.ended_at != 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()); - // } + 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()); + } self.send_message(&response).await; } /// Forward message to Iota - async fn forward_to_iota(&self, mut cv: CommunicationValue) { + async fn forward_to_iota(&self, cv: CommunicationValue) { if let Some(user_id) = self.get_user_id().await { if let Some(rho_conn) = self.get_rho_connection().await { let updated_cv = cv.with_sender(user_id); diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index e7189a8..e404d95 100644 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -1,19 +1,15 @@ use crate::util::print::PrintType; use crate::util::print::line; use crate::util::print::line_err; -use ansi_term::Color; use async_tungstenite::WebSocketReceiver; use async_tungstenite::WebSocketSender; -use async_tungstenite::tungstenite::protocol::WebSocketConfig; -use async_tungstenite::{WebSocketStream, tungstenite::Message}; -use futures::FutureExt; -use futures::SinkExt; +use async_tungstenite::tungstenite::Message; use json::JsonValue; use std::{ collections::HashMap, sync::{Arc, Weak}, }; -use tokio::sync::{Mutex, RwLock}; +use tokio::sync::RwLock; use tokio_util::compat::Compat; use tungstenite::Utf8Bytes; use uuid::Uuid; @@ -280,12 +276,12 @@ impl IotaConnection { } /// Handle GET_CHATS message - async fn handle_get_chats(&self, mut cv: CommunicationValue) { + async fn handle_get_chats(&self, cv: CommunicationValue) { let receiver_id = cv.get_receiver(); - let mut interested_ids: Vec = Vec::new(); + let interested_ids: Vec = Vec::new(); // Process contacts and add call information - if let Some(contacts_data) = cv.get_data(DataTypes::user_ids) { + if let Some(_contacts_data) = cv.get_data(DataTypes::user_ids) { // Parse contacts JSON array and enrich with call data // This would need proper JSON parsing implementation // For now, placeholder logic: @@ -321,7 +317,7 @@ impl IotaConnection { async fn forward_to_client(&self, cv: CommunicationValue) { if let Some(rho_conn) = self.get_rho_connection().await { let updated_cv = cv.with_sender(self.get_iota_id().await); - let receiver_id = updated_cv.get_receiver(); + let _receiver_id = updated_cv.get_receiver(); rho_conn.message_to_client(updated_cv).await; } } diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 53ec5b4..e0342b4 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -4,9 +4,6 @@ use crate::data::{ user::UserStatus, }; use crate::omega::omega_connection::OmegaConnection; -use crate::util::print::PrintType; -use crate::util::print::line; -use crate::util::print::line_err; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index 7a26603..df9df92 100644 --- a/src/rho/rho_manager.rs +++ b/src/rho/rho_manager.rs @@ -68,36 +68,3 @@ pub async fn connection_count() -> usize { let connections = RHO_CONNECTIONS.read().await; connections.len() } - -#[cfg(test)] -mod tests { - use super::*; - use crate::rho::iota_connection::IotaConnection; - use std::sync::Arc; - use tokio::sync::Mutex; - - #[tokio::test] - async fn test_add_and_get_rho() { - // Clear any existing connections - { - let mut connections = RHO_CONNECTIONS.write().await; - connections.clear(); - } - - let iota_id = Uuid::new_v4(); - let user_ids = vec![Uuid::new_v4(), Uuid::new_v4()]; - - // Skip this test due to WebSocket complexity - would require proper mock setup - return; - - // This test would need proper WebSocket stream mocking: - // let mock_session = create_mock_websocket_stream(); - // let iota_conn = IotaConnection::new_with_ids(iota_id, user_ids.clone(), mock_session); - // let rho_conn = Arc::new(RhoConnection::new(iota_conn, user_ids.clone())); - - // Test assertions would go here: - // add_rho(Arc::clone(&rho_conn)).await; - // assert!(contains_iota(iota_id).await); - // etc. - } -}