From 7c0febc6e3b7431103835b828554b716deb090dc Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Sat, 9 May 2026 14:52:22 +0200 Subject: [PATCH] [Add] Parrallel message sending --- Cargo.lock | 79 +++++++++++++++-------------------- src/omega/omega_connection.rs | 27 ++++++++++-- src/rho/connection.rs | 5 +-- src/rho/server.rs | 27 +++++++++++- 4 files changed, 84 insertions(+), 54 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 106aa7f..13fca3d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -385,9 +385,9 @@ checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" [[package]] name = "cc" -version = "1.2.61" +version = "1.2.62" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d16d90359e986641506914ba71350897565610e87ce0ad9e6f28569db3dd5c6d" +checksum = "a1dce859f0832a7d088c4f1119888ab94ef4b5d6795d1ce05afb7fe159d79f98" dependencies = [ "find-msvc-tools", "jobserver", @@ -702,9 +702,9 @@ dependencies = [ [[package]] name = "digest" -version = "0.11.2" +version = "0.11.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4850db49bf08e663084f7fb5c87d202ef91a3907271aff24a94eb97ff039153c" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" dependencies = [ "block-buffer 0.12.0", "const-oid 0.10.2", @@ -1160,9 +1160,9 @@ dependencies = [ [[package]] name = "hashbrown" -version = "0.17.0" +version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" [[package]] name = "heck" @@ -1267,7 +1267,7 @@ version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" dependencies = [ - "digest 0.11.2", + "digest 0.11.3", ] [[package]] @@ -1510,7 +1510,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", - "hashbrown 0.17.0", + "hashbrown 0.17.1", "serde", "serde_core", ] @@ -1543,16 +1543,6 @@ version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" -[[package]] -name = "iri-string" -version = "0.7.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "25e659a4bb38e810ebc252e53b5814ff908a8c58c2a9ce2fae1bbec24cbf4e20" -dependencies = [ - "memchr", - "serde", -] - [[package]] name = "itertools" version = "0.11.0" @@ -1633,9 +1623,9 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.97" +version = "0.3.98" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1840c94c045fbcf8ba2812c95db44499f7c64910a912551aaaa541decebcacf" +checksum = "67df7112613f8bfd9150013a0314e196f4800d3201ae742489d999db2f979f08" dependencies = [ "cfg-if", "futures-util", @@ -2170,15 +2160,14 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "openssl" -version = "0.10.78" +version = "0.10.79" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f38c4372413cdaaf3cc79dd92d29d7d9f5ab09b51b10dded508fb90bb70b9222" +checksum = "bf0b434746ee2832f4f0baf10137e1cabb18cbe6912c69e2e33263c45250f542" dependencies = [ "bitflags", "cfg-if", "foreign-types", "libc", - "once_cell", "openssl-macros", "openssl-sys", ] @@ -2202,9 +2191,9 @@ checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" [[package]] name = "openssl-sys" -version = "0.9.114" +version = "0.9.115" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "13ce1245cd07fcc4cfdb438f7507b0c7e4f3849a69fd84d52374c66d83741bb6" +checksum = "158fe5b292746440aa6e7a7e690e55aeb72d41505e2804c23c6973ad0e9c9781" dependencies = [ "cc", "libc", @@ -3070,7 +3059,7 @@ checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" dependencies = [ "cfg-if", "cpufeatures 0.3.0", - "digest 0.11.2", + "digest 0.11.3", ] [[package]] @@ -3315,9 +3304,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.52.1" +version = "1.52.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" dependencies = [ "bytes", "libc", @@ -3393,20 +3382,20 @@ dependencies = [ [[package]] name = "tower-http" -version = "0.6.8" +version = "0.6.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8" +checksum = "68d6fdd9f81c2819c9a8b0e0cd91660e7746a8e6ea2ba7c6b2b057985f6bcb51" dependencies = [ "bitflags", "bytes", "futures-util", "http 1.4.0", "http-body", - "iri-string", "pin-project-lite", "tower", "tower-layer", "tower-service", + "url", ] [[package]] @@ -3461,7 +3450,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "ttp-core" version = "0.1.0" -source = "git+https://git.methanium.net/Tensamin/TTP.git#0cc917d75562953b49a83804f9f37e89aca6fd88" +source = "git+https://git.methanium.net/Tensamin/TTP.git#929cb9d3a6aebebe6365973f13062ac2a8e03af6" dependencies = [ "base64 0.22.1", "byteorder", @@ -3474,7 +3463,7 @@ dependencies = [ [[package]] name = "ttp-native" version = "0.1.0" -source = "git+https://git.methanium.net/Tensamin/TTP.git#0cc917d75562953b49a83804f9f37e89aca6fd88" +source = "git+https://git.methanium.net/Tensamin/TTP.git#929cb9d3a6aebebe6365973f13062ac2a8e03af6" dependencies = [ "quinn", "rustls", @@ -3663,9 +3652,9 @@ dependencies = [ [[package]] name = "wasm-bindgen" -version = "0.2.120" +version = "0.2.121" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df52b6d9b87e0c74c9edfa1eb2d9bf85e5d63515474513aa50fa181b3c4f5db1" +checksum = "49ace1d07c165b0864824eee619580c4689389afa9dc9ed3a4c75040d82e6790" dependencies = [ "cfg-if", "once_cell", @@ -3676,9 +3665,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-futures" -version = "0.4.70" +version = "0.4.71" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af934872acec734c2d80e6617bbb5ff4f12b052dd8e6332b0817bce889516084" +checksum = "96492d0d3ffba25305a7dc88720d250b1401d7edca02cc3bcd50633b424673b8" dependencies = [ "js-sys", "wasm-bindgen", @@ -3686,9 +3675,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.120" +version = "0.2.121" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78b1041f495fb322e64aca85f5756b2172e35cd459376e67f2a6c9dffcedb103" +checksum = "8e68e6f4afd367a562002c05637acb8578ff2dea1943df76afb9e83d177c8578" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -3696,9 +3685,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.120" +version = "0.2.121" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9dcd0ff20416988a18ac686d4d4d0f6aae9ebf08a389ff5d29012b05af2a1b41" +checksum = "d95a9ec35c64b2a7cb35d3fead40c4238d0940c86d107136999567a4703259f2" dependencies = [ "bumpalo", "proc-macro2", @@ -3709,9 +3698,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.120" +version = "0.2.121" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49757b3c82ebf16c57d69365a142940b384176c24df52a087fb748e2085359ea" +checksum = "c4e0100b01e9f0d03189a92b96772a1fb998639d981193d7dbab487302513441" dependencies = [ "unicode-ident", ] @@ -3752,9 +3741,9 @@ dependencies = [ [[package]] name = "web-sys" -version = "0.3.97" +version = "0.3.98" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2eadbac71025cd7b0834f20d1fe8472e8495821b4e9801eb0a60bd1f19827602" +checksum = "4b572dff8bcf38bad0fa19729c89bb5748b2b9b1d8be70cf90df697e3a8f32aa" dependencies = [ "js-sys", "wasm-bindgen", diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index f36102c..211ed6d 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -17,7 +17,7 @@ use tokio::{ time::{Instant, sleep}, }; use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue, rand_u32}; -use ttp_native::{Receiver, Sender}; +use ttp_native::{Policy, Receiver, SendMode, Sender}; use uuid::Uuid; // ============================================================================ @@ -243,9 +243,28 @@ impl OmegaConnection { let addr_str = format!("https://{}:{}", self.host, self.port); - let (sender, mut receiver) = ttp_native::client::connect(&addr_str, None) - .await - .map_err(|e| format!("Connection failed: {}", e))?; + let (sender, mut receiver) = ttp_native::client::connect( + &addr_str, + None, + Policy { + send_mode: SendMode::SingleStreamPerMessage, + max_message_size: 1_000_000_000, + close_frame_len: u32::MAX, + application_close_code: 0, + open_stream_timeout: Duration::from_millis(2_000), + write_timeout: Duration::from_millis(2_000), + accept_stream_timeout: Duration::from_millis(10_000), + read_timeout: Duration::from_millis(30_000), + keep_alive_interval: Some(Duration::from_secs(6)), + max_idle_timeout: Some(Duration::from_secs(30)), + force_close_delay: Duration::from_millis(300), + max_transient_recv_errors: 20, + transient_recv_backoff: Duration::from_millis(100), + receiver_queue_capacity: 1000, + }, + ) + .await + .map_err(|e| format!("Connection failed: {}", e))?; log_in!( 0, diff --git a/src/rho/connection.rs b/src/rho/connection.rs index f5866ad..7f1fafc 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -411,9 +411,6 @@ impl GeneralConnection { *self.rho_connection.write().await = rho.clone(); if let Some(rho_conn) = rho { - rho_conn.bind_user_id(user_id).await; - rho_conn.add_client_connection(client.clone()).await; - let session_id = *self.session_id.read().await as i64; let iota_msg = CommunicationValue::new(CommunicationType::client_connected) .add_data(DataTypes::user_id, DataValue::Number(user_id)) @@ -546,6 +543,8 @@ impl GeneralConnection { } log_cv_out!(ident_resp); let _ = self.sender.send(&ident_resp).await; + rho_conn.bind_user_id(user_id).await; + rho_conn.add_client_connection(client.clone()).await; } } else { log_err!( diff --git a/src/rho/server.rs b/src/rho/server.rs index 9e3946d..caaf371 100644 --- a/src/rho/server.rs +++ b/src/rho/server.rs @@ -1,16 +1,39 @@ +use std::time::Duration; + use crate::{ log, rho::connection::GeneralConnection, util::{file_util::load_file_vec, logger::PrintType}, }; -use ttp_native::Host; +use ttp_native::{Host, Policy, SendMode, host}; pub async fn start(port: u16) -> Result<(), Box> { let cert_pem = load_file_vec("certs", "cert.pem").expect("Error loading Pemfile"); let key_pem = load_file_vec("certs", "key.pem").expect("Error loading Keyfile"); - let mut host: Host = ttp_native::host(port, cert_pem, key_pem).await?; + let mut host: Host = host( + port, + cert_pem, + key_pem, + Policy { + send_mode: SendMode::SingleStreamPerMessage, + max_message_size: 1_000_000_000, + close_frame_len: u32::MAX, + application_close_code: 0, + open_stream_timeout: Duration::from_millis(2_000), + write_timeout: Duration::from_millis(2_000), + accept_stream_timeout: Duration::from_millis(10_000), + read_timeout: Duration::from_millis(30_000), + keep_alive_interval: Some(Duration::from_secs(6)), + max_idle_timeout: Some(Duration::from_secs(30)), + force_close_delay: Duration::from_millis(300), + max_transient_recv_errors: 20, + transient_recv_backoff: Duration::from_millis(100), + receiver_queue_capacity: 1000, + }, + ) + .await?; log!(0, PrintType::General, "Server listening on port {}", port); while let Some((sender, receiver)) = host.next().await {