From 5728280ac98ed69e40b04185aa0bab7c227fd053 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Wed, 3 Jun 2026 20:55:22 +0200 Subject: [PATCH] Message States, Tauri, Global and Local Settings --- Cargo.lock | 76 +++---- Cargo.toml | 4 +- .../anonymous_client_connection.rs | 49 ++++- src/data/user.rs | 1 + src/omega/omega_connection.rs | 4 +- src/rho/client_connection.rs | 192 ++++++++++++++---- src/rho/rho_connection.rs | 93 ++++++--- 7 files changed, 301 insertions(+), 118 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 660d6b8..c3f935a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -333,9 +333,9 @@ dependencies = [ [[package]] name = "bitflags" -version = "2.11.1" +version = "2.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" +checksum = "84d7ced0ae9557296835c32bf1b1e02b44c746701f898460fb000d7eaa84f00a" [[package]] name = "block-buffer" @@ -397,9 +397,9 @@ checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" [[package]] name = "cc" -version = "1.2.62" +version = "1.2.63" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1dce859f0832a7d088c4f1119888ab94ef4b5d6795d1ce05afb7fe159d79f98" +checksum = "556e016178bb5662a08681bbe0f00f8e17631781a4dfc8c45e466e4b185ec27f" dependencies = [ "find-msvc-tools", "jobserver", @@ -455,9 +455,9 @@ dependencies = [ [[package]] name = "cmov" -version = "0.5.3" +version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f88a43d011fc4a6876cb7344703e297c71dda42494fee094d5f7c76bf13f746" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" [[package]] name = "combine" @@ -736,9 +736,9 @@ dependencies = [ [[package]] name = "displaydoc" -version = "0.2.5" +version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" +checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" dependencies = [ "proc-macro2", "quote", @@ -1284,9 +1284,9 @@ checksum = "1a9fcbcc408c5526c3ab80d534e5c86e7967c1fb7aa0a8c76abd1edc27deb877" [[package]] name = "http" -version = "1.4.0" +version = "1.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a" +checksum = "8be7462df143984c4598a256ef469b251d7d7f9e271135073e78fc535414f3d0" dependencies = [ "bytes", "itoa", @@ -1332,9 +1332,9 @@ dependencies = [ [[package]] name = "hyper" -version = "1.9.0" +version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6299f016b246a94207e63da54dbe807655bf9e00044f73ded42c3ac5305fbcca" +checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" dependencies = [ "atomic-waker", "bytes", @@ -1706,9 +1706,9 @@ checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" [[package]] name = "livekit-api" -version = "0.4.24" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e2611299f65786fa733f38ef9aab9d796679e6f2f77572b84170949769ca9a25" +checksum = "c53d6335323ed9e4fd0c3ede761478fc607d147c6e3f2c29dc71ca73bcf18723" dependencies = [ "async-tungstenite", "base64 0.21.7", @@ -1734,9 +1734,9 @@ dependencies = [ [[package]] name = "livekit-protocol" -version = "0.7.7" +version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dd59f3759f1a14e60b6bc6d4d415414e1aed686c8e4d2c7862542d95301cdc70" +checksum = "565d83442aa99684dba5d0841fa342eb572fbd916ba2aa5e92801ecf76620026" dependencies = [ "pbjson", "pbjson-types", @@ -1755,9 +1755,9 @@ dependencies = [ [[package]] name = "log" -version = "0.4.29" +version = "0.4.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +checksum = "113b30b4cd05f7c06868fdb2854f66a7b9fece9a48425351cd532e810d74024f" dependencies = [ "value-bag", ] @@ -1770,9 +1770,9 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[package]] name = "memchr" -version = "2.8.0" +version = "2.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +checksum = "6b947ae49db0d222b1dbc6b113ce7248a3fc3a6ca21b696717bfc000ba4484d8" [[package]] name = "minimal-lexical" @@ -1782,9 +1782,9 @@ checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" [[package]] name = "mio" -version = "1.2.0" +version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" +checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" dependencies = [ "libc", "wasi", @@ -2844,9 +2844,9 @@ dependencies = [ [[package]] name = "rustls-native-certs" -version = "0.8.3" +version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63" +checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" dependencies = [ "openssl-probe", "rustls-pki-types", @@ -3045,9 +3045,9 @@ dependencies = [ [[package]] name = "shlex" -version = "1.3.0" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" [[package]] name = "signal-hook-registry" @@ -3083,9 +3083,9 @@ checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" [[package]] name = "socket2" -version = "0.6.3" +version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" +checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", "windows-sys 0.61.2", @@ -3420,7 +3420,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "ttp-core" version = "0.1.0" -source = "git+https://git.methanium.net/Tensamin/TTP.git#886edbf51d26636e46745c3637b80610df82f5dc" +source = "git+https://git.methanium.net/Tensamin/TTP.git#23438fa8f884e6ad0d32ca1004c0dedcce0cc8d2" dependencies = [ "base64 0.22.1", "byteorder", @@ -3433,7 +3433,7 @@ dependencies = [ [[package]] name = "ttp-native" version = "0.1.0" -source = "git+https://git.methanium.net/Tensamin/TTP.git#886edbf51d26636e46745c3637b80610df82f5dc" +source = "git+https://git.methanium.net/Tensamin/TTP.git#23438fa8f884e6ad0d32ca1004c0dedcce0cc8d2" dependencies = [ "quinn", "rustls", @@ -3483,9 +3483,9 @@ dependencies = [ [[package]] name = "typenum" -version = "1.20.0" +version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "40ce102ab67701b8526c123c1bab5cbe42d7040ccfd0f64af1a385808d2f43de" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" [[package]] name = "unicode-ident" @@ -3547,9 +3547,9 @@ checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" [[package]] name = "uuid" -version = "1.23.1" +version = "1.23.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ddd74a9687298c6858e9b88ec8935ec45d22e8fd5e6394fa1bd4e99a87789c76" +checksum = "d258b83ceec21034727ecee8c382cfa6c3e133699b0742c64571814fb420c9f7" dependencies = [ "getrandom 0.4.2", "js-sys", @@ -4229,18 +4229,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.48" +version = "0.8.50" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eed437bf9d6692032087e337407a86f04cd8d6a16a37199ed57949d415bd68e9" +checksum = "3b065d4f0e55f82fae73202e189638116a87c55ab6b8e6c2721e13dd9d854ad1" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.48" +version = "0.8.50" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4" +checksum = "0b631b19d36a892ab55420c92dbc83ccd79274f25be714855d3074aa71cab639" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index bc0734c..d31b956 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,8 +31,8 @@ dotenv = "0.15.0" hkdf = "0.13" strum = "0.28.0" strum_macros = "0.28.0" -livekit-api = { version = "0.4.19", features = ["native-tls"] } -livekit-protocol = "0.7.5" +livekit-api = { version = "0.5.0", features = ["native-tls"] } +livekit-protocol = "0.7.8" thiserror = "2.0.18" hickory-resolver = "0.25.2" serde = "1.0.228" diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index 22e25bd..1378af9 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -8,7 +8,8 @@ use uuid::Uuid; use crate::anonymous_clients::anonymous_manager::{self, generate_username}; use crate::calls::call_manager; -use crate::omega::omega_connection::get_omega_connection; +use crate::data::user::UserStatus; +use crate::omega::omega_connection::{OmegaConnection, get_omega_connection}; use crate::rho::connection::GeneralConnection; use crate::rho::rho_manager; use crate::util::logger::PrintType; @@ -309,11 +310,12 @@ impl AnonymousClientConnection { } /// Handle client status change - 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; - }*/ + async fn handle_client_changed(self: Arc, cv: CommunicationValue) { + if let DataValue::Str(status_str) = cv.get_data(DataTypes::user_state) { + let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online); + OmegaConnection::client_changed(self.user_id as i64, self.user_id as i64, user_status) + .await; + } } /// Handle call invite @@ -360,6 +362,25 @@ impl AnonymousClientConnection { let target_rho = match rho_manager::get_rho_con_for_user(receiver_id).await { Some(rho) => rho, _ => { + // Get sender user ID + let sender_id = self.get_user_id(); + + // User is offline - send push notification for call invite + let push_cv = CommunicationValue::new(CommunicationType::push_notification) + .with_receiver(receiver_id as u64) + .add_data(DataTypes::sender_id, DataValue::Number(sender_id as i64)) + .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string())) + .add_data( + DataTypes::notifications, + DataValue::Str("call_invite".to_string()), + ); + + let omega_conn = get_omega_connection(); + // Send fire-and-forget, don't await to avoid blocking + tokio::spawn(async move { + let _ = omega_conn.send_message(&push_cv).await; + }); + let error_cv = CommunicationValue::new(CommunicationType::error_not_found) .with_id(cv.get_id()) .add_data(DataTypes::receiver_id, DataValue::Number(receiver_id)); @@ -396,8 +417,11 @@ impl AnonymousClientConnection { DataValue::Str(id_str) => match Uuid::parse_str(&id_str.to_string()) { Ok(id) => id, Err(_) => { - self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_call_id) - .await; + self.send_error_response( + &cv.get_id(), + CommunicationType::error_invalid_call_id, + ) + .await; return; } }, @@ -493,12 +517,17 @@ impl AnonymousClientConnection { #[allow(dead_code)] /// Check if interested in a user and send notification - pub async fn are_you_interested(self: Arc, user_id: i64) { + pub async fn are_you_interested(self: Arc, user_id: i64, user_status: &str) { let interested_guard = self.clone().get_interested_users().await; if interested_guard.contains(&user_id) { + let status = if user_status == "user_invisible" { + "user_offline" + } else { + user_status + }; let notification = CommunicationValue::new(CommunicationType::client_changed) .add_data(DataTypes::user_id, DataValue::Str(user_id.to_string())) - .add_data(DataTypes::user_state, DataValue::Str("online".to_string())); + .add_data(DataTypes::user_state, DataValue::Str(status.to_string())); self.send_message(¬ification).await; } diff --git a/src/data/user.rs b/src/data/user.rs index a175e6d..8587ffe 100644 --- a/src/data/user.rs +++ b/src/data/user.rs @@ -10,6 +10,7 @@ pub enum UserStatus { user_idle, user_wc, user_borked, + user_invisible, iota_offline, iota_online, iota_borked, diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index a5e08dc..563d6a8 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -719,11 +719,13 @@ impl OmegaConnection { let msg_type = match state { UserStatus::iota_offline => CommunicationType::user_disconnected, UserStatus::user_offline => CommunicationType::user_disconnected, + UserStatus::user_invisible => CommunicationType::user_disconnected, _ => CommunicationType::user_connected, }; let cv = CommunicationValue::new(msg_type) - .add_data(DataTypes::user_id, DataValue::Number(user_id)); + .add_data(DataTypes::user_id, DataValue::Number(user_id)) + .add_data(DataTypes::user_state, DataValue::Str(state.to_string())); OMEGA_CONNECTION.send_message(&cv).await; } diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 10669a5..e3d11fa 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -101,6 +101,8 @@ impl ClientConnection { } log_cv_in!(PrintType::Client, cv); + let mut cv = cv; + // Handle client status changes if cv.is_type(CommunicationType::client_changed) { self.handle_client_changed(cv).await; @@ -190,6 +192,76 @@ impl ClientConnection { .await; return; } + + let is_per_device_settings = cv.is_type(CommunicationType::settings_save) + || cv.is_type(CommunicationType::settings_load) + || cv.is_type(CommunicationType::settings_list); + let is_global_settings = cv.is_type(CommunicationType::global_settings_save) + || cv.is_type(CommunicationType::global_settings_load); + + if is_per_device_settings || is_global_settings { + let expected_session_id = self.session_id as i64; + let session_id = cv.get_data(DataTypes::session_id).as_number(); + + if is_per_device_settings { + let Some(session_id) = session_id else { + let response = + CommunicationValue::new(CommunicationType::error_invalid_data) + .with_id(cv.get_id()) + .with_receiver(self.user_id) + .add_data( + DataTypes::message, + DataValue::Str("Missing session_id".to_string()), + ) + .add_data( + DataTypes::session_id, + DataValue::Number(expected_session_id), + ); + self.send_message(&response).await; + return; + }; + + if session_id != expected_session_id { + let response = + CommunicationValue::new(CommunicationType::error_invalid_data) + .with_id(cv.get_id()) + .with_receiver(self.user_id) + .add_data( + DataTypes::message, + DataValue::Str("session_id mismatch".to_string()), + ) + .add_data( + DataTypes::session_id, + DataValue::Number(expected_session_id), + ); + self.send_message(&response).await; + return; + } + } else if let Some(session_id) = session_id { + if session_id != expected_session_id { + let response = + CommunicationValue::new(CommunicationType::error_invalid_data) + .with_id(cv.get_id()) + .with_receiver(self.user_id) + .add_data( + DataTypes::message, + DataValue::Str("session_id mismatch".to_string()), + ) + .add_data( + DataTypes::session_id, + DataValue::Number(expected_session_id), + ); + self.send_message(&response).await; + return; + } + } else { + cv = cv.add_data( + DataTypes::session_id, + DataValue::Number(expected_session_id), + ); + } + } + // Forward other messages to Iota self.forward_to_iota(cv).await; }); @@ -236,8 +308,8 @@ impl ClientConnection { /// Handle client status change async fn handle_client_changed(self: Arc, cv: CommunicationValue) { let user_id = self.get_user_id().await; - if let DataValue::Str(_status_str) = cv.get_data(DataTypes::user_state) { - let user_status = UserStatus::user_online; + if let DataValue::Str(status_str) = cv.get_data(DataTypes::user_state) { + let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online); if let Some(rho_conn) = self.get_rho_connection().await { OmegaConnection::client_changed( rho_conn.get_iota_id().await as i64, @@ -287,9 +359,28 @@ impl ClientConnection { } // Find target RhoConnection - let target_rho = match rho_manager::get_rho_con_for_user(receiver_id).await { + let target_rho = match rho_manager::get_rho_con_for_user(receiver_id).await { Some(rho) => rho, _ => { + // Get sender user ID + let sender_id = self.get_user_id().await; + + // User is offline - send push notification for call invite + let push_cv = CommunicationValue::new(CommunicationType::push_notification) + .with_receiver(receiver_id as u64) + .add_data(DataTypes::sender_id, DataValue::Number(sender_id as i64)) + .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string())) + .add_data( + DataTypes::notifications, + DataValue::Str("call_invite".to_string()), + ); + + let omega_conn = get_omega_connection(); + // Send fire-and-forget, don't await to avoid blocking + tokio::spawn(async move { + let _ = omega_conn.send_message(&push_cv).await; + }); + let error_cv = CommunicationValue::new(CommunicationType::error_not_found) .with_id(cv.get_id()) .add_data(DataTypes::receiver_id, DataValue::Number(receiver_id)); @@ -404,61 +495,73 @@ impl ClientConnection { } async fn handle_call_timeout_user(self: Arc, cv: CommunicationValue) { - let call_id = - Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap(); + let Ok(call_id) = Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")) + else { + self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id) + .await; + return; + }; let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0); let untill = cv.get_data(DataTypes::untill).as_number().unwrap_or(0); - let call = call_manager::get_call(call_id).await; - if let Some(call) = call { - if call - .get_caller(self.get_user_id().await) - .await - .unwrap() - .has_admin() - { - let _ = call_util::remove_participant(call_id, user_id as u64).await; - call.get_caller(user_id as u64) - .await - .unwrap() - .set_timeout(untill) - .await; + let Some(call) = call_manager::get_call(call_id).await else { + self.send_error_response(cv.get_id(), CommunicationType::error_not_found) + .await; + return; + }; + + let Some(caller) = call.get_caller(self.get_user_id().await).await else { + self.send_error_response(cv.get_id(), CommunicationType::error_invalid_user_id) + .await; + return; + }; + if caller.has_admin() { + let _ = call_util::remove_participant(call_id, user_id as u64).await; + if let Some(target) = call.get_caller(user_id as u64).await { + target.set_timeout(untill).await; } } } async fn handle_call_disconnect_user(self: Arc, cv: CommunicationValue) { - let call_id = - Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap(); + let Ok(call_id) = Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")) + else { + self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id) + .await; + return; + }; let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0); - let call = call_manager::get_call(call_id).await; - if let Some(call) = call { - if call - .get_caller(self.get_user_id().await) - .await - .unwrap() - .has_admin() - { - call.remove_caller(user_id as u64).await; - } + let Some(call) = call_manager::get_call(call_id).await else { + self.send_error_response(cv.get_id(), CommunicationType::error_not_found) + .await; + return; + }; + let Some(caller) = call.get_caller(self.get_user_id().await).await else { + self.send_error_response(cv.get_id(), CommunicationType::error_invalid_user_id) + .await; + return; + }; + if caller.has_admin() { + call.remove_caller(user_id as u64).await; } } async fn handle_call_set_anonymous_joining(self: Arc, cv: CommunicationValue) { - let call_id = - Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap(); + let Ok(call_id) = Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")) + else { + self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id) + .await; + return; + }; let enable = cv.get_data(DataTypes::enabled).as_bool().unwrap_or(true); let call = call_manager::get_call(call_id).await; let mut short_link = None; if let Some(call) = call { - if call - .get_caller(self.get_user_id().await) - .await - .unwrap() - .has_admin() - { - call.set_anonymous_joining(enable).await; + if let Some(caller) = call.get_caller(self.get_user_id().await).await { + if caller.has_admin() { + call.set_anonymous_joining(enable).await; + } } short_link = call.get_short_link().await; } @@ -647,12 +750,17 @@ impl ClientConnection { /// Check if interested in a user and send notification #[allow(dead_code)] - pub async fn are_you_interested(self: Arc, user_id: i64) { + pub async fn are_you_interested(self: Arc, user_id: i64, user_status: &str) { let interested_guard = self.clone().get_interested_users().await; if interested_guard.contains(&user_id) { + let status = if user_status == "user_invisible" { + "user_offline" + } else { + user_status + }; let notification = CommunicationValue::new(CommunicationType::client_changed) .add_data(DataTypes::user_id, DataValue::Str(user_id.to_string())) - .add_data(DataTypes::user_state, DataValue::Str("online".to_string())); + .add_data(DataTypes::user_state, DataValue::Str(status.to_string())); self.send_message(¬ification).await; } diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 80daf20..d1f192a 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -122,47 +122,79 @@ impl RhoConnection { /// Add a client connection #[allow(dead_code)] pub async fn add_client_connection(&self, connection: Arc) { - let notification = CommunicationValue::new(CommunicationType::client_connected).add_data( - DataTypes::user_id, - DataValue::Number(connection.get_user_id().await as i64), - ); + let user_id = connection.user_id as i64; + let session_id = connection.session_id as i64; + + let notification = CommunicationValue::new(CommunicationType::client_connected) + .add_data(DataTypes::user_id, DataValue::Number(user_id)); self.iota_connection.send_message(¬ification).await; + let mut should_notify_online = false; { let mut connections = self.client_connections.write().await; - connections.push(Arc::clone(&connection)); + let mut keep = Vec::new(); + let mut had_user = false; + + for con in connections.drain(..) { + if con.user_id as i64 == user_id { + had_user = true; + if con.session_id as i64 == session_id { + continue; + } + } + keep.push(con); + } + + if !had_user { + should_notify_online = true; + } + + keep.push(Arc::clone(&connection)); + *connections = keep; } - OmegaConnection::client_changed( - self.get_iota_id().await as i64, - connection.get_user_id().await as i64, - UserStatus::user_online, - ) - .await; + if should_notify_online { + OmegaConnection::client_changed( + self.get_iota_id().await as i64, + user_id, + UserStatus::user_online, + ) + .await; + } } /// Remove a client connection pub async fn close_client_connection(&self, connection: Arc) { - let target_user_id = connection.get_user_id().await; + let target_user_id = connection.user_id as i64; + let target_session_id = connection.session_id as i64; + let mut remaining_for_user = false; + { let mut connections = self.client_connections.write().await; let mut keep = Vec::new(); for con in connections.drain(..) { - if con.get_user_id().await != target_user_id { - keep.push(con); + if con.user_id as i64 == target_user_id + && con.session_id as i64 == target_session_id + { + continue; } + if con.user_id as i64 == target_user_id { + remaining_for_user = true; + } + keep.push(con); } *connections = keep; } - // Notify OmegaConnection - OmegaConnection::client_changed( - self.get_iota_id().await as i64, - connection.get_user_id().await as i64, - UserStatus::user_offline, - ) - .await; + if !remaining_for_user { + OmegaConnection::client_changed( + self.get_iota_id().await as i64, + target_user_id, + UserStatus::user_offline, + ) + .await; + } } /// Close the Iota connection and all associated client connections @@ -184,10 +216,18 @@ impl RhoConnection { pub async fn message_to_client(&self, cv: CommunicationValue) { let connections = self.client_connections.read().await; let receiver_id = cv.get_receiver(); + let session_id = cv.get_data(DataTypes::session_id).as_number(); + for connection in connections.iter() { - if connection.get_user_id().await == receiver_id { - connection.clone().send_message(&cv).await; + if connection.user_id != receiver_id { + continue; } + if let Some(session_id) = session_id { + if connection.session_id as i64 != session_id { + continue; + } + } + connection.clone().send_message(&cv).await; } } @@ -213,10 +253,13 @@ impl RhoConnection { /// Check if clients are interested in a user #[allow(dead_code)] - pub async fn are_they_interested(&self, user_id: i64) { + pub async fn are_they_interested(&self, user_id: i64, user_status: &str) { let connections = self.client_connections.read().await; for connection in connections.iter() { - connection.clone().are_you_interested(user_id).await; + connection + .clone() + .are_you_interested(user_id, user_status) + .await; } }