diff --git a/Cargo.lock b/Cargo.lock index b3259bf..cc4af11 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -20,6 +20,7 @@ dependencies = [ "http", "hyper", "json", + "loom", "once_cell", "pkcs8", "rand 0.8.5", @@ -683,6 +684,20 @@ dependencies = [ "slab", ] +[[package]] +name = "generator" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "605183a538e3e2a9c1038635cc5c2d194e2ee8fd0d1b66b8349fad7dbacce5a2" +dependencies = [ + "cc", + "cfg-if", + "libc", + "log", + "rustversion", + "windows 0.61.3", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -1187,6 +1202,28 @@ version = "0.4.28" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "34080505efa8e45a4b816c349525ebe327ceaa8559756f0356cba97ef3bf7432" +[[package]] +name = "loom" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "419e0dc8046cb947daa77eb95ae174acfbddb7673b4151f56d1eed8e93fbfaca" +dependencies = [ + "cfg-if", + "generator", + "scoped-tls", + "tracing", + "tracing-subscriber", +] + +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "matchit" version = "0.8.4" @@ -2027,7 +2064,7 @@ dependencies = [ "ntapi", "once_cell", "rayon", - "windows", + "windows 0.52.0", ] [[package]] @@ -2319,10 +2356,14 @@ version = "0.3.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2054a14f5307d601f88daf0553e1cbf472acc4f2c51afab632431cdcd72124d5" dependencies = [ + "matchers", "nu-ansi-term", + "once_cell", + "regex-automata", "sharded-slab", "smallvec", "thread_local", + "tracing", "tracing-core", "tracing-log", ] @@ -2644,6 +2685,28 @@ dependencies = [ "windows-targets", ] +[[package]] +name = "windows" +version = "0.61.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893" +dependencies = [ + "windows-collections", + "windows-core 0.61.2", + "windows-future", + "windows-link 0.1.3", + "windows-numerics", +] + +[[package]] +name = "windows-collections" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3beeceb5e5cfd9eb1d76b381630e82c4241ccd0d27f1a39ed41b2760b255c5e8" +dependencies = [ + "windows-core 0.61.2", +] + [[package]] name = "windows-core" version = "0.52.0" @@ -2653,6 +2716,19 @@ dependencies = [ "windows-targets", ] +[[package]] +name = "windows-core" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link 0.1.3", + "windows-result 0.3.4", + "windows-strings 0.4.2", +] + [[package]] name = "windows-core" version = "0.62.1" @@ -2666,6 +2742,17 @@ dependencies = [ "windows-strings 0.5.0", ] +[[package]] +name = "windows-future" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e" +dependencies = [ + "windows-core 0.61.2", + "windows-link 0.1.3", + "windows-threading", +] + [[package]] name = "windows-implement" version = "0.60.1" @@ -2700,6 +2787,16 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "45e46c0661abb7180e7b9c281db115305d49ca1709ab8242adf09666d2173c65" +[[package]] +name = "windows-numerics" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1" +dependencies = [ + "windows-core 0.61.2", + "windows-link 0.1.3", +] + [[package]] name = "windows-registry" version = "0.5.3" @@ -2790,6 +2887,15 @@ dependencies = [ "windows_x86_64_msvc", ] +[[package]] +name = "windows-threading" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6" +dependencies = [ + "windows-link 0.1.3", +] + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" diff --git a/Cargo.toml b/Cargo.toml index 3fb4d12..434fe00 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ edition = "2024" [dependencies] tokio = { version = "*", features = ["full"] } -tokio-tungstenite = "*" +tokio-tungstenite = { version = "*" } tokio-util = { version = "*", features = ["full"] } hyper = { version = "*", features = ["full"] } futures-util = "*" @@ -41,3 +41,4 @@ dashmap = "*" warp = { version = "*" } chrono = "*" tokio-stream = "*" +loom = "0.7.2" diff --git a/src/auth/auth_connector.rs b/src/auth/auth_connector.rs index 4e9a6dc..6855a7a 100644 --- a/src/auth/auth_connector.rs +++ b/src/auth/auth_connector.rs @@ -1,13 +1,9 @@ use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes}; use crate::util::config_util::CONFIG; -use hex; -use json::JsonValue; -use reqwest::header::CONTENT_TYPE; use reqwest::{Client, Response}; -use sha2::{Digest, Sha256}; -use std::collections::HashMap; use std::time::Duration; use uuid::Uuid; + #[derive(Debug, Clone)] pub struct AuthUser { pub created_at: i64, @@ -30,12 +26,12 @@ fn client() -> Client { } pub async fn get_user(user_id: Uuid) -> Option { - let url = format!("https://auth.tensamin.methanium.net/api/get/{}/", user_id); + let url = format!("https://auth.tensamin.methanium.net/api/get/{}", user_id); let client = client(); let res = client.get(&url).send().await.ok()?; let json = res.text().await.ok()?; - let mut cv = CommunicationValue::from_json(&json); + let cv = CommunicationValue::from_json(&json); if cv.comm_type != CommunicationType::success { return None; } @@ -77,10 +73,7 @@ pub async fn get_iota_id(user_id: Uuid) -> Option { let client = client(); let res = client .get(&url) - .header( - "Authorization", - CONFIG.read().unwrap().omikron_id.to_string(), - ) + .header("Authorization", CONFIG.lock().await.omikron_id.to_string()) .header("Content-Type", "application/json") .send() .await @@ -89,7 +82,7 @@ pub async fn get_iota_id(user_id: Uuid) -> Option { let json = res.text().await.ok()?; let json = json.replace("iota_uuid", "iota_id"); - let mut cv = CommunicationValue::from_json(&json); + let cv = CommunicationValue::from_json(&json); if cv.comm_type != CommunicationType::success { return None; } @@ -99,17 +92,15 @@ pub async fn get_iota_id(user_id: Uuid) -> Option { } pub async fn is_private_key_valid(user_id: Uuid, pk_hash: &str) -> bool { let url = format!( - "https://auth.tensamin.methanium.net/api/get/private-key-hash/{}", + "https://auth.tensamin.methanium.net/api/get/private-key-hash/{}/", user_id ); let client = client(); + println!("Auth: {}", CONFIG.lock().await.omikron_id); let res = client .get(&url) - .header( - "Authorization", - CONFIG.read().unwrap().omikron_id.to_string(), - ) + .header("Authorization", CONFIG.lock().await.omikron_id.to_string()) .header("PrivateKeyHash", pk_hash) .header("Accept", "application/json") .send() @@ -123,7 +114,8 @@ pub async fn is_private_key_valid(user_id: Uuid, pk_hash: &str) -> bool { return false; }; - let mut cv = CommunicationValue::from_json(&body); + let cv = CommunicationValue::from_json(&body); + println!("Auth: {}", &body); if cv.comm_type != CommunicationType::success { return false; } @@ -135,7 +127,7 @@ pub async fn is_private_key_valid(user_id: Uuid, pk_hash: &str) -> bool { } pub async fn get_public_key(user_id: Uuid) -> Option { let url = format!( - "https://auth.tensamin.methanium.net/api/{}/public-key/", + "https://auth.tensamin.methanium.net/api/{}/public-key", user_id ); @@ -158,12 +150,12 @@ pub async fn get_public_key(user_id: Uuid) -> Option { } pub async fn get_register() -> Option { - let url = "https://auth.tensamin.methanium.net/api/register/init/".to_string(); + let url = "https://auth.tensamin.methanium.net/api/register/init".to_string(); let client = client(); let res = client.get(&url).send().await.ok()?; let json = res.text().await.ok()?; - let mut cv = CommunicationValue::from_json(&json); + let cv = CommunicationValue::from_json(&json); Uuid::parse_str(&*cv.get_data(DataTypes::user_id).unwrap().to_string()).ok() } diff --git a/src/calls/call_connection.rs b/src/calls/call_connection.rs index a203154..d147da5 100644 --- a/src/calls/call_connection.rs +++ b/src/calls/call_connection.rs @@ -1,17 +1,15 @@ -// part of calls or in your ws-server file - use std::sync::Arc; -use futures::{SinkExt, StreamExt, channel::mpsc::UnboundedSender, lock::Mutex}; -use json::JsonValue; -use tokio::sync::mpsc::unbounded_channel; -use tungstenite::{Message, Utf8Bytes}; -use uuid::Uuid; - use crate::{ - calls::{call_group::CallGroup, call_group::Caller, call_manager::CallManagerState}, + calls::{call_manager::CallManagerState, caller::Caller}, data::communication::{CommunicationType, CommunicationValue, DataTypes}, }; +use futures::{SinkExt, StreamExt, lock::Mutex}; +use json::JsonValue; +use tokio::sync::mpsc::UnboundedSender; +use tokio::sync::mpsc::unbounded_channel; +use tungstenite::Utf8Bytes; +use uuid::Uuid; pub async fn handle_connection( raw_stream: tokio::net::TcpStream, @@ -21,11 +19,11 @@ pub async fn handle_connection( .await .expect("Error during the websocket handshake"); - let (mut outgoing, mut incoming) = ws_stream.split(); + let (outgoing, mut incoming) = ws_stream.split(); // We create a channel so other parts can send to this client pub type Tx = UnboundedSender; - let (tx, mut rx): (_, _) = unbounded_channel(); + let (tx, mut rx): (UnboundedSender<_>, _) = unbounded_channel(); // Spawn a task to forward from rx → outgoing let mut outgoing_clone = outgoing; @@ -48,89 +46,64 @@ pub async fn handle_connection( }; if let tokio_tungstenite::tungstenite::Message::Text(s) = msg { // parse CommunicationValue - if let mut cv = CommunicationValue::from_json(&s) { - match cv.comm_type { - CommunicationType::identification => { - // extract fields - let user_str = cv.get_data(DataTypes::user_id).unwrap().as_str().unwrap(); - let call_str = cv.get_data(DataTypes::call_id).unwrap().as_str().unwrap(); - let _secret_sha = cv - .get_data(DataTypes::call_secret_sha) - .unwrap() - .as_str() - .unwrap(); + let cv = CommunicationValue::from_json(&s); + match cv.comm_type { + CommunicationType::identification => { + // extract fields + let user_str = cv.get_data(DataTypes::user_id).unwrap().as_str().unwrap(); + let call_str = cv.get_data(DataTypes::call_id).unwrap().as_str().unwrap(); + let _secret_sha = cv + .get_data(DataTypes::call_secret_sha) + .unwrap() + .as_str() + .unwrap(); - let user_id = Uuid::parse_str(user_str).ok(); - let call_id = Uuid::parse_str(call_str).ok(); - if let (Some(uid), Some(cid)) = (user_id, call_id) { - maybe_user_id = Some(uid); - maybe_call_id = Some(cid); - // Add to call group - let mut st = state.lock().await; - let group = state.lock().await.get_or_create_group(cid, &_secret_sha); - group.clone().add_member(uid, tx.clone()); - - // Response: identification_response plus states - let mut response = - CommunicationValue::new(CommunicationType::identification_response) - .with_id(cv.get_id().clone()); - // build user states - let mut users = JsonValue::new_object(); - for c in group.callers.values() { - let mut user_info = JsonValue::new_object(); - user_info.insert("state", JsonValue::from("muted".to_string())); - user_info.insert("streaming", JsonValue::from(false)); + let user_id = Uuid::parse_str(user_str).ok(); + let call_id = Uuid::parse_str(call_str).ok(); + if let (Some(uid), Some(cid)) = (user_id, call_id) { + maybe_user_id = Some(uid); + maybe_call_id = Some(cid); + // Add to call group + let group = state.lock().await.get_or_create_group(cid, &_secret_sha); + group.lock().await.add_member(uid, tx.clone()); + // Response: identification_response plus states + let mut response = + CommunicationValue::new(CommunicationType::identification_response) + .with_id(cv.get_id().clone()); + // build user states + let mut users = JsonValue::new_object(); + for c in group.lock().await.callers.values() { + let mut user_info = JsonValue::new_object(); + let _ = user_info.insert("state", JsonValue::from("muted".to_string())); + let _ = user_info.insert("streaming", JsonValue::from(false)); + let _ = users.insert(&c.user_id.to_string(), JsonValue::from(user_info)); - } - response = response.add_data(DataTypes::about, JsonValue::from(users)); - // send back - let _ = tx.send(Utf8Bytes::from(response.to_json().to_string())); } + response = response.add_data(DataTypes::about, JsonValue::from(users)); + // send back + let _ = tx.send(Utf8Bytes::from(response.to_json().to_string())); } + } - CommunicationType::ping => { - // optionally parse LAST_PING - // reply with PONG with same message_id - let mut resp = CommunicationValue::new(CommunicationType::pong) - .with_id(cv.get_id().clone()); - let _ = tx.send(Utf8Bytes::from(resp.to_json().to_string())); - } + CommunicationType::ping => { + // optionally parse LAST_PING + // reply with PONG with same message_id + let resp = CommunicationValue::new(CommunicationType::pong) + .with_id(cv.get_id().clone()); + let _ = tx.send(Utf8Bytes::from(resp.to_json().to_string())); + } - CommunicationType::client_changed => { - match (maybe_user_id, cv.get_data(DataTypes::call_state)) { - (Some(uid), Some(JsonValue::String(state_str))) => { - if let Some(cid) = maybe_call_id { - { - let mut st = state.lock().await; - if let Some(group) = st.call_groups.get_mut(&cid) { - 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()); - } - } - } - } - _ => (), - } - } - - CommunicationType::start_stream | CommunicationType::end_stream => { - if let Some(uid) = maybe_user_id { + CommunicationType::client_changed => { + match (maybe_user_id, cv.get_data(DataTypes::call_state)) { + (Some(uid), Some(JsonValue::String(state_str))) => { if let Some(cid) = maybe_call_id { - let mut st = state.lock().await; - if let Some(group) = st.call_groups.get_mut(&cid) { - if let Some(caller) = group.caller_state_mut(&uid) { - let streaming = - cv.comm_type == CommunicationType::start_stream; - // set streaming - // caller.streaming = streaming; // if you store streaming - } + { + let mut st = state.lock().await; + let mut group = + st.call_groups.get_mut(&cid).unwrap().lock().await; + let caller = group.caller_state_mut(&uid).unwrap(); + caller_state_change(caller, &state_str); + let mut bc = cv.clone(); bc = bc.add_data( DataTypes::sender_id, @@ -140,35 +113,57 @@ pub async fn handle_connection( } } } + _ => (), } + } - CommunicationType::webrtc_sdp - | CommunicationType::webrtc_ice - | CommunicationType::watch_stream => { - if let Some(uid) = maybe_user_id { - 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(cid) = maybe_call_id { - let mut st = state.lock().await; - if let Some(group) = st.call_groups.get(&cid) { - let mut bc = cv.clone(); - bc = bc.add_data( - DataTypes::sender_id, - JsonValue::String(uid.to_string()), - ); - group.send_to(&receiver_id, &bc.to_json().to_string()); - } + CommunicationType::start_stream | CommunicationType::end_stream => { + if let Some(uid) = maybe_user_id { + if let Some(cid) = maybe_call_id { + let mut st = state.lock().await; + let mut group = st.call_groups.get_mut(&cid).unwrap().lock().await; + if let Some(caller) = group.caller_state_mut(&uid) { + let streaming = cv.comm_type == CommunicationType::start_stream; + // set streaming + // caller.streaming = streaming; // if you store streaming + } + let mut bc = cv.clone(); + bc = bc + .add_data(DataTypes::sender_id, JsonValue::String(uid.to_string())); + group.broadcast(&bc.to_json().to_string()); + } + } + } + + CommunicationType::webrtc_sdp + | CommunicationType::webrtc_ice + | CommunicationType::watch_stream => { + if let Some(uid) = maybe_user_id { + 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(cid) = maybe_call_id { + let st = state.lock().await; + if let Some(group) = st.call_groups.get(&cid) { + 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()); } } } } } + } - _ => { - // other types you may handle - } + _ => { + // other types you may handle } } } @@ -178,9 +173,9 @@ pub async fn handle_connection( if let (Some(uid), Some(cid)) = (maybe_user_id, maybe_call_id) { let mut st = state.lock().await; if let Some(group) = st.call_groups.get_mut(&cid) { - group.remove_member(&uid); + group.lock().await.remove_member(uid); } - st.remove_inactive(); + st.remove_inactive().await; } } diff --git a/src/calls/call_group.rs b/src/calls/call_group.rs index 44f29fb..3d196de 100644 --- a/src/calls/call_group.rs +++ b/src/calls/call_group.rs @@ -1,106 +1,47 @@ -use std::{ - collections::{HashMap, HashSet}, - sync::Arc, -}; +use crate::calls::caller::Caller; +use std::collections::HashMap; use tokio::sync::mpsc::UnboundedSender; - -use json::JsonValue; use tungstenite::Utf8Bytes; use uuid::Uuid; -use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes}; - -pub type Tx = UnboundedSender; - -#[derive(Debug)] pub struct CallGroup { pub call_id: Uuid, - secret: String, pub callers: HashMap, - pub inform_on_end: HashSet, - pub started_at: u64, - pub ended_at: Option, } impl CallGroup { - pub fn new(call_id: Uuid, secret: String) -> Self { - CallGroup { + pub fn new(call_id: Uuid) -> Self { + Self { call_id, - secret, callers: HashMap::new(), - inform_on_end: HashSet::new(), - started_at: chrono::Utc::now().timestamp_millis() as u64, - ended_at: None, } } - pub fn add_member(mut self, user_id: Uuid, tx: Tx) { - // If already present, drop old - if let Some(old) = self.callers.remove(&user_id) { - // optionally notify or close old - // no direct close since we don’t hold session here - } - // Notify existing about new client - let mut cv = CommunicationValue::new(CommunicationType::client_connected); - cv = cv - .add_data(DataTypes::user_id, JsonValue::String(user_id.to_string())) - .add_data( - DataTypes::call_state, - JsonValue::String("muted".to_string()), - ); - // broadcast - let msg = cv.to_json().to_string(); - for c in self.callers.values() { - let _ = c.tx.send(Utf8Bytes::from(msg.clone())); - } + pub fn add_member(&mut self, user_id: Uuid, tx: UnboundedSender) { self.callers.insert(user_id, Caller { user_id, tx }); } - pub fn remove_member(&mut self, user_id: &Uuid) { - if self.callers.remove(user_id).is_some() { - let cv = CommunicationValue::new(CommunicationType::client_closed) - .add_data(DataTypes::user_id, JsonValue::String(user_id.to_string())) - .to_json() - .to_string(); - for c in self.callers.values() { - let _ = c.tx.send(Utf8Bytes::from(cv.clone())); - } - } - if self.callers.is_empty() { - self.ended_at = Some(chrono::Utc::now().timestamp_millis() as u64); - // Optionally notify inform_on_end - let end_msg = CommunicationValue::new(CommunicationType::end_call) - .add_data( - DataTypes::call_id, - JsonValue::String(self.call_id.to_string()), - ) - .to_json() - .to_string(); - for &uid in &self.inform_on_end { - // here you’d send via your Rho / other channel to user - // e.g. RhoManager::message_to(uid, end_msg.clone()); - } + pub fn remove_member(&mut self, user_id: Uuid) { + self.callers.remove(&user_id); + } + + pub fn is_empty(&self) -> bool { + self.callers.is_empty() + } + + pub fn send_to(&self, user_id: &Uuid, message: &str) { + if let Some(caller) = self.callers.get(user_id) { + let _ = caller.send(Utf8Bytes::from(message.to_string())); } } - pub fn broadcast(&self, msg: &str) { - for c in self.callers.values() { - let _ = c.tx.send(Utf8Bytes::from(msg.clone())); + pub fn broadcast(&self, message: &str) { + for caller in self.callers.values() { + let _ = caller.send(Utf8Bytes::from(message.to_string())); } } - pub fn send_to(&self, target: &Uuid, msg: &str) { - if let Some(c) = self.callers.get(target) { - let _ = c.tx.send(Utf8Bytes::from(msg.clone())); - } - } - - pub fn caller_state_mut(&mut self, uid: &Uuid) -> Option<&mut Caller> { - self.callers.get_mut(uid) + pub fn caller_state_mut(&mut self, user_id: &Uuid) -> Option<&mut Caller> { + self.callers.get_mut(user_id) } } -#[derive(Debug)] -pub struct Caller { - pub user_id: Uuid, - pub tx: Tx, -} diff --git a/src/calls/call_manager.rs b/src/calls/call_manager.rs index 1f6b04c..ba96fa4 100644 --- a/src/calls/call_manager.rs +++ b/src/calls/call_manager.rs @@ -1,19 +1,15 @@ use crate::calls::call_group::CallGroup; -use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes}; -use futures::{SinkExt, StreamExt}; -use json::JsonValue; -use std::collections::{HashMap, HashSet}; -use std::sync::{Arc, Mutex}; -use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}; +use futures::lock::Mutex; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::mpsc::UnboundedSender; use tungstenite::Utf8Bytes; use uuid::Uuid; -use warp::Filter; pub type Tx = UnboundedSender; -#[derive(Debug)] pub struct CallManagerState { - pub call_groups: HashMap>, + pub call_groups: HashMap>>, } impl CallManagerState { @@ -23,18 +19,27 @@ impl CallManagerState { } } - pub fn get_or_create_group(&mut self, call_id: Uuid, secret: &str) -> Arc { + pub fn get_or_create_group(&mut self, call_id: Uuid, secret: &str) -> Arc> { let g = self.call_groups.get_mut(&call_id); if let Some(group) = g { group.clone() } else { - let cg = Arc::new(CallGroup::new(call_id, secret.to_string())); + let cg = Arc::new(Mutex::new(CallGroup::new(call_id))); self.call_groups.insert(call_id, cg); self.call_groups.get(&call_id).unwrap().clone() } } - pub fn remove_inactive(&mut self) { - self.call_groups.retain(|_k, v| v.callers.len() > 0); + pub async fn remove_inactive(&mut self) { + let mut rem = Vec::new(); + for cg in self.call_groups.keys() { + let group = self.call_groups.get(cg).unwrap().lock().await; + if group.callers.is_empty() { + rem.push(cg.clone()); + } + } + for cg in rem { + self.call_groups.remove(&cg); + } } } diff --git a/src/calls/caller.rs b/src/calls/caller.rs new file mode 100644 index 0000000..452ebf9 --- /dev/null +++ b/src/calls/caller.rs @@ -0,0 +1,14 @@ +use tokio::sync::mpsc::UnboundedSender; +use tungstenite::Utf8Bytes; +use uuid::Uuid; + +#[derive(Debug)] +pub struct Caller { + pub user_id: Uuid, + pub tx: UnboundedSender, +} +impl Caller { + pub fn send(&self, msg: impl Into) { + let _ = self.tx.send(msg.into()); + } +} diff --git a/src/calls/mod.rs b/src/calls/mod.rs index 360fafa..98613e7 100644 --- a/src/calls/mod.rs +++ b/src/calls/mod.rs @@ -1,3 +1,4 @@ pub mod call_connection; pub mod call_group; pub mod call_manager; +pub mod caller; diff --git a/src/data/communication.rs b/src/data/communication.rs index 846a602..2a771da 100644 --- a/src/data/communication.rs +++ b/src/data/communication.rs @@ -1,9 +1,6 @@ -use axum::Json; use json::number::Number; -use json::{Array, JsonValue, object, parse, stringify}; -use std::any::Any; +use json::{Array, JsonValue, object, parse}; use std::collections::HashMap; -use std::env::VarsOs; use std::str::FromStr; use std::time::{SystemTime, UNIX_EPOCH}; use uuid::Uuid; @@ -11,6 +8,7 @@ use uuid::Uuid; #[derive(Eq, Hash, PartialEq, Clone, Debug)] pub enum DataTypes { error_type, + accepted_ids, uuid, chat_partner_id, iota_id, @@ -157,6 +155,9 @@ impl DataTypes { #[derive(PartialEq, Clone, Debug)] pub enum CommunicationType { error, + error_invalid_user_id, + error_no_iota, + error_invalid_private_key, success, message, message_send, @@ -366,9 +367,9 @@ impl CommunicationValue { pub fn ack_message(message_id: Uuid, sender: Uuid) -> CommunicationValue { let mut cv = CommunicationValue::new(CommunicationType::message).with_id(message_id); - if let s = sender { - cv = cv.add_data(DataTypes::send_time, JsonValue::String(s.to_string())); - } + 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 { diff --git a/src/main.rs b/src/main.rs index 0ee3b59..64288cd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,19 +5,103 @@ mod omega; mod rho; mod util; -use crate::rho::client_connection; -use crate::rho::iota_connection; -use crate::util::config_util::Config; -use crate::util::file_util; -use tokio_websocket_server::socket::{WebSocketMessage, WebsocketServer}; -use tracing::info; -use tracing_subscriber::fmt; +use futures::StreamExt; +use std::sync::Arc; +use tokio::net::TcpListener; +use tokio_tungstenite::accept_hdr_async; +use tungstenite::handshake::server::{Request, Response}; +use crate::{ + omega::omega_connection::OmegaConnection, + rho::{client_connection::ClientConnection, iota_connection::IotaConnection}, +}; #[tokio::main] async fn main() { - fmt::init(); + OmegaConnection::new().connect().await; - loop { - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + let listener = TcpListener::bind("0.0.0.0:959").await.unwrap(); + println!("WebSocket server listening on 0.0.0.0:959"); + + while let Ok((stream, _)) = listener.accept().await { + tokio::spawn(async move { + let mut path: String = "/".to_string(); + let callback = |req: &Request, response: Response| { + path = format!("{}", &req.uri().path()); + Ok(response) + }; + let ws_stream = match accept_hdr_async(stream, callback).await { + Ok(ws) => ws, + Err(e) => { + eprintln!("WebSocket upgrade failed: {}", e); + return; + } + }; + println!("New {} connection", path); + if path == "/ws/client/" { + let client_conn: Arc = + Arc::from(ClientConnection::new(ws_stream)); + loop { + let msg_result = { + let mut session_lock = client_conn.session.lock().await; + session_lock.next().await + }; + + match msg_result { + Some(Ok(msg)) => { + if msg.is_text() { + let text = msg.into_text().unwrap(); + client_conn.clone().handle_message(text).await; + } else if msg.is_close() { + println!("Client disconnected"); + client_conn.handle_close().await; + return; + } + } + Some(Err(e)) => { + eprintln!("WebSocket error: {}", e); + client_conn.handle_close().await; + return; + } + None => { + println!("Client stream ended"); + client_conn.handle_close().await; + return; + } + } + } + } else if path == "/ws/iota/" { + let iota_conn: Arc = Arc::from(IotaConnection::new(ws_stream)); + loop { + let msg_result = { + let mut session_lock = iota_conn.session.lock().await; + session_lock.next().await + }; + + match msg_result { + Some(Ok(msg)) => { + if msg.is_text() { + let text = msg.into_text().unwrap(); + iota_conn.clone().handle_message(text).await; + } else if msg.is_close() { + println!("Iota disconnected"); + iota_conn.handle_close().await; + return; + } + } + Some(Err(e)) => { + eprintln!("WebSocket error: {}", e); + iota_conn.handle_close().await; + return; + } + None => { + // Stream ended + println!("Iota stream ended"); + iota_conn.handle_close().await; + return; + } + } + } + } + }); } } diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index 9202cd1..8a5b18a 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -1,25 +1,21 @@ use std::sync::Arc; use std::time::Duration; -use std::{collections::HashMap, net::TcpStream}; use crate::{ data::{ communication::DataTypes, user::{User, UserStatus}, }, - rho::{self, rho_manager}, + rho::rho_manager, util::config_util::CONFIG, }; -use axum::Json; use dashmap::DashMap; use futures::{SinkExt, StreamExt}; use json::JsonValue; use once_cell::sync::Lazy; use tokio::sync::Mutex; use tokio::time::sleep; -use tokio_tungstenite::{ - MaybeTlsStream, WebSocketStream, connect_async, tungstenite::protocol::Message, -}; +use tokio_tungstenite::{connect_async, tungstenite::protocol::Message}; use tungstenite::Utf8Bytes; use uuid::Uuid; @@ -43,33 +39,26 @@ impl OmegaConnection { self.connect_internal(0).await; } async fn connect_internal(&self, mut retry: usize) { - let uri = "wss://tensamin.methanium.net/ws/omega/"; - loop { - if retry > 20 { + if retry > 5 { eprintln!("Max retry attempts reached, giving up."); return; } - match connect_async(uri).await { - Ok((ws_stream, _)) => { - let mut guard = Some(ws_stream); - - // Send IDENTIFICATION + match connect_async("wss://tensamin.methanium.net/ws/omega").await { + Ok((_, _)) => { + retry = 0; let identify_msg = CommunicationValue::new(CommunicationType::identification) .add_data( DataTypes::uuid, - JsonValue::String(CONFIG.read().unwrap().omikron_id.to_string()), + JsonValue::String(CONFIG.lock().await.omikron_id.to_string()), ); self.send_message(&identify_msg).await; - // Spawn reader loop let ws_stream_clone = self.ws_stream.clone(); tokio::spawn(async move { OmegaConnection::read_loop(ws_stream_clone).await; }); - - break; // success, exit the loop } Err(e) => { eprintln!("WebSocket connection failed (attempt {}): {}", retry, e); @@ -92,7 +81,7 @@ impl OmegaConnection { match ws.next().await { Some(Ok(Message::Text(msg))) => { - let mut cv = CommunicationValue::from_json(&msg); + let cv = CommunicationValue::from_json(&msg); let msg_id = cv.get_id(); // Handle waiting tasks @@ -122,16 +111,14 @@ impl OmegaConnection { let user = User::new(iota_id, user_id, status); for rho_con in rho_manager::get_all_connections().await { - rho_con.are_they_interested(&user); + rho_con.are_they_interested(&user).await; } } } Some(Ok(Message::Close(_))) | None => { - OmegaConnection::reconnect(); break; } Some(Err(_)) => { - OmegaConnection::reconnect(); break; } _ => {} @@ -139,11 +126,6 @@ impl OmegaConnection { } } - async fn reconnect() { - let conn = OmegaConnection::new(); - conn.connect().await; - } - pub async fn send_message(&self, cv: &CommunicationValue) { let mut guard = self.ws_stream.lock().await; if let Some(ws) = guard.as_mut() { @@ -192,7 +174,7 @@ impl OmegaConnection { WAITING_TASKS.insert( msg_id, Box::new(move |response: CommunicationValue| { - Box::pin(async move |response2: CommunicationValue| { + let _ = Box::pin(async move |_: CommunicationValue| { let rho = rho_manager::get_rho_con_for_user(user_id).await; if let Some(rho) = rho { for client in rho.get_client_connections_for_user(user_id).await { diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 170e4d0..be5d4f6 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -1,10 +1,7 @@ use futures::SinkExt; -use http::header::AUTHORIZATION; -use std::{ - collections::HashMap, - sync::{Arc, Weak}, -}; -use tokio::sync::{Mutex, RwLock}; +use std::sync::{Arc, Weak}; +use tokio::sync::Mutex; +use tokio::sync::RwLock; use tokio_tungstenite::{WebSocketStream, tungstenite::Message}; use tungstenite::Utf8Bytes; use uuid::Uuid; @@ -23,17 +20,17 @@ use crate::{ /// ClientConnection represents a WebSocket connection from a client device pub struct ClientConnection { /// WebSocket session - session: Arc>>, + pub session: Arc>>, /// User ID associated with this client - user_id: Arc>>, + pub user_id: Arc>>, /// Whether this connection has been identified/authenticated - identified: Arc>, + pub identified: Arc>, /// Ping latency tracking - ping: Arc>, + pub ping: Arc>, /// Weak reference to RhoConnection to avoid circular references - rho_connection: Arc>>>, + pub rho_connection: Arc>>>, /// List of user IDs this client is interested in receiving updates about - interested_users: Arc>>, + pub interested_users: Arc>>, } impl ClientConnection { @@ -97,7 +94,7 @@ impl ClientConnection { } /// Handle incoming message from client - pub async fn handle_message(self: Arc, message: String) { + pub async fn handle_message(self: Arc, message: Utf8Bytes) { let cv = CommunicationValue::from_json(&message); // Handle identification @@ -145,23 +142,16 @@ impl ClientConnection { Some(id_str) => match Uuid::parse_str(&id_str.to_string()) { Ok(id) => id, Err(_) => { - self.send_error_response(&cv.get_id(), CommunicationType::error) - .await; + self.send_error_response( + &cv.get_id(), + CommunicationType::error_invalid_user_id, + ) + .await; return; } }, None => { - self.send_error_response(&cv.get_id(), CommunicationType::error) - .await; - return; - } - }; - - // Find RhoConnection for this user - let rho_connection = match rho_manager::get_rho_con_for_user(user_id).await { - Some(rho) => rho, - None => { - self.send_error_response(&cv.get_id(), CommunicationType::error) + self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_user_id) .await; return; } @@ -169,20 +159,36 @@ impl ClientConnection { // Validate private key if let Some(private_key_hash) = cv.get_data(DataTypes::private_key_hash) { + println!("private_key_hash: {}", private_key_hash); let is_valid = auth_connector::is_private_key_valid(user_id, &private_key_hash.to_string()).await; if !is_valid { - self.send_error_response(&cv.get_id(), CommunicationType::error) - .await; + println!("Invalid private key"); + self.send_error_response( + &cv.get_id(), + CommunicationType::error_invalid_private_key, + ) + .await; return; } } else { - self.send_error_response(&cv.get_id(), CommunicationType::error) + println!("Missing private key"); + self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_private_key) .await; return; } + // Find RhoConnection for this user + let rho_connection = match rho_manager::get_rho_con_for_user(user_id).await { + Some(rho) => rho, + None => { + self.send_error_response(&cv.get_id(), CommunicationType::error_no_iota) + .await; + return; + } + }; + // Set identification data { let mut user_id_guard = self.user_id.write().await; @@ -193,14 +199,11 @@ impl ClientConnection { *identified_guard = true; } - // Set up RhoConnection reference self.set_rho_connection(Arc::downgrade(&rho_connection)) .await; - // Add this client to the RhoConnection rho_connection.add_client_connection(Arc::from(sarc)).await; - // Send success response let response = CommunicationValue::new(CommunicationType::identification_response) .with_id(cv.get_id()); self.send_message(&response).await; @@ -242,14 +245,15 @@ impl ClientConnection { rho_conn.get_iota_id().await, user_id, user_status, - ); + ) + .await; } } } } /// Handle call invite - async fn handle_call_invite(&self, mut cv: CommunicationValue) { + async fn handle_call_invite(&self, cv: CommunicationValue) { let receiver_id = match cv.get_data(DataTypes::receiver_id) { Some(id_str) => match Uuid::parse_str(&id_str.to_string()) { Ok(id) => id, @@ -470,46 +474,3 @@ impl std::fmt::Debug for ClientConnection { .finish() } } - -#[cfg(test)] -mod tests { - use super::*; - use tokio::sync::Mutex; - - #[tokio::test] - async fn test_client_connection_creation() { - // Mock session - in real implementation this would be a proper WebSocket stream - let mock_session = tokio_tungstenite::WebSocketStream::from_raw_socket( - tokio::net::TcpStream::connect("127.0.0.1:0") - .await - .unwrap_or_else(|_| { - // This is just for testing, create a dummy stream - panic!("Cannot create test stream") - }), - tokio_tungstenite::tungstenite::protocol::Role::Client, - None, - ) - .await; - - // This test would fail in practice due to the mock stream - // but shows the intended API - // let client_conn = ClientConnection::new(mock_session); - - // assert!(!client_conn.is_identified().await); - // assert_eq!(client_conn.get_ping().await, -1); - // assert!(client_conn.get_user_id().await.is_none()); - } - - #[tokio::test] - async fn test_interested_users() { - // This would also need a proper mock setup - // but shows the intended functionality - - // let client_conn = ClientConnection::new(mock_session); - // let user_ids = vec![Uuid::new_v4(), Uuid::new_v4()]; - - // client_conn.set_interested_users(user_ids.clone()).await; - // - // // Test would verify that the interested users are stored correctly - } -} diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 3529045..ce6ccda 100644 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -1,5 +1,5 @@ use futures::SinkExt; -use json::{JsonValue, number::Number}; +use json::JsonValue; use std::{ collections::HashMap, sync::{Arc, Weak}, @@ -13,21 +13,18 @@ use super::{rho_connection::RhoConnection, rho_manager}; use crate::{ auth::auth_connector, // calls::call_manager::CallManager, - data::{ - communication::{CommunicationType, CommunicationValue, DataTypes}, - user::User, - }, + data::communication::{CommunicationType, CommunicationValue, DataTypes}, omega::omega_connection::OmegaConnection, }; /// IotaConnection represents a WebSocket connection from an Iota device pub struct IotaConnection { - session: Arc>>, - iota_id: Arc>, - user_ids: Arc>>, - identified: Arc>, - ping: Arc>, - rho_connection: Arc>>>, + pub session: Arc>>, + pub iota_id: Arc>, + pub user_ids: Arc>>, + pub identified: Arc>, + pub ping: Arc>, + pub rho_connection: Arc>>>, } impl IotaConnection { @@ -43,22 +40,6 @@ impl IotaConnection { }) } - /// Create with known IDs (for testing) - pub fn new_with_ids( - iota_id: Uuid, - user_ids: Vec, - session: Arc>>, - ) -> Arc { - Arc::new(Self { - session, - iota_id: Arc::new(RwLock::new(iota_id)), - user_ids: Arc::new(RwLock::new(user_ids)), - identified: Arc::new(RwLock::new(true)), - ping: Arc::new(RwLock::new(0)), - rho_connection: Arc::new(RwLock::new(None)), - }) - } - /// Get the Iota ID pub async fn get_iota_id(&self) -> Uuid { *self.iota_id.read().await @@ -98,18 +79,24 @@ impl IotaConnection { /// Send a message to the Iota pub async fn send_message_str(&self, message: &str) { let mut session = self.session.lock().await; - let _ = session + if let Err(e) = session .send(Message::Text(Utf8Bytes::from(message.to_string()))) - .await; + .await + { + eprintln!("Failed to send WebSocket message: {:?}", e); + } } /// Send a CommunicationValue to the Iota pub async fn send_message(&self, cv: CommunicationValue) { + if !cv.is_type(CommunicationType::pong) { + println!("{}", cv.to_json().to_string()); + } self.send_message_str(&cv.to_json().to_string()).await; } /// Handle incoming message from Iota - pub async fn handle_message(self: Arc, message: String) { + pub async fn handle_message(self: Arc, message: Utf8Bytes) { let cv = CommunicationValue::from_json(&message); // Handle identification @@ -149,7 +136,7 @@ impl IotaConnection { } /// Handle identification message - async fn handle_identification(self: Arc, mut cv: CommunicationValue) { + async fn handle_identification(self: Arc, cv: CommunicationValue) { // Parse Iota ID let iota_id = match cv.get_data(DataTypes::iota_id) { Some(id_str) => match Uuid::parse_str(&id_str.to_string()) { @@ -166,16 +153,17 @@ impl IotaConnection { }; // Parse user IDs - let mut validated_user_ids = Vec::new(); + let mut validated_user_ids: Vec = Vec::new(); if let Some(user_ids_str) = cv.get_data(DataTypes::user_ids) { for id_str in user_ids_str.to_string().split(',') { if id_str.is_empty() { continue; } if let Ok(user_id) = Uuid::parse_str(id_str.trim()) { - let auth_iota_id = auth_connector::get_iota_id(user_id).await.unwrap(); - if auth_iota_id == iota_id { - validated_user_ids.push(user_id); + if let Some(auth_iota_id) = auth_connector::get_iota_id(user_id).await { + if auth_iota_id == iota_id { + validated_user_ids.push(user_id); + } } } } @@ -214,16 +202,20 @@ impl IotaConnection { rho_manager::add_rho(rho_connection).await; // Send response + let mut str = String::new(); + for id in &validated_user_ids { + str.push_str(&format!(",{}", id)); + } let response = CommunicationValue::new(CommunicationType::identification_response) .with_id(cv.get_id()) + .add_data_str(DataTypes::accepted_ids, str) .add_data_str(DataTypes::accepted, validated_user_ids.len().to_string()); self.send_message(response).await; } /// Handle ping message - async fn handle_ping(&self, mut cv: CommunicationValue) { - // Update our ping if provided + async fn handle_ping(&self, cv: CommunicationValue) { if let Some(last_ping) = cv.get_data(DataTypes::last_ping) { if let Ok(ping_val) = last_ping.to_string().parse::() { let mut ping_guard = self.ping.write().await; @@ -231,14 +223,12 @@ impl IotaConnection { } } - // Collect client pings let client_pings = if let Some(rho_conn) = self.get_rho_connection().await { rho_conn.get_client_pings().await } else { HashMap::new() }; - // Send pong response let pings = client_pings .into_iter() .map(|(k, v)| (k, JsonValue::String(v.to_string()))) @@ -252,18 +242,15 @@ impl IotaConnection { /// Handle message forwarding to other Iotas async fn handle_forward_message(&self, cv: CommunicationValue) { let receiver_id = cv.get_receiver(); - // Validate sender if present let sender_id = cv.get_sender(); if !self.get_user_ids().await.contains(&sender_id) { self.send_error_response(&cv.get_id()).await; return; } - // Find target RhoConnection if let Some(target_rho) = rho_manager::get_rho_con_for_user(receiver_id).await { target_rho.message_to_iota(cv).await; } else { - // Send error if target not found let error = CommunicationValue::new(CommunicationType::error) .with_id(cv.get_id()) .with_sender(cv.get_sender()); @@ -298,7 +285,7 @@ impl IotaConnection { } // Notify OmegaConnection about user states - OmegaConnection::user_states(receiver_id, interested_ids.clone()); + OmegaConnection::user_states(receiver_id, interested_ids.clone()).await; // Set interested users in RhoConnection if let Some(rho_conn) = self.get_rho_connection().await { @@ -318,13 +305,11 @@ impl IotaConnection { } } - /// Send error response async fn send_error_response(&self, message_id: &Uuid) { let error = CommunicationValue::new(CommunicationType::error).with_id(*message_id); self.send_message(error).await; } - /// Handle connection close pub async fn handle_close(&self) { if self.is_identified().await { if let Some(rho_conn) = self.get_rho_connection().await { diff --git a/src/rho/mod.rs b/src/rho/mod.rs index 99067bc..72799fc 100644 --- a/src/rho/mod.rs +++ b/src/rho/mod.rs @@ -1,21 +1,4 @@ -//! Rho module - Connection management between Iota and Client connections -//! -//! This module implements the Rho connection management system that maps -//! single Iota connections to multiple Client connections, providing -//! bidirectional communication capabilities. - pub mod client_connection; pub mod iota_connection; pub mod rho_connection; pub mod rho_manager; - -// Re-export commonly used types for convenience -pub use client_connection::ClientConnection; -pub use iota_connection::IotaConnection; -pub use rho_connection::RhoConnection; - -// Re-export key manager functions -pub use rho_manager::{ - add_rho, connection_count, contains_iota, get_all_connections, get_rho_by_iota, - get_rho_con_for_user, remove_rho, -}; diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 2b6bca6..35d7448 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -1,5 +1,5 @@ -use std::collections::{self, HashMap}; -use std::sync::{Arc, Weak}; +use std::collections::HashMap; +use std::sync::Arc; use tokio::sync::RwLock; use uuid::Uuid; @@ -26,27 +26,23 @@ impl RhoConnection { }; // Notify OmegaConnection about the new Iota - OmegaConnection::connect_iota(rho_connection.get_iota_id().await, user_ids); + OmegaConnection::connect_iota(rho_connection.get_iota_id().await, user_ids).await; rho_connection } - /// Get the Iota ID pub async fn get_iota_id(&self) -> Uuid { self.iota_connection.get_iota_id().await } - /// Get the user IDs associated with this Rho connection pub fn get_user_ids(&self) -> &Vec { &self.user_ids } - /// Get reference to the IotaConnection pub fn get_iota_connection(&self) -> &Arc { &self.iota_connection } - /// Get all client connections pub async fn get_client_connections(&self) -> Vec> { let connections = self.client_connections.read().await; connections.clone() @@ -69,7 +65,6 @@ impl RhoConnection { /// Add a client connection pub async fn add_client_connection(&self, connection: Arc) { - // Notify Iota about new client let notification = CommunicationValue::new(CommunicationType::client_connected) .add_data_str( DataTypes::user_id, @@ -82,18 +77,17 @@ impl RhoConnection { self.iota_connection.send_message(notification).await; - // Add to our list { let mut connections = self.client_connections.write().await; connections.push(Arc::clone(&connection)); } - // Notify OmegaConnection OmegaConnection::client_changed( self.get_iota_id().await, connection.get_user_id().await.unwrap_or(Uuid::nil()), UserStatus::online, - ); + ) + .await; } /// Remove a client connection @@ -109,7 +103,6 @@ impl RhoConnection { }) }); - // Push the new connection connections.push(Arc::clone(&connection)); } @@ -118,7 +111,8 @@ impl RhoConnection { self.get_iota_id().await, connection.get_user_id().await.unwrap_or(Uuid::nil()), UserStatus::user_offline, - ); + ) + .await; } /// Close the Iota connection and all associated client connections @@ -133,7 +127,7 @@ impl RhoConnection { rho_manager::remove_rho(self.get_iota_id().await).await; // Notify OmegaConnection - OmegaConnection::close_iota(self.get_iota_id().await); + OmegaConnection::close_iota(self.get_iota_id().await).await; } /// Send message from Iota to specific client by user ID @@ -220,32 +214,3 @@ impl RhoConnection { connections.len() } } -#[cfg(test)] -mod tests { - use super::*; - use tokio::sync::Mutex; - - #[tokio::test] - async fn test_rho_connection_creation() { - let iota_id = Uuid::new_v4(); - let user_ids = vec![Uuid::new_v4(), Uuid::new_v4()]; - - // Mock session - // Create a mock WebSocket stream (this would fail in practice but shows the API) - // In real implementation, this would be a proper WebSocketStream - let mock_stream = - std::ptr::null_mut() as *mut tokio_tungstenite::WebSocketStream; - let mock_stream = unsafe { std::ptr::read(mock_stream) }; - let iota_conn = IotaConnection::new_with_ids( - iota_id, - user_ids.clone(), - Arc::new(tokio::sync::Mutex::new(mock_stream)), - ); - - let rho_conn = RhoConnection::new(iota_conn, user_ids.clone()).await; - - assert_eq!(rho_conn.get_iota_id().await, iota_id); - assert_eq!(rho_conn.get_user_ids(), &user_ids); - assert_eq!(rho_conn.client_count().await, 0); - } -} diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index c6cb0c4..c832273 100644 --- a/src/rho/rho_manager.rs +++ b/src/rho/rho_manager.rs @@ -7,14 +7,17 @@ use uuid::Uuid; use super::rho_connection::RhoConnection; -// Static storage for RhoConnections, keyed by Iota ID pub static RHO_CONNECTIONS: LazyLock>>>> = LazyLock::new(|| Arc::new(RwLock::new(HashMap::new()))); -/// Get a RhoConnection for a specific user ID pub async fn get_rho_con_for_user(user_id: Uuid) -> Option> { let connections = RHO_CONNECTIONS.read().await; + println!("Checking user ID: {:?}", user_id); for rho_connection in connections.values() { + println!( + "Comparing user IDs: {:?}", + rho_connection.get_user_ids().to_vec() + ); if rho_connection.get_user_ids().contains(&user_id) { return Some(Arc::clone(rho_connection)); } @@ -22,7 +25,6 @@ pub async fn get_rho_con_for_user(user_id: Uuid) -> Option> { None } -/// Check if an Iota ID exists in the connections pub async fn contains_iota(iota_id: Uuid) -> bool { let connections = RHO_CONNECTIONS.read().await; connections.contains_key(&iota_id) diff --git a/src/util/config_util.rs b/src/util/config_util.rs index e2753da..a13bdd6 100644 --- a/src/util/config_util.rs +++ b/src/util/config_util.rs @@ -1,13 +1,9 @@ use crate::util::file_util::load_file; -use json::JsonValue; +use futures::lock::Mutex; use once_cell::sync::Lazy; use serde::{Deserialize, Serialize}; -use std::sync::RwLock; use uuid::Uuid; -const CONFIG_PATH: &str = ""; -const CONFIG_FILENAME: &str = "config.json"; -// Define the Config structure #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Config { pub omega_server: String, @@ -19,13 +15,12 @@ pub struct Config { pub port: u16, } -// Provide default values (similar to Java static defaults) impl Default for Config { fn default() -> Self { Self { omega_server: "omega.tensamin.methanium.net".into(), auth_server: "auth.tensamin.methanium.net".into(), - omikron_id: Uuid::parse_str("a9e92dd6-08a6-4765-abf1-9fa39d0a99f9").unwrap(), + omikron_id: Uuid::parse_str("a9e92dd6-08a6-4765-abf1-9fa39d0a99f9").unwrap_or_default(), keep_people_stored_for: 90, max_data: 1000 * 1000 * 1000 * 8, ip: "0.0.0.0".into(), @@ -34,21 +29,22 @@ impl Default for Config { } } -// Global instance with thread-safe read/write access -pub static CONFIG: Lazy> = Lazy::new(|| { - let config = Config::load().unwrap_or_default(); - RwLock::new(config) -}); +pub static CONFIG: Lazy> = Lazy::new(|| Mutex::new(Config::load())); impl Config { - pub fn load() -> Option { - let content = load_file(CONFIG_PATH, CONFIG_FILENAME); + pub fn load() -> Self { + let content = load_file("", "config.json"); if content.trim().is_empty() { - return None; + return Config::default(); } - let json = json::parse(&content).ok()?; - Some(Self { + let json = json::parse(&content).unwrap(); + println!("{:?}", json); + println!( + "{:?}", + Uuid::parse_str(json["omikron_id"].as_str().unwrap_or_default()).unwrap_or_default() + ); + Self { omega_server: json["omega_server"].as_str().unwrap_or_default().into(), auth_server: json["auth_server"].as_str().unwrap_or_default().into(), omikron_id: Uuid::parse_str(json["omikron_id"].as_str().unwrap_or_default()) @@ -58,6 +54,6 @@ impl Config { max_data: json["max_data"].as_u64().unwrap_or_default(), ip: json["ip"].as_str().unwrap_or_default().into(), port: json["port"].as_u64().unwrap_or_default() as u16, - }) + } } } diff --git a/src/util/file_util.rs b/src/util/file_util.rs index 76e873f..774a5e3 100644 --- a/src/util/file_util.rs +++ b/src/util/file_util.rs @@ -1,10 +1,7 @@ use std::ffi::OsStr; -use std::fmt::Write as FmtWrite; use std::fs::{self, File}; -use std::io::{Read, Write}; +use std::io::Read; use std::path::{Path, PathBuf}; -use std::process; -use std::time::SystemTime; use sysinfo::System; use uuid::Uuid; use walkdir::WalkDir;