From 494e08241ce797c891c1ce1258067ec1bea69a42 Mon Sep 17 00:00:00 2001 From: Alex-Emmet Date: Tue, 20 Jan 2026 00:05:14 +0100 Subject: [PATCH] add a conversation by name, more stable client connection --- Cargo.lock | 57 ++++------ Cargo.toml | 36 ++++++- src/data/communication.rs | 201 +++++------------------------------ src/rho/client_connection.rs | 139 +++++++++++++++++------- src/rho/iota_connection.rs | 24 ++--- src/rho/rho_connection.rs | 26 ++--- 6 files changed, 203 insertions(+), 280 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a1d40d7..e77d12a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -15,7 +15,6 @@ dependencies = [ "block-modes", "bytes", "cbc", - "chacha20poly1305", "chrono", "cmake", "crossterm", @@ -41,6 +40,8 @@ dependencies = [ "rustls 0.23.36", "serde", "sha2", + "strum", + "strum_macros", "sys-info", "sysinfo", "tokio", @@ -552,30 +553,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" -[[package]] -name = "chacha20" -version = "0.9.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818" -dependencies = [ - "cfg-if", - "cipher", - "cpufeatures", -] - -[[package]] -name = "chacha20poly1305" -version = "0.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35" -dependencies = [ - "aead", - "chacha20", - "cipher", - "poly1305", - "zeroize", -] - [[package]] name = "chrono" version = "0.4.42" @@ -597,7 +574,6 @@ checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ "crypto-common", "inout", - "zeroize", ] [[package]] @@ -2427,17 +2403,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "poly1305" -version = "0.8.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8159bd90725d2df49889a078b54f4f79e87f1f8a8444194cdca81d38f5393abf" -dependencies = [ - "cpufeatures", - "opaque-debug", - "universal-hash", -] - [[package]] name = "polyval" version = "0.6.2" @@ -3279,6 +3244,24 @@ version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" +[[package]] +name = "strum" +version = "0.27.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf" + +[[package]] +name = "strum_macros" +version = "0.27.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7695ce3845ea4b33927c055a39dc438a45b059f7c1b3d91d38d10355fb8cbca7" +dependencies = [ + "heck 0.5.0", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "subtle" version = "2.6.1" diff --git a/Cargo.toml b/Cargo.toml index 5652dd2..6158341 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,38 @@ edition = "2024" [dependencies] ansi_term = "*" -async-tungstenite = { version = "0.32.0", features = ["futures-03-sink", "futures-util", "handshake", "__rustls-tls", "async-native-tls", "async-std", "async-std-runtime", "async-tls", "gio", "gio-runtime", "glib", "openssl", "real-async-native-tls", "real-async-tls", "real-native-tls", "real-tokio-native-tls", "real-tokio-openssl", "real-tokio-rustls", "rustls-native-certs", "rustls-pki-types", "tokio", "tokio-native-tls", "tokio-openssl", "tokio-runtime", "tokio-rustls-manual-roots", "tokio-rustls-native-certs", "tokio-rustls-webpki-roots", "url", "verbose-logging", "webpki-roots"] } +async-tungstenite = { version = "0.32.0", features = [ + "futures-03-sink", + "futures-util", + "handshake", + "__rustls-tls", + "async-native-tls", + "async-std", + "async-std-runtime", + "async-tls", + "gio", + "gio-runtime", + "glib", + "openssl", + "real-async-native-tls", + "real-async-tls", + "real-native-tls", + "real-tokio-native-tls", + "real-tokio-openssl", + "real-tokio-rustls", + "rustls-native-certs", + "rustls-pki-types", + "tokio", + "tokio-native-tls", + "tokio-openssl", + "tokio-runtime", + "tokio-rustls-manual-roots", + "tokio-rustls-native-certs", + "tokio-rustls-webpki-roots", + "url", + "verbose-logging", + "webpki-roots", +] } axum = "*" base64 = "0.22.1" bytes = "*" @@ -52,7 +83,8 @@ dotenv = "0.15.0" aes-gcm = "0.10.3" tokio-native-tls = "0.3.1" hkdf = "0.12.4" -chacha20poly1305 = "0.10.1" block-modes = "0.9.1" cbc = "0.1.2" aes = "0.8.4" +strum = "0.27.2" +strum_macros = "0.27.2" diff --git a/src/data/communication.rs b/src/data/communication.rs index b2b5898..fd57fe1 100644 --- a/src/data/communication.rs +++ b/src/data/communication.rs @@ -2,9 +2,11 @@ use json::number::Number; use json::{Array, JsonValue, object, parse}; use std::collections::HashMap; use std::time::{SystemTime, UNIX_EPOCH}; +use strum::IntoEnumIterator; +use strum_macros::EnumIter; use uuid::Uuid; -#[derive(Eq, Hash, PartialEq, Clone, Debug)] +#[derive(Eq, Hash, PartialEq, EnumIter, Clone, Debug)] #[allow(non_camel_case_types, dead_code)] pub enum DataTypes { error_type, @@ -14,6 +16,7 @@ pub enum DataTypes { settings, settings_name, chat_partner_id, + chat_partner_name, iota_id, user_id, user_ids, @@ -88,91 +91,21 @@ pub enum DataTypes { impl DataTypes { pub fn parse(p0: String) -> DataTypes { - let normalized = p0.to_lowercase().replace('_', ""); - - match normalized.as_str() { - "errortype" => DataTypes::error_type, - "chatpartnerid" => DataTypes::chat_partner_id, - "registerid" => DataTypes::register_id, - "uuid" => DataTypes::uuid, - "settings" => DataTypes::settings, - "settingsname" => DataTypes::settings_name, - "iotaid" => DataTypes::iota_id, - "userid" => DataTypes::user_id, - "userids" => DataTypes::user_ids, - "iotaids" => DataTypes::iota_ids, - "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, - "acceptedprofiles" => DataTypes::accepted_profiles, - "deniedprofiles" => DataTypes::denied_profiles, - "content" => DataTypes::content, - "messages" => DataTypes::messages, - "sendtime" => DataTypes::send_time, - "gettime" => DataTypes::get_time, - "getvariant" => DataTypes::get_variant, - "sharedsecretown" => DataTypes::shared_secret_own, - "sharedsecretother" => DataTypes::shared_secret_other, - "sharedsecretsign" => DataTypes::shared_secret_sign, - "sharedsecret" => DataTypes::shared_secret, - "callid" => DataTypes::call_id, - "calltoken" => DataTypes::call_token, - "untill" => DataTypes::untill, - "enable" => DataTypes::enable, - "startdate" => DataTypes::start_date, - "enddate" => DataTypes::end_date, - "receiverid" => DataTypes::receiver_id, - "senderid" => DataTypes::sender_id, - "signature" => DataTypes::signature, - "signed" => DataTypes::signed, - "message" => DataTypes::message, - "lastping" => DataTypes::last_ping, - "pingiota" => DataTypes::ping_iota, - "pingclients" => DataTypes::ping_clients, - "matches" => DataTypes::matches, - "omikron" => DataTypes::omikron, - "offset" => DataTypes::offset, - "amount" => DataTypes::amount, - "position" => DataTypes::position, - "name" => DataTypes::name, - "path" => DataTypes::path, - "codec" => DataTypes::codec, - "function" => DataTypes::function, - "payload" => DataTypes::payload, - "result" => DataTypes::result, - "interactables" => DataTypes::interactables, - "wanttowatch" => DataTypes::want_to_watch, - "watcher" => DataTypes::watcher, - "createdat" => DataTypes::created_at, - "username" => DataTypes::username, - "display" => DataTypes::display, - "avatar" => DataTypes::avatar, - "about" => DataTypes::about, - "status" => DataTypes::status, - "publickey" => DataTypes::public_key, - "sublevel" => DataTypes::sub_level, - "subend" => DataTypes::sub_end, - "communityaddress" => DataTypes::community_address, - "challenge" => DataTypes::challenge, - "communitytitle" => DataTypes::community_title, - "communities" => DataTypes::communities, - "rhoconnections" => DataTypes::rho_connections, - "user" => DataTypes::user, - "onlinestatus" => DataTypes::online_status, - "omikronid" => DataTypes::omikron_id, - "omikronconnections" => DataTypes::omikron_connections, - "resettoken" => DataTypes::reset_token, - "newtoken" => DataTypes::new_token, - _ => DataTypes::error_type, // fallback if unknown + for datatype in DataTypes::iter() { + if datatype.to_string().to_lowercase().replace('_', "") + == p0.to_lowercase().replace('_', "") + { + return datatype; + } } + DataTypes::error_type + } + pub fn to_string(&self) -> String { + return format!("{:?}", self); } } -#[derive(PartialEq, Clone, Debug)] +#[derive(PartialEq, Clone, EnumIter, Debug)] #[allow(non_camel_case_types, dead_code)] pub enum CommunicationType { error, @@ -218,7 +151,7 @@ pub enum CommunicationType { register_iota_success, ping, pong, - add_chat, + add_conversation, send_chat, client_changed, client_connected, @@ -266,99 +199,17 @@ pub enum CommunicationType { } impl CommunicationType { pub fn parse(p0: String) -> CommunicationType { - let normalized = p0.to_lowercase().replace('_', ""); - - match normalized.as_str() { - "watchstream" => CommunicationType::watch_stream, - "calltoken" => CommunicationType::call_token, - "callinvite" => CommunicationType::call_invite, - "calldisconnectuser" => CommunicationType::call_disconnect_user, - "calltimeoutuser" => CommunicationType::call_timeout_user, - "callsetanonymousjoining" => CommunicationType::call_set_anonymous_joining, - "endcall" => CommunicationType::end_call, - "function" => CommunicationType::function, - "update" => CommunicationType::update, - "createuser" => CommunicationType::create_user, - "errorinternal" => CommunicationType::error_internal, - "errorinvaliddata" => CommunicationType::error_invalid_data, - "errorinvaliduserid" => CommunicationType::error_invalid_user_id, - "errorinvalidomikronid" => CommunicationType::error_invalid_omikron_id, - "errornotfound" => CommunicationType::error_not_found, - "errornotauthenticated" => CommunicationType::error_not_authenticated, - "errornoiota" => CommunicationType::error_no_iota, - "errorinvalidchallenge" => CommunicationType::error_invalid_challenge, - "errorinvalidpublickey" => CommunicationType::error_invalid_public_key, - "errorinvalidsecret" => CommunicationType::error_invalid_secret, - "errorinvalidprivatekey" => CommunicationType::error_invalid_private_key, - "errornouserid" => CommunicationType::error_no_user_id, - "errornocallid" => CommunicationType::error_no_call_id, - "errorinvalidcallid" => CommunicationType::error_invalid_call_id, - "success" => CommunicationType::success, - "settingssave" => CommunicationType::settings_save, - "settingsload" => CommunicationType::settings_load, - "settingslist" => CommunicationType::settings_list, - "message" => CommunicationType::message, - "messagesend" => CommunicationType::message_send, - "messagelive" => CommunicationType::message_live, - "messageotheriota" => CommunicationType::message_other_iota, - "messagechunk" => CommunicationType::message_chunk, - "messagesget" => CommunicationType::messages_get, - "changeconfirm" => CommunicationType::change_confirm, - "confirmreceive" => CommunicationType::confirm_receive, - "confirmread" => CommunicationType::confirm_read, - "getchats" => CommunicationType::get_chats, - "getstates" => CommunicationType::get_states, - "addcommunity" => CommunicationType::add_community, - "removecommunity" => CommunicationType::remove_community, - "getcommunities" => CommunicationType::get_communities, - "challenge" => CommunicationType::challenge, - "challengeresponse" => CommunicationType::challenge_response, - "register" => CommunicationType::register, - "registerresponse" => CommunicationType::register_response, - "identification" => CommunicationType::identification, - "identificationresponse" => CommunicationType::identification_response, - "registeriota" => CommunicationType::register_iota, - "registeriotasuccess" => CommunicationType::register_iota_success, - "ping" => CommunicationType::ping, - "pong" => CommunicationType::pong, - "addchat" => CommunicationType::add_chat, - "sendchat" => CommunicationType::send_chat, - "clientchanged" => CommunicationType::client_changed, - "clientconnected" => CommunicationType::client_connected, - "clientdisconnected" => CommunicationType::client_disconnected, - "clientclosed" => CommunicationType::client_closed, - "publickey" => CommunicationType::public_key, - "privatekey" => CommunicationType::private_key, - "webrtcsdp" => CommunicationType::webrtc_sdp, - "webrtcice" => CommunicationType::webrtc_ice, - "startstream" => CommunicationType::start_stream, - "endstream" => CommunicationType::end_stream, - "rhoupdate" => CommunicationType::rho_update, - - "iotaconnected" => CommunicationType::iota_connected, - "iotadisconnected" => CommunicationType::iota_disconnected, - "userconnected" => CommunicationType::user_connected, - "userdisconnected" => CommunicationType::user_disconnected, - "syncclientiotastatus" => CommunicationType::sync_client_iota_status, - - "getuserdata" => CommunicationType::get_user_data, - "getiotadata" => CommunicationType::get_iota_data, - "iotauserdata" => CommunicationType::iota_user_data, - - "changeuserdata" => CommunicationType::change_user_data, - "changeiotadata" => CommunicationType::change_iota_data, - - "getregister" => CommunicationType::get_register, - "completeregisteruser" => CommunicationType::complete_register_user, - "completeregisteriota" => CommunicationType::complete_register_iota, - "deleteuser" => CommunicationType::delete_user, - "deleteiota" => CommunicationType::delete_iota, - - "startregister" => CommunicationType::start_register, - "completeregister" => CommunicationType::complete_register, - - _ => CommunicationType::error, + for datatype in CommunicationType::iter() { + if datatype.to_string().to_lowercase().replace('_', "") + == p0.to_lowercase().replace('_', "") + { + return datatype; + } } + CommunicationType::error + } + pub fn to_string(&self) -> String { + return format!("{:?}", self); } } diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 71109ec..a72247c 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -4,7 +4,7 @@ use json::JsonValue; use json::number::Number; use rand::Rng; use rand::distributions::Alphanumeric; -use std::sync::{Arc, Weak}; +use std::sync::Arc; use std::time::Duration; use tokio::sync::RwLock; use tokio_util::compat::Compat; @@ -38,6 +38,7 @@ pub struct ClientConnection { pub_key: Arc>>>, pub rho_connection: Arc>>>, pub interested_users: Arc>>, + is_open: Arc>, } impl ClientConnection { @@ -57,6 +58,7 @@ impl ClientConnection { pub_key: Arc::new(RwLock::new(None)), rho_connection: Arc::new(RwLock::new(None)), interested_users: Arc::new(RwLock::new(Vec::new())), + is_open: Arc::new(RwLock::new(true)), }) } @@ -81,7 +83,7 @@ impl ClientConnection { } /// Send a string message to the client - pub async fn send_message_str(&self, message: &str) { + pub async fn send_message_str(self: Arc, message: &str) { let mut session = self.sender.write().await; if let Err(e) = session .send(Message::Text(Utf8Bytes::from(message.to_string()))) @@ -92,7 +94,14 @@ impl ClientConnection { } /// Send a CommunicationValue to the client - pub async fn send_message(&self, cv: &CommunicationValue) { + pub async fn send_message(self: Arc, cv: &CommunicationValue) { + if !*self.is_open.read().await { + log_out!( + PrintType::Client, + "Attempted to send message to a closed connection." + ); + return; + } if !cv.is_type(CommunicationType::pong) { log_out!(PrintType::Client, "{}", &cv.to_json().to_string()); } @@ -119,7 +128,8 @@ impl ClientConnection { .unwrap_or(0); if user_id == 0 { log_out!(PrintType::Client, "Invalid USER ID"); - self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_data) + self.clone() + .send_error_response(&cv.get_id(), CommunicationType::error_invalid_data) .await; self.close().await; return; @@ -137,7 +147,8 @@ impl ClientConnection { if let Ok(response_cv) = response_cv { if !response_cv.is_type(CommunicationType::get_user_data) { - self.send_error_response(&cv.get_id(), CommunicationType::error_internal) + self.clone() + .send_error_response(&cv.get_id(), CommunicationType::error_internal) .await; self.close().await; return; @@ -151,11 +162,12 @@ impl ClientConnection { let pub_key = match load_public_key(base64_pub) { Some(pk) => pk, None => { - self.send_error_response( - &cv.get_id(), - CommunicationType::error_invalid_public_key, - ) - .await; + self.clone() + .send_error_response( + &cv.get_id(), + CommunicationType::error_invalid_public_key, + ) + .await; self.close().await; return; } @@ -190,7 +202,8 @@ impl ClientConnection { self.send_message(&challenge_msg).await; } else { - self.send_error_response(&cv.get_id(), CommunicationType::error_internal) + self.clone() + .send_error_response(&cv.get_id(), CommunicationType::error_internal) .await; self.close().await; return; @@ -221,6 +234,7 @@ impl ClientConnection { return; } }; + rho_connection.add_client_connection(self.clone()).await; // Set identification data { @@ -238,11 +252,12 @@ impl ClientConnection { .with_id(cv.get_id()); self.send_message(&response).await; } else { - self.send_error_response( - &cv.get_id(), - CommunicationType::error_not_authenticated, - ) - .await; + self.clone() + .send_error_response( + &cv.get_id(), + CommunicationType::error_not_authenticated, + ) + .await; self.close().await; return; } @@ -250,7 +265,8 @@ impl ClientConnection { } if !self.is_identified().await { - self.send_error_response(&cv.get_id(), CommunicationType::error_not_authenticated) + self.clone() + .send_error_response(&cv.get_id(), CommunicationType::error_not_authenticated) .await; self.close().await; return; @@ -288,12 +304,11 @@ impl ClientConnection { self.handle_omega_forward(cv).await; return; } - // Forward other messages to Iota self.forward_to_iota(cv).await; }); } - async fn handle_omega_forward(&self, cv: CommunicationValue) { + async fn handle_omega_forward(self: Arc, cv: CommunicationValue) { let client_for_closure = self.clone(); WAITING_TASKS.insert( cv.get_id(), @@ -311,7 +326,7 @@ impl ClientConnection { } /// Handle ping message - async fn handle_ping(&self, cv: CommunicationValue) { + async fn handle_ping(self: Arc, 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::() { @@ -336,7 +351,7 @@ impl ClientConnection { } /// Handle client status change - async fn handle_client_changed(&self, cv: CommunicationValue) { + async fn handle_client_changed(self: Arc, cv: CommunicationValue) { let user_id = self.get_user_id().await; if let Some(_status_str) = cv.get_data(DataTypes::user_state) { let user_status = UserStatus::online; @@ -348,7 +363,7 @@ impl ClientConnection { } /// Handle call invite - async fn handle_call_invite(&self, cv: CommunicationValue) { + async fn handle_call_invite(self: Arc, cv: CommunicationValue) { let receiver_id: i64 = cv .get_data(DataTypes::receiver_id) .unwrap_or(&json::JsonValue::Number(Number::from(0))) @@ -415,7 +430,7 @@ impl ClientConnection { } /// Handle get call request - async fn handle_get_call(&self, cv: CommunicationValue) { + async fn handle_get_call(self: Arc, cv: CommunicationValue) { let user_id = self.get_user_id().await; let call_id = match cv.get_data(DataTypes::call_id) { @@ -446,22 +461,59 @@ impl ClientConnection { return; } } - async fn handle_call_timeout_user(&self, cv: CommunicationValue) { + async fn handle_call_timeout_user(self: Arc, cv: CommunicationValue) { let user_id = cv.get_data(DataTypes::call_id).unwrap(); let call_id = cv.get_data(DataTypes::user_id).unwrap(); // JA man braucht CALL_ID } - async fn handle_call_disconnect_user(&self, cv: CommunicationValue) { + async fn handle_call_disconnect_user(self: Arc, cv: CommunicationValue) { let user_id = cv.get_data(DataTypes::call_id).unwrap(); let call_id = cv.get_data(DataTypes::user_id).unwrap(); // JA man braucht CALL_ID let untill = cv.get_data(DataTypes::untill).unwrap(); } - async fn handle_call_set_anonymous_joining(&self, cv: CommunicationValue) { + async fn handle_call_set_anonymous_joining(self: Arc, cv: CommunicationValue) { let call_id = cv.get_data(DataTypes::user_id).unwrap(); let enable = cv.get_data(DataTypes::enable).unwrap(); } /// Forward message to Iota - async fn forward_to_iota(&self, cv: CommunicationValue) { + async fn forward_to_iota(self: Arc, cv: CommunicationValue) { + if cv.is_type(CommunicationType::add_conversation) + && cv.get_data(DataTypes::chat_partner_id).is_none() + { + let chat_partner_name = cv + .get_data(DataTypes::chat_partner_name) + .unwrap_or(&JsonValue::Null) + .as_str() + .unwrap_or(""); + + let load_uuid_response = get_omega_connection() + .await_response( + &CommunicationValue::new(CommunicationType::get_user_data) + .with_id(cv.get_id()) + .add_data(DataTypes::username, JsonValue::from(chat_partner_name)), + Some(Duration::from_secs(20)), + ) + .await; + let chat_partner_id = { + if let Ok(load_uuid_response) = load_uuid_response { + load_uuid_response + .get_data(DataTypes::user_id) + .unwrap_or(&JsonValue::Null) + .clone() + } else { + JsonValue::Null + } + }; + + if let Some(rho_conn) = self.get_rho_connection().await { + let updated_cv = cv + .with_sender(self.get_user_id().await) + .add_data(DataTypes::chat_partner_id, chat_partner_id); + rho_conn.message_to_iota(updated_cv).await; + } + return; + } + if let Some(rho_conn) = self.get_rho_connection().await { let updated_cv = cv.with_sender(self.get_user_id().await); rho_conn.message_to_iota(updated_cv).await; @@ -469,26 +521,40 @@ impl ClientConnection { } /// Send error response - async fn send_error_response(&self, message_id: &Uuid, error_type: CommunicationType) { + async fn send_error_response( + self: Arc, + message_id: &Uuid, + error_type: CommunicationType, + ) { let error = CommunicationValue::new(error_type).with_id(*message_id); self.send_message(&error).await; } /// Close the connection pub async fn close(&self) { + let mut is_open_guard = self.is_open.write().await; + if *is_open_guard { + return; + } + *is_open_guard = false; + let mut session = self.sender.write().await; let _ = session.close(None).await; } /// Set interested users list - pub async fn set_interested_users(&self, interested_ids: Vec) { + pub async fn set_interested_users(self: Arc, interested_ids: Vec) { let mut interested_guard = self.interested_users.write().await; *interested_guard = interested_ids; } + pub async fn get_interested_users(self: Arc) -> Vec { + let interested_guard = self.interested_users.read().await; + interested_guard.clone() + } /// Check if interested in a user and send notification - pub async fn are_you_interested(&self, user: &User) { - let interested_guard = self.interested_users.read().await; + pub async fn are_you_interested(self: Arc, user: &User) { + let interested_guard = self.clone().get_interested_users().await; if interested_guard.contains(&user.user_id) { let notification = CommunicationValue::new(CommunicationType::client_changed) .add_data_str(DataTypes::user_id, user.user_id.to_string()) @@ -528,16 +594,7 @@ impl Clone for ClientConnection { pub_key: Arc::clone(&self.pub_key), rho_connection: Arc::clone(&self.rho_connection), interested_users: Arc::clone(&self.interested_users), + is_open: Arc::clone(&self.is_open), } } } - -impl std::fmt::Debug for ClientConnection { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("ClientConnection") - .field("user_id", &"[async]") - .field("identified", &"[async]") - .field("ping", &"[async]") - .finish() - } -} diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 83addb9..d42a426 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -2,12 +2,12 @@ use crate::calls::call_group::CallGroup; use crate::calls::call_manager; use crate::get_private_key; use crate::get_public_key; +use crate::log; use crate::log_err; use crate::log_in; use crate::log_out; use crate::omega::omega_connection::WAITING_TASKS; use crate::omega::omega_connection::get_omega_connection; -use crate::util::crypto_helper::encrypt; use crate::util::crypto_helper::load_public_key; use crate::util::crypto_helper::public_key_to_base64; use crate::util::crypto_util::DataFormat; @@ -30,7 +30,6 @@ use tokio::sync::mpsc; use tokio_util::compat::Compat; use tungstenite::Utf8Bytes; use uuid::Uuid; -use warp::filters::method::get; use x448::PublicKey; use super::{rho_connection::RhoConnection, rho_manager}; @@ -395,11 +394,15 @@ impl IotaConnection { self.close().await; return; } + // Handle GET_CHATS + if cv.is_type(CommunicationType::get_chats) { + self.handle_get_chats(cv).await; + return; + } - log_in!(PrintType::Iota, "{}", &cv.to_json().to_string()); // Handle forwarding to other Iotas or clients let receiver_id = cv.get_receiver(); - if !self.get_user_ids().await.contains(&receiver_id) + if (receiver_id != 0 && !self.get_user_ids().await.contains(&receiver_id)) || cv.is_type(CommunicationType::message_other_iota) || cv.is_type(CommunicationType::send_chat) { @@ -407,12 +410,6 @@ impl IotaConnection { return; } - // Handle GET_CHATS - if cv.is_type(CommunicationType::get_chats) { - self.handle_get_chats(cv).await; - return; - } - if cv.is_type(CommunicationType::change_iota_data) || cv.is_type(CommunicationType::get_user_data) || cv.is_type(CommunicationType::get_iota_data) @@ -510,6 +507,7 @@ impl IotaConnection { let receiver_id = cv.get_receiver(); let mut interested_ids: Vec = Vec::new(); + // loading Calls let calls: Vec> = call_manager::get_call_groups(receiver_id).await; let mut invites: HashMap> = HashMap::new(); let empty = &calls.is_empty(); @@ -543,8 +541,7 @@ impl IotaConnection { for user_json in user_ids { let user_id = user_json["user_id"].as_i64().unwrap_or(0); interested_ids.push(user_id); - let mut enriched_contact = JsonValue::new_object(); - let _ = enriched_contact.insert("user_id", user_id); + let mut enriched_contact = user_json.clone(); if let Some(calls) = invites.get(&user_id) { let _ = enriched_contact.insert("calls", JsonValue::Array(calls.clone())); @@ -575,8 +572,9 @@ 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(); rho_conn.message_to_client(updated_cv).await; + } else { + log_err!(PrintType::General, "Failed to forward message to client"); } } diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 704f4a6..50d6cfe 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -1,9 +1,13 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection, rho_manager}; -use crate::data::{ - communication::{CommunicationType, CommunicationValue, DataTypes}, - user::UserStatus, -}; use crate::omega::omega_connection::OmegaConnection; +use crate::util::logger::PrintType; +use crate::{ + data::{ + communication::{CommunicationType, CommunicationValue, DataTypes}, + user::UserStatus, + }, + log, +}; use json::{JsonValue, number::Number}; use std::collections::HashMap; use std::sync::Arc; @@ -83,16 +87,12 @@ impl RhoConnection { /// Remove a client connection pub async fn close_client_connection(&self, connection: Arc) { + let target_user_id = connection.get_user_id().await; { let mut connections = self.client_connections.write().await; - - let target_user_id = connection.get_user_id().await; - connections.retain(|con| { futures::executor::block_on(async { con.get_user_id().await != target_user_id }) }); - - connections.push(Arc::clone(&connection)); } // Notify OmegaConnection @@ -122,9 +122,10 @@ impl RhoConnection { /// Send message from Iota to specific client pub async fn message_to_client(&self, cv: CommunicationValue) { let connections = self.client_connections.read().await; + let receiver_id = cv.get_receiver(); for connection in connections.iter() { - if connection.get_user_id().await == cv.get_receiver() { - connection.send_message(&cv).await; + if connection.get_user_id().await == receiver_id { + connection.clone().send_message(&cv).await; } } } @@ -141,6 +142,7 @@ impl RhoConnection { let conn_user_id = connection.get_user_id().await; if conn_user_id == user_id { connection + .clone() .set_interested_users(interested_ids.clone()) .await; break; @@ -152,7 +154,7 @@ impl RhoConnection { pub async fn are_they_interested(&self, user: &crate::data::user::User) { let connections = self.client_connections.read().await; for connection in connections.iter() { - connection.are_you_interested(user).await; + connection.clone().are_you_interested(user).await; } }