From c999f68aa42315b65bdb8a4d031f02230eed35ce Mon Sep 17 00:00:00 2001 From: Alois Date: Tue, 28 Jul 2026 12:44:48 +0200 Subject: [PATCH] (fix): connections --- Cargo.lock | 32 ++--- .../anonymous_client_connection.rs | 32 +---- src/omega/omega_connection.rs | 121 ++---------------- src/omega/ping_pong_task.rs | 41 ------ src/rho/app_connection.rs | 49 +------ src/rho/client_connection.rs | 48 +------ src/rho/iota_connection.rs | 57 +-------- src/rho/rho_connection.rs | 14 -- 8 files changed, 31 insertions(+), 363 deletions(-) delete mode 100644 src/omega/ping_pong_task.rs diff --git a/Cargo.lock b/Cargo.lock index 5cea467..ecfd4e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -616,7 +616,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -1581,7 +1581,7 @@ dependencies = [ [[package]] name = "mtp" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#fa271e62bec7918a8c994eea59ecdd62be76ead4" dependencies = [ "mtp-client", "mtp-codec", @@ -1597,7 +1597,7 @@ dependencies = [ [[package]] name = "mtp-client" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#fa271e62bec7918a8c994eea59ecdd62be76ead4" dependencies = [ "mtp-codec", "mtp-common", @@ -1610,7 +1610,7 @@ dependencies = [ [[package]] name = "mtp-codec" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#fa271e62bec7918a8c994eea59ecdd62be76ead4" dependencies = [ "base64 0.22.1", "byteorder", @@ -1623,7 +1623,7 @@ dependencies = [ [[package]] name = "mtp-common" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#fa271e62bec7918a8c994eea59ecdd62be76ead4" dependencies = [ "quinn", "rustls", @@ -1634,7 +1634,7 @@ dependencies = [ [[package]] name = "mtp-crypto" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#fa271e62bec7918a8c994eea59ecdd62be76ead4" dependencies = [ "base64 0.22.1", "chacha20poly1305", @@ -1656,7 +1656,7 @@ dependencies = [ [[package]] name = "mtp-files" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#fa271e62bec7918a8c994eea59ecdd62be76ead4" dependencies = [ "mtp-crypto", "rand 0.10.2", @@ -1667,7 +1667,7 @@ dependencies = [ [[package]] name = "mtp-host" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#fa271e62bec7918a8c994eea59ecdd62be76ead4" dependencies = [ "mtp-codec", "mtp-common", @@ -1682,7 +1682,7 @@ dependencies = [ [[package]] name = "mtp-transport" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#fa271e62bec7918a8c994eea59ecdd62be76ead4" dependencies = [ "async-trait", "mtp-codec", @@ -1700,7 +1700,7 @@ dependencies = [ [[package]] name = "mtp-type-map" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#fa271e62bec7918a8c994eea59ecdd62be76ead4" dependencies = [ "serde", "serde_yaml", @@ -1709,7 +1709,7 @@ dependencies = [ [[package]] name = "mtp-webserver" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#fa271e62bec7918a8c994eea59ecdd62be76ead4" dependencies = [ "async-trait", "bytes", @@ -2327,7 +2327,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2581,7 +2581,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2640,7 +2640,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3035,7 +3035,7 @@ dependencies = [ "getrandom 0.3.4", "once_cell", "rustix", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3596,7 +3596,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index da2c650..a333260 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -20,7 +20,6 @@ pub struct AnonymousClientConnection { pub sender: Arc, pub receiver: Arc, - pub ping: Arc>, pub interested_users: Arc>>, is_open: Arc>, pub user_name: Arc>, @@ -36,7 +35,6 @@ impl AnonymousClientConnection { state: general.state.clone(), user_id: user_id, - ping: Arc::new(RwLock::new(0)), interested_users: Arc::new(RwLock::new(Vec::new())), is_open: Arc::new(RwLock::new(true)), sender: general.sender.clone(), @@ -91,9 +89,7 @@ impl AnonymousClientConnection { ); return; } - if !cv.is_type(CommunicationType::Pong) { - log_cv_out!(PrintType::Client, &cv); - } + log_cv_out!(PrintType::Client, &cv); if let Err(e) = self.sender.send(&cv).await { log_out!( self.user_id as i64, @@ -111,10 +107,6 @@ impl AnonymousClientConnection { }; tokio::spawn(async move { let _permit = permit; - if cv.is_type(CommunicationType::Ping) { - self.handle_ping(cv).await; - return; - } log_cv_in!(PrintType::Client, &cv); if cv.is_type(CommunicationType::Identification) { @@ -230,11 +222,6 @@ impl AnonymousClientConnection { .await; } - // Handle ping - if cv.is_type(CommunicationType::Ping) { - self.handle_ping(cv).await; - return; - } // Handle client status changes if cv.is_type(CommunicationType::ClientChanged) { self.handle_client_changed(cv).await; @@ -334,22 +321,6 @@ impl AnonymousClientConnection { }); } - /// Handle ping message - async fn handle_ping(self: Arc, cv: CommunicationValue) { - // Update our ping if provided - if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) { - if let Ok(ping_val) = last_ping.to_string().parse::() { - let mut ping_guard = self.ping.write().await; - *ping_guard = ping_val; - } - } - - // Send pong response - let response = CommunicationValue::new(CommunicationType::Pong).with_id(cv.get_id()); - - self.send_message(&response).await; - } - /// Handle client status change async fn handle_client_changed(self: Arc, cv: CommunicationValue) { if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) { @@ -648,7 +619,6 @@ impl Clone for AnonymousClientConnection { sender: Arc::clone(&self.sender), receiver: Arc::clone(&self.receiver), user_id: self.user_id, - ping: Arc::clone(&self.ping), interested_users: Arc::clone(&self.interested_users), is_open: Arc::clone(&self.is_open), user_name: Arc::clone(&self.user_name), diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index a7806ea..a9c9dbb 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -11,7 +11,7 @@ use mtp::{ host::{Policy, SendMode}, }; use mtp_transport::ConnectionHandle; -use std::{collections::HashMap, sync::Arc, time::Duration}; +use std::{sync::Arc, time::Duration}; use tokio::{ sync::{Mutex, RwLock, mpsc, watch}, task::JoinHandle, @@ -23,8 +23,8 @@ use uuid::Uuid; const RECONNECT_DELAY: Duration = Duration::from_secs(5); const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(300); const CONNECTION_TIMEOUT: Duration = Duration::from_secs(10); -const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5); -const MAX_HEARTBEAT_FAILURES: usize = 3; +const PING_INTERVAL: Duration = Duration::from_secs(5); +const MAX_MISSED_PINGS: usize = 3; const TASK_CLEANUP_INTERVAL: Duration = Duration::from_secs(60); const TASK_MAX_AGE: Duration = Duration::from_secs(60); const MAX_CONCURRENT_REQUESTS: usize = 128; @@ -118,9 +118,6 @@ pub struct OmegaConnection { host: String, port: u16, omikron_id: u64, - last_ping: Arc>, - heartbeat_handle: Arc>>>, - message_send_times: Arc>>, pub connection_id: Uuid, shutdown_tx: Arc>>>, // Track if we should reconnect on close @@ -147,9 +144,6 @@ impl OmegaConnection { host: config.omega_host.clone(), port: config.omega_port, omikron_id: config.omikron_id, - last_ping: Arc::new(Mutex::new(-1)), - heartbeat_handle: Arc::new(Mutex::new(None)), - message_send_times: Arc::new(Mutex::new(HashMap::new())), connection_id: Uuid::new_v4(), shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))), reconnect_on_close: Arc::new(RwLock::new(true)), @@ -204,10 +198,6 @@ impl OmegaConnection { handle.abort(); } - if let Some(handle) = self.heartbeat_handle.lock().await.take() { - handle.abort(); - } - // Close sender if connected if let Some(sender) = self.sender.read().await.as_ref() { sender.close().await; @@ -292,7 +282,9 @@ impl OmegaConnection { .with_receiver_queue_capacity(1000) .with_max_concurrent_stream_tasks(64) .with_persistent_stream_retries(5, Duration::from_secs(5)), - ); + ) + .with_ping_interval(PING_INTERVAL) + .with_max_missed_pings(MAX_MISSED_PINGS); let host_public_key = load_public_key_bundle("./omega.mpkb") .map_err(|e| format!("Failed to load omega.mpkb: {}", e))?; @@ -323,13 +315,6 @@ impl OmegaConnection { read_self.read_loop(connection, sender_handle).await; }); - // Start heartbeat - let heartbeat_self = self.clone(); - let heartbeat_handle = tokio::spawn(async move { - heartbeat_self.heartbeat_loop().await; - }); - *self.heartbeat_handle.lock().await = Some(heartbeat_handle); - // Tell omega our current state now that we're actually connected - // doing this after teardown (as before) sent into a sender that had // already been cleared, silently dropping the sync every time. @@ -357,10 +342,6 @@ impl OmegaConnection { *self.sender.write().await = None; *self.state.write().await = ConnectionState::Disconnected; - if let Some(handle) = self.heartbeat_handle.lock().await.take() { - handle.abort(); - } - match result { Ok(()) => { // Check if we should reconnect @@ -405,7 +386,7 @@ impl OmegaConnection { } // ------------------------------------------------------------------------- - // Read Loop & Heartbeat + // Read Loop // ------------------------------------------------------------------------- async fn read_loop( @@ -421,14 +402,7 @@ impl OmegaConnection { result = connection.receive() => { match result { Ok(cv) => { - if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) { - log_cv_in!(PrintType::Omega, &cv); - } - - if cv.is_type(CommunicationType::Pong) || cv.is_type(CommunicationType::Ping) { - self.handle_pong(&cv).await; - continue; - } + log_cv_in!(PrintType::Omega, &cv); let msg_id = cv.get_id(); if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) { @@ -477,89 +451,12 @@ impl OmegaConnection { } } - async fn heartbeat_loop(self: Arc) { - let mut consecutive_failures = 0; - loop { - sleep(HEARTBEAT_INTERVAL).await; - - // Check if still connected - if !self.state.read().await.is_connected() { - break; - } - - // Check if sender is closed - if let Some(sender) = self.sender.read().await.as_ref() { - if sender.is_closed() { - log_err!(0, PrintType::Omega, "Sender closed, stopping heartbeat"); - break; - } - } else { - break; - } - - match self.send_ping().await { - Ok(()) => consecutive_failures = 0, - Err(error) => { - consecutive_failures += 1; - log_err!( - 0, - PrintType::Omega, - "Heartbeat send failed ({}/{}): {}", - consecutive_failures, - MAX_HEARTBEAT_FAILURES, - error - ); - if consecutive_failures >= MAX_HEARTBEAT_FAILURES { - log_err!( - 0, - PrintType::Omega, - "Closing unhealthy Omega connection after repeated heartbeat failures" - ); - if let Some(sender) = self.sender.read().await.as_ref().cloned() { - sender.close().await; - } - break; - } - } - } - } - } - - async fn send_ping(&self) -> Result<(), String> { - let ping = CommunicationValue::new(CommunicationType::Ping).add_typed_default( - DataType::SendTime, - DataValue::SignedNumber( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs() as i128, - ), - ); - self.try_send_message(&ping).await - } - - async fn handle_pong(&self, cv: &CommunicationValue) { - let timestamp = cv - .get_data(DataType::SendTime) - .as_number() - .unwrap_or_else(|| { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs() as i128 - }); - - *self.last_ping.lock().await = timestamp.try_into().unwrap(); - } - // ------------------------------------------------------------------------- // Public API // ------------------------------------------------------------------------- async fn try_send_message(&self, cv: &CommunicationValue) -> Result<(), String> { - if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) { - log_cv_out!(PrintType::Omega, &cv); - } + log_cv_out!(PrintType::Omega, &cv); let sender_guard = self.sender.read().await; if let Some(sender) = sender_guard.as_ref() { diff --git a/src/omega/ping_pong_task.rs b/src/omega/ping_pong_task.rs deleted file mode 100644 index a4e4687..0000000 --- a/src/omega/ping_pong_task.rs +++ /dev/null @@ -1,41 +0,0 @@ -use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; -use std::time::Duration; -use tokio::time::Instant; - -use crate::omega::omega_connection::OmegaConnection; - -const PING_TIMEOUT: Duration = Duration::from_secs(30); - -impl OmegaConnection { - pub async fn send_ping(&self) { - let id = rand::random::(); - let send_time = Instant::now(); - - let mut message_send_times = self.message_send_times.lock().await; - message_send_times.retain(|_uuid, time| time.elapsed() < PING_TIMEOUT); - message_send_times.insert(id as i64, send_time); - - self.send_ping_message(id).await; - } - - pub async fn send_ping_message(&self, id: u32) { - let ping_message = CommunicationValue::new(CommunicationType::Ping) - .with_id(id) - .add_data( - DataType::LastPing, - DataValue::SignedNumber(self.last_ping.lock().await.unwrap()), - ); - - self.send_message(&ping_message).await; - } - - /// Handles incoming pong and calculates latency - pub async fn handle_pong(&self, cv: &CommunicationValue, _log: bool) { - let id = cv.get_id(); - let mut message_send_times = self.message_send_times.lock().await; - if let Some(send_time) = message_send_times.remove(&(id as i64)) { - let ping = Instant::now().duration_since(send_time).as_millis() as i64; - *self.last_ping.lock().await = ping; - } - } -} diff --git a/src/rho/app_connection.rs b/src/rho/app_connection.rs index 855104b..bb71c00 100644 --- a/src/rho/app_connection.rs +++ b/src/rho/app_connection.rs @@ -6,7 +6,7 @@ use crate::util::logger::PrintType; use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use std::sync::Arc; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::Duration; use tokio::sync::RwLock; use uuid::Uuid; @@ -19,7 +19,6 @@ pub struct AppConnection { pub sender: Arc, pub receiver: Arc, - pub ping: Arc>, pub_key: Arc>>>, pub rho_connection: Arc>>>, is_open: Arc>, @@ -29,7 +28,6 @@ impl AppConnection { pub async fn from_general(general: Arc, user_id: u64) -> Arc { Arc::new(Self { state: general.state.clone(), - ping: Arc::new(RwLock::new(0)), pub_key: Arc::new(RwLock::new(None)), rho_connection: general.rho_connection.clone(), is_open: Arc::new(RwLock::new(true)), @@ -66,12 +64,6 @@ impl AppConnection { self.user_id } - /// Get current ping - #[allow(unused)] - pub async fn get_ping(&self) -> i64 { - *self.ping.read().await - } - /// Get RhoConnection if available pub async fn get_rho_connection(&self) -> Option> { self.rho_connection.read().await.clone() @@ -87,19 +79,13 @@ impl AppConnection { ); return; } - if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) { - log_cv_out!(PrintType::App, &cv); - } + log_cv_out!(PrintType::App, &cv); let _ = self.sender.send(&cv).await; } /// Handle incoming message from app pub async fn handle_message(self: Arc, cv: CommunicationValue) { tokio::spawn(async move { - if cv.is_type(CommunicationType::Ping) { - self.handle_ping(cv).await; - return; - } log_cv_in!(PrintType::App, cv); if cv.is_type(CommunicationType::GetUserData) { @@ -168,36 +154,6 @@ impl AppConnection { }); } - /// Handle ping message - async fn handle_ping(self: Arc, cv: CommunicationValue) { - // Update our ping if provided - if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) { - let current = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_millis(); - let mut ping_guard = self.ping.write().await; - *ping_guard = (current as i128 - *last_ping) as i64; - } - - // Get Iota ping from RhoConnection - let iota_ping = if let Some(rho_conn) = self.get_rho_connection().await { - rho_conn.get_iota_connection().get_ping().await - } else { - -1 - }; - - // Send pong response - let response = CommunicationValue::new(CommunicationType::Pong) - .with_id(cv.get_id()) - .add_typed_default( - DataType::PingIota, - DataValue::SignedNumber(iota_ping.into()), - ); - - self.send_message(&response).await; - } - /// Forward message to Iota async fn forward_to_iota(self: Arc, cv: CommunicationValue) { let sender_user_id = self.get_user_id().await; @@ -353,7 +309,6 @@ impl Clone for AppConnection { app_identifier: self.app_identifier.clone(), app_session: self.app_session, client_version: self.client_version.clone(), - ping: Arc::clone(&self.ping), pub_key: Arc::clone(&self.pub_key), rho_connection: Arc::clone(&self.rho_connection), is_open: Arc::clone(&self.is_open), diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 438abd7..0a618d8 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -9,7 +9,7 @@ use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use std::str::FromStr; use std::sync::Arc; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::Duration; use tokio::sync::RwLock; use trust_dns_resolver::TokioAsyncResolver; use uuid::Uuid; @@ -22,7 +22,6 @@ pub struct ClientConnection { pub sender: Arc, pub receiver: Arc, - pub ping: Arc>, pub_key: Arc>>>, pub rho_connection: Arc>>>, pub interested_users: Arc>>, @@ -34,7 +33,6 @@ impl ClientConnection { pub async fn from_general(general: Arc, user_id: u64) -> Arc { Arc::new(Self { state: general.state.clone(), - ping: Arc::new(RwLock::new(0)), pub_key: Arc::new(RwLock::new(None)), rho_connection: general.rho_connection.clone(), interested_users: Arc::new(RwLock::new(Vec::new())), @@ -72,11 +70,6 @@ impl ClientConnection { self.user_id } - /// Get current ping - pub async fn get_ping(&self) -> i64 { - *self.ping.read().await - } - /// Get RhoConnection if available pub async fn get_rho_connection(&self) -> Option> { self.rho_connection.read().await.clone() @@ -92,9 +85,7 @@ impl ClientConnection { ); return; } - if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) { - log_cv_out!(PrintType::Client, &cv); - } + log_cv_out!(PrintType::Client, &cv); let _ = self.sender.send(&cv).await; } @@ -105,10 +96,6 @@ impl ClientConnection { }; tokio::spawn(async move { let _permit = permit; - if cv.is_type(CommunicationType::ClientPing) { - self.handle_ping(cv).await; - return; - } log_cv_in!(PrintType::Client, cv); let mut cv = cv; @@ -311,36 +298,6 @@ impl ClientConnection { } } - /// Handle ping message - async fn handle_ping(self: Arc, cv: CommunicationValue) { - // Update our ping if provided - if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) { - let current = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_millis(); - let mut ping_guard = self.ping.write().await; - *ping_guard = (current as i128 - *last_ping) as i64; - } - - // Get Iota ping from RhoConnection - let iota_ping = if let Some(rho_conn) = self.get_rho_connection().await { - rho_conn.get_iota_connection().get_ping().await - } else { - -1 - }; - - // Send pong response - let response = CommunicationValue::new(CommunicationType::ClientPing) - .with_id(cv.get_id()) - .add_typed_default( - DataType::PingIota, - DataValue::SignedNumber(iota_ping.into()), - ); - - self.send_message(&response).await; - } - /// Handle client status change async fn handle_client_changed(self: Arc, cv: CommunicationValue) { let user_id = self.get_user_id().await; @@ -917,7 +874,6 @@ impl Clone for ClientConnection { user_id: self.user_id, session_id: self.session_id, client_version: self.client_version.clone(), - ping: Arc::clone(&self.ping), pub_key: Arc::clone(&self.pub_key), rho_connection: Arc::clone(&self.rho_connection), interested_users: Arc::clone(&self.interested_users), diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 7f6fe4c..7d1b78e 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -31,7 +31,6 @@ pub struct IotaConnection { pub sender: Arc, pub receiver: Arc, pub user_ids: Arc>>, - pub ping: Arc>, pub_key: Arc>>>, pub waiting_tasks: DashMap, CommunicationValue) -> bool + Send + Sync>>, @@ -44,7 +43,6 @@ impl IotaConnection { pub async fn from_general(general: Arc, iota_id: u64) -> Arc { Arc::new(Self { state: general.state.clone(), - ping: Arc::new(RwLock::new(0)), pub_key: Arc::new(RwLock::new(None)), rho_connection: general.rho_connection.clone(), user_ids: Arc::new(RwLock::new(Vec::new())), @@ -151,11 +149,6 @@ impl IotaConnection { .push(cv); } - /// Get current ping - pub async fn get_ping(&self) -> i64 { - *self.ping.read().await - } - /// Set the RhoConnection reference pub async fn set_rho_connection(&self, rho_connection: Arc) { let mut rho_ref = self.rho_connection.write().await; @@ -174,9 +167,7 @@ impl IotaConnection { /// Send a CommunicationValue to the Iota pub async fn send_message(&self, cv: &CommunicationValue) { - if !cv.is_type(CommunicationType::Pong) { - log_cv_out!(PrintType::Iota, cv); - } + log_cv_out!(PrintType::Iota, cv); if let Err(e) = self.sender.send(&cv).await { log_err!( self.iota_id as i64, @@ -200,12 +191,6 @@ impl IotaConnection { } } - // Handle ping - if cv.is_type(CommunicationType::Ping) || cv.is_type(CommunicationType::Pong) { - self.handle_ping(cv).await; - return; - } - log_cv_in!(PrintType::Iota, cv); let cv = if cv.is_type(CommunicationType::ClientStateSync) { @@ -452,46 +437,6 @@ impl IotaConnection { } } } - /// Handle ping message - async fn handle_ping(&self, cv: CommunicationValue) { - if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) { - if let Ok(ping_val) = last_ping.to_string().parse::() { - let mut ping_guard = self.ping.write().await; - *ping_guard = ping_val; - } - } - - let client_pings = if let Some(rho_conn) = self.get_rho_connection().await { - rho_conn.get_client_pings().await - } else { - HashMap::new() - }; - - let tm = TypeMap::latest(); - let pings: Vec = client_pings - .into_iter() - .map(|(k, v)| { - let mut map = BTreeMap::new(); - if let Ok(uid) = k.parse::() { - map.insert( - data_type_id(DataType::UserId, &tm), - DataValue::SignedNumber(uid), - ); - } - map.insert( - data_type_id(DataType::LastPing, &tm), - DataValue::SignedNumber(v.into()), - ); - DataValue::container_from_map(&map) - }) - .collect(); - let response = CommunicationValue::new(CommunicationType::Pong) - .with_id(cv.get_id()) - .add_typed_default(DataType::PingClients, DataValue::Array(pings)); - - self.send_message(&response).await; - } - /// Handle message forwarding to other Iotas async fn handle_forward_message(&self, cv: CommunicationValue) { let receiver_id = cv.get_receiver(); diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index c7e4f65..35e259d 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -3,7 +3,6 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection use crate::{data::user::UserStatus, rho::app_connection::AppConnection}; use dashmap::DashMap; use mtp::codec::{CommunicationValue, DataType}; -use std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; use uuid::Uuid; @@ -263,19 +262,6 @@ impl RhoConnection { } } - /// Get ping information for all clients - pub async fn get_client_pings(&self) -> HashMap { - let connections = self.get_client_connections().await; - let mut pings = HashMap::new(); - - for connection in connections.iter() { - let user_id = connection.get_user_id().await; - pings.insert(user_id.to_string(), connection.get_ping().await); - } - - pings - } - /// Check if this RhoConnection contains a specific user ID #[allow(dead_code)] pub async fn contains_user(&self, user_id: &i64) -> bool {