diff --git a/Cargo.lock b/Cargo.lock index 0e1cb25..deeca7d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -688,58 +688,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "communities" -version = "0.1.0" -dependencies = [ - "actix-web", - "actix-web-actors", - "aes-gcm", - "async-trait", - "base64", - "chrono", - "crossterm", - "dashmap", - "futures", - "futures-util", - "hex", - "hkdf", - "hyper", - "hyper-util", - "iota-auth", - "iota-logger", - "iota-state", - "iota-storage", - "iota-util", - "json", - "lazy_static", - "once_cell", - "open", - "pnet", - "rand 0.8.6", - "rand_core 0.6.4", - "ratatui", - "reqwest", - "rusqlite", - "rustls", - "rustls-pemfile", - "serde_json", - "sha2 0.10.9", - "strum 0.27.2", - "strum_macros 0.27.2", - "sysinfo", - "tokio", - "tokio-tungstenite", - "ttp-core", - "ttp-native", - "tungstenite", - "uuid", - "walkdir", - "warp", - "x448", - "zip", -] - [[package]] name = "compact_str" version = "0.9.0" @@ -1071,9 +1019,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", @@ -1455,9 +1403,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.13" +version = "0.4.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54" +checksum = "171fefbc92fe4a4de27e0698d6a5b392d6a0e333506bc49133760b3bcf948733" dependencies = [ "atomic-waker", "bytes", @@ -1648,7 +1596,7 @@ dependencies = [ "bytes", "futures-channel", "futures-core", - "h2 0.4.13", + "h2 0.4.14", "http 1.4.0", "http-body", "httparse", @@ -2145,16 +2093,6 @@ dependencies = [ "serde", ] -[[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 = "is-docker" version = "0.2.0" @@ -2250,9 +2188,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", @@ -2671,15 +2609,14 @@ dependencies = [ [[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 2.11.1", "cfg-if", "foreign-types", "libc", - "once_cell", "openssl-macros", "openssl-sys", ] @@ -2703,9 +2640,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", @@ -2722,6 +2659,58 @@ dependencies = [ "num-traits", ] +[[package]] +name = "other-iota" +version = "0.1.0" +dependencies = [ + "actix-web", + "actix-web-actors", + "aes-gcm", + "async-trait", + "base64", + "chrono", + "crossterm", + "dashmap", + "futures", + "futures-util", + "hex", + "hkdf", + "hyper", + "hyper-util", + "iota-auth", + "iota-logger", + "iota-state", + "iota-storage", + "iota-util", + "json", + "lazy_static", + "once_cell", + "open", + "pnet", + "rand 0.8.6", + "rand_core 0.6.4", + "ratatui", + "reqwest", + "rusqlite", + "rustls", + "rustls-pemfile", + "serde_json", + "sha2 0.10.9", + "strum 0.27.2", + "strum_macros 0.27.2", + "sysinfo", + "tokio", + "tokio-tungstenite", + "ttp-core", + "ttp-native", + "tungstenite", + "uuid", + "walkdir", + "warp", + "x448", + "zip", +] + [[package]] name = "parking_lot" version = "0.12.5" @@ -2874,18 +2863,18 @@ dependencies = [ [[package]] name = "pin-project" -version = "1.1.11" +version = "1.1.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1749c7ed4bcaf4c3d0a3efc28538844fb29bcdd7d2b67b2be7e20ba861ff517" +checksum = "cbf0d9e68100b3a7989b4901972f265cd542e560a3a8a724e1e20322f4d06ce9" dependencies = [ "pin-project-internal", ] [[package]] name = "pin-project-internal" -version = "1.1.11" +version = "1.1.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d9b20ed30f105399776b9c883e68e536ef602a16ae6f596d2c473591d6ad64c6" +checksum = "a990e22f43e84855daf260dded30524ef4a9021cc7541c26540500a50b624389" dependencies = [ "proc-macro2", "quote", @@ -3373,7 +3362,7 @@ dependencies = [ "bytes", "encoding_rs", "futures-core", - "h2 0.4.13", + "h2 0.4.14", "http 1.4.0", "http-body", "http-body-util", @@ -3719,7 +3708,7 @@ checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" dependencies = [ "cfg-if", "cpufeatures 0.3.0", - "digest 0.11.2", + "digest 0.11.3", ] [[package]] @@ -3741,7 +3730,7 @@ checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" dependencies = [ "cfg-if", "cpufeatures 0.3.0", - "digest 0.11.2", + "digest 0.11.3", ] [[package]] @@ -3805,9 +3794,9 @@ checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" [[package]] name = "siphasher" -version = "1.0.2" +version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b2aa850e253778c88a04c3d7323b043aeda9d3e30d5971937c1855769763678e" +checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" [[package]] name = "slab" @@ -4169,9 +4158,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.52.1" +version = "1.52.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6" +checksum = "110a78583f19d5cdb2c5ccf321d1290344e71313c6c37d43520d386027d18386" dependencies = [ "bytes", "libc", @@ -4259,20 +4248,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 2.11.1", "bytes", "futures-util", "http 1.4.0", "http-body", - "iri-string", "pin-project-lite", "tower", "tower-layer", "tower-service", + "url", ] [[package]] @@ -4328,7 +4317,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "ttp-core" version = "0.1.0" -source = "git+https://git.methanium.net/Tensamin/TTP.git#9bd66c762f2de5ebee6ab0e591702b0b992f041e" +source = "git+https://git.methanium.net/Tensamin/TTP.git#db53f44d9323fddceb0eba6ec854c27a3f5e0da7" dependencies = [ "base64", "byteorder", @@ -4341,7 +4330,7 @@ dependencies = [ [[package]] name = "ttp-native" version = "0.1.0" -source = "git+https://git.methanium.net/Tensamin/TTP.git#9bd66c762f2de5ebee6ab0e591702b0b992f041e" +source = "git+https://git.methanium.net/Tensamin/TTP.git#db53f44d9323fddceb0eba6ec854c27a3f5e0da7" dependencies = [ "quinn", "rustls", @@ -4522,9 +4511,9 @@ dependencies = [ [[package]] name = "warp" -version = "0.4.2" +version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "51d06d9202adc1f15d709c4f4a2069be5428aa912cc025d6f268ac441ab066b0" +checksum = "c0a808122a8a77eecdabaefd88ddb1913c4be5ea1465399f63ba64c7aa705fea" dependencies = [ "bytes", "futures-util", @@ -4573,9 +4562,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", @@ -4586,9 +4575,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", @@ -4596,9 +4585,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", @@ -4606,9 +4595,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", @@ -4619,9 +4608,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", ] @@ -4670,9 +4659,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/Cargo.toml b/Cargo.toml index 34fa81d..c62208e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,3 +1,3 @@ [workspace] -members = ["iota-storage", "client", "communities", "iota-auth", "other-iota", "iota-updater", "iota-terms", "iota-state", "iota-cli", "iota-core", "omikron-connector", "web-server", "web-ui", "iota-logger", "iota-util"] +members = ["iota-storage", "client", "iota-auth", "other-iota", "iota-updater", "iota-terms", "iota-state", "iota-cli", "iota-core", "omikron-connector", "web-server", "web-ui", "iota-logger", "iota-util"] resolver = "3" diff --git a/client/src/client_connection.rs b/client/src/client_connection.rs index b13311a..ea238a5 100644 --- a/client/src/client_connection.rs +++ b/client/src/client_connection.rs @@ -1,6 +1,6 @@ use dashmap::DashMap; -use iota_logger::{log, log_cv_in, log_cv_out, log_t}; -use iota_state::{ACTIVE_TASKS, SHUTDOWN}; +use iota_logger::{log_cv_in, log_cv_out, log_t}; +use iota_state::SHUTDOWN; use iota_storage::users::contact::Contact; use iota_storage::util::chat_files::{MessageState, change_message_state}; use iota_storage::util::chats_util::{get_user, mod_user}; @@ -10,127 +10,73 @@ use iota_storage::util::{chat_files, chats_util}; use iota_util::crypto_helper; use iota_util::crypto_util::{DataFormat, SecurePayload}; use iota_util::file_util::{get_children, load_file, save_file}; -use json::JsonValue; -use std::collections::HashMap; -use std::sync::{Arc, LazyLock}; -use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::{Mutex, RwLock, mpsc, watch}; use tokio::task::JoinHandle; -use tokio::time::sleep; use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue}; use ttp_native::{Receiver, Sender}; use uuid::Uuid; -// ============================================================================ -// Configuration -// ============================================================================ - -const CONNECTION_TIMEOUT: Duration = Duration::from_secs(10); -const TASK_CLEANUP_INTERVAL: Duration = Duration::from_secs(60); -const TASK_MAX_AGE: Duration = Duration::from_secs(60); - // ============================================================================ // Waiting Task System // ============================================================================ -pub struct WaitingTask { - pub task: Box, CommunicationValue) -> bool + Send + Sync>, - pub inserted_at: Instant, -} - -// ============================================================================ -// Connection State -// ============================================================================ - -#[derive(Clone, Copy, PartialEq, Eq, Debug)] -pub enum ConnectionState { - Disconnected, - Connecting, - Connected { identified: bool }, -} - -impl ConnectionState { - pub fn is_connected(&self) -> bool { - matches!(self, ConnectionState::Connected { .. }) - } - - pub fn is_identified(&self) -> bool { - matches!(self, ConnectionState::Connected { identified: true }) - } -} - -// ============================================================================ -// Omikron Connection (Client-side with auto-reconnect) -// ============================================================================ - -#[allow(dead_code)] // message_send_times is unused. -pub struct OmikronConnection { - state: Arc>, +#[allow(dead_code)] +pub struct ClientConnection { sender: Arc>>>, + receiver: Receiver, connection_loop_handle: Arc>>>, - host: String, - port: u16, - pub last_ping: Arc>, - heartbeat_handle: Arc>>>, + pub ping: Arc>, pub connection_id: Uuid, shutdown_tx: Arc>>>, - reconnect_on_close: Arc>, - pub app_challenges: Arc>>, - pub app_sessions: Arc>>, + pub waiting_tasks: + DashMap, CommunicationValue) -> bool + Send + Sync>>, } -impl OmikronConnection { - pub fn new() -> Self { - Self::with_host(OMIKRON_HOST_DEFAULT, OMIKRON_PORT_DEFAULT) - } - - pub fn with_host(host: &str, port: u16) -> Self { - let (shutdown_tx, _) = watch::channel(false); - - OmikronConnection { - state: Arc::new(RwLock::new(ConnectionState::Disconnected)), - sender: Arc::new(RwLock::new(None)), - connection_loop_handle: Arc::new(Mutex::new(None)), - host: host.to_string(), - port, - last_ping: Arc::new(Mutex::new(-1)), - heartbeat_handle: Arc::new(Mutex::new(None)), - connection_id: Uuid::new_v4(), - shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))), - reconnect_on_close: Arc::new(RwLock::new(true)), - app_challenges: Arc::new(RwLock::new(HashMap::new())), - app_sessions: Arc::new(RwLock::new(HashMap::new())), +impl ClientConnection { + pub fn new( + sender: Arc>>>, + receiver: Receiver, + connection_loop_handle: Arc>>>, + ping: Arc>, + connection_id: Uuid, + shutdown_tx: Arc>>>, + waiting_tasks: DashMap< + u32, + Box, CommunicationValue) -> bool + Send + Sync>, + >, + ) -> Self { + Self { + sender, + receiver, + connection_loop_handle, + ping, + connection_id, + shutdown_tx, + waiting_tasks, } } - // ------------------------------------------------------------------------- - // Connection Management - // ------------------------------------------------------------------------- - - pub async fn connect(self: &Arc) { - if self.connection_loop_handle.lock().await.is_none() { - self.clone().start().await; - } - } - - pub async fn start(self: Arc) { - if let Some(handle) = self.connection_loop_handle.lock().await.take() { - handle.abort(); - } - - *self.reconnect_on_close.write().await = true; - + pub fn start(self: Arc) { let self_clone = self.clone(); - let handle = tokio::spawn(async move { - self_clone.connection_loop().await; - }); + tokio::spawn(async move { + while let Ok(cv) = self_clone.receiver.receive().await { + if *SHUTDOWN.read().await { + return; + } - *self.connection_loop_handle.lock().await = Some(handle); + self.clone().handle_message(cv).await; + + if !self_clone.receiver.is_open() { + break; + } + } + // Handle Close + }); } pub async fn stop(&self) { - *self.reconnect_on_close.write().await = false; - if let Some(tx) = self.shutdown_tx.lock().await.take() { let _ = tx.send(true); } @@ -139,270 +85,44 @@ impl OmikronConnection { handle.abort(); } - if let Some(handle) = self.heartbeat_handle.lock().await.take() { - handle.abort(); - } - if let Some(sender) = self.sender.read().await.as_ref() { sender.close(); } - *self.state.write().await = ConnectionState::Disconnected; *self.sender.write().await = None; } - async fn connection_loop(self: Arc) { - let mut reconnect_delay = RECONNECT_DELAY; - let shutdown_rx = self.shutdown_tx.lock().await.as_ref().unwrap().subscribe(); - let mut shutdown_rx = shutdown_rx; - - loop { - if *shutdown_rx.borrow() || *SHUTDOWN.read().await { - log_t!("omikron_connection_loop_shutdown"); - break; - } - - if !*self.reconnect_on_close.read().await { - break; - } - - match self.clone().connect_once().await { - Ok(()) => { - if *self.reconnect_on_close.read().await { - log!("Connection lost, reconnecting in {:?}...", reconnect_delay); - } else { - break; - } - } - Err(e) => { - log!( - "Connection failed: {}, retrying in {:?}...", - e, - reconnect_delay - ); - } - } - - tokio::select! { - _ = sleep(reconnect_delay) => {} - _ = shutdown_rx.changed() => { - if *shutdown_rx.borrow() { - break; - } - } - } - - reconnect_delay = std::cmp::min(reconnect_delay * 2, MAX_RECONNECT_DELAY); + // ------------------------------------------------------------------------- + // Message Handling + // ------------------------------------------------------------------------- + async fn handle_ping(self: Arc, cv: CommunicationValue) { + // Update our ping if provided + if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) { + let current = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis(); + let mut ping_guard = self.ping.write().await; + *ping_guard = current as i64 - last_ping; } + + // Send pong response + let response = CommunicationValue::new(CommunicationType::pong) + .with_id(cv.get_id()) + .add_data(DataTypes::ping_iota, DataValue::Number(0)); + + self.send_message(&response).await; } - async fn connect_once(self: Arc) -> Result<(), String> { - *self.state.write().await = ConnectionState::Connecting; - log_t!("omikron_connecting"); - - let addr_str = format!("https://{}:{}/ws/iota/", self.host, self.port); - - let (sender, mut receiver) = ttp_native::client::connect(&addr_str, None) - .await - .map_err(|e| format!("Connection failed: {}", e))?; - - log_t!("omikron_connection_success"); - - let sender_arc = Arc::new(sender); - *self.sender.write().await = Some(sender_arc.clone()); - *self.state.write().await = ConnectionState::Connected { identified: false }; - - // Start read loop - let read_self = self.clone(); - let read_handle = tokio::spawn(async move { - read_self.read_loop(&mut receiver).await; - }); - - // Handle registration/identification - self.handle_authentication().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); - - { - ACTIVE_TASKS.insert("Omikron Listener".to_string()); - } - - // Wait for read loop to complete - let result = read_handle.await; - *self.sender.write().await = None; - *self.state.write().await = ConnectionState::Disconnected; - { - ACTIVE_TASKS.remove("Omikron Listener"); - } - - if let Some(handle) = self.heartbeat_handle.lock().await.take() { - handle.abort(); - } - - match result { - Ok(()) => { - if *self.reconnect_on_close.read().await { - Err("Connection closed, will reconnect".to_string()) - } else { - Ok(()) - } - } - Err(e) => Err(format!("Read loop error: {}", e)), - } - } - - // ------------------------------------------------------------------------- - // Authentication (Registration/Identification) - // ------------------------------------------------------------------------- - - async fn handle_authentication(&self) { - let conf = CONFIG.read().await; - let iota_id = conf.get_iota_id(); - let public_key = conf.get_public_key(); - let private_key = conf.get_private_key(); - drop(conf); - - if iota_id == 0 { - log_t!("iota_register_new"); - - let (pub_k, _priv_k) = if let (Some(pk), Some(sk)) = (public_key, private_key) { - (pk, sk) - } else { - let key_pair = crypto_helper::generate_keypair(); - let public_key_base64 = crypto_helper::public_key_to_base64(&key_pair.public); - let private_key_base64 = crypto_helper::secret_key_to_base64(&key_pair.secret); - - let mut conf_write = CONFIG.write().await; - conf_write.change("public_key", JsonValue::from(public_key_base64.clone())); - conf_write.change("private_key", JsonValue::from(private_key_base64.clone())); - conf_write.update(); - drop(conf_write); - (public_key_base64, private_key_base64) - }; - - let register_msg = CommunicationValue::new(CommunicationType::register_iota) - .add_data(DataTypes::public_key, DataValue::Str(pub_k)); - - let msg_id = register_msg.get_id(); - - WAITING_TASKS.insert( - msg_id, - WaitingTask { - task: Box::new(|selfc, cv| { - if !cv.is_type(CommunicationType::success) { - return false; - } - - let iota_value = cv.get_data(DataTypes::iota_id); - let iota_id = iota_value.as_number().unwrap_or(0); - - if iota_id != 0 { - tokio::spawn(async move { - let mut conf_write = CONFIG.write().await; - conf_write.change("iota_id", JsonValue::from(iota_id)); - conf_write.update(); - drop(conf_write); - log!("Registered with Iota-ID: {}", iota_id); - - // Send identification after registration - let identify_msg = - CommunicationValue::new(CommunicationType::identification) - .add_data(DataTypes::iota_id, DataValue::Number(iota_id)); - selfc.send_message(&identify_msg).await; - }); - } else { - log!("Iota registration failed."); - } - true - }), - inserted_at: Instant::now(), - }, - ); - - self.send_message(®ister_msg).await; - } else { - let identify_msg = CommunicationValue::new(CommunicationType::identification) - .add_data(DataTypes::iota_id, DataValue::Number(iota_id)); - self.send_message(&identify_msg).await; - } - } - - // ------------------------------------------------------------------------- - // Read Loop & Heartbeat - // ------------------------------------------------------------------------- - - async fn read_loop(self: Arc, receiver: &mut Receiver) { - loop { - let result = receiver.receive().await; - match result { - Ok(cv) => { - self.clone().handle_message(cv).await; - } - Err(e) => { - self.fail_all_waiting_tasks(format!( - "Connection receive error: {} (connection_id={})", - e, self.connection_id - )) - .await; - break; - } - } - if !receiver.is_open() { - self.fail_all_waiting_tasks(format!( - "Connection closed (connection_id={}, receiver_open=false)", - self.connection_id - )) - .await; - break; - } - } - } - - async fn heartbeat_loop(self: Arc) { - loop { - sleep(HEARTBEAT_INTERVAL).await; - - if !self.state.read().await.is_connected() { - break; - } - - if let Some(sender) = self.sender.read().await.as_ref() { - if !sender.is_open() { - break; - } - } else { - break; - } - - self.send_ping().await; - } - } - - // ------------------------------------------------------------------------- - // Message Handling (Preserved from original) - // ------------------------------------------------------------------------- - pub async fn handle_message(self: Arc, cv: CommunicationValue) { if !cv.is_type(CommunicationType::ping) && !cv.is_type(CommunicationType::pong) { log_cv_in!(&cv); } - let msg_id = cv.get_id(); + let _msg_id = cv.get_id(); - // Dispatch waiting task for this message id - if let Some((_, task)) = WAITING_TASKS.remove(&msg_id) { - if (task.task)(self.clone(), cv.clone()) { - return; - } - } - - if cv.is_type(CommunicationType::pong) { - self.handle_pong(&cv).await; + if cv.is_type(CommunicationType::ping) { + self.handle_ping(cv).await; return; } @@ -411,115 +131,14 @@ impl OmikronConnection { return; } - if cv.is_type(CommunicationType::app_identification) { - let sender_id = cv.get_sender(); - let app_identifier = cv - .get_data(DataTypes::app_identifier) - .as_str() - .unwrap_or("") - .to_string(); - let app_public_key = cv - .get_data(DataTypes::app_public_key) - .as_str() - .unwrap_or("") - .to_string(); - let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0) as i64; - - let mut trusted = false; - if let Some(user) = iota_storage::users::user_manager::get_user(user_id) { - if let Some(pub_k) = user.trusted_apps.get(&app_identifier) { - if pub_k == &app_public_key { - trusted = true; - } - } - } - - if trusted { - use iota_util::crypto_util::{DataFormat, SecurePayload}; - - let challenge = Uuid::new_v4().to_string(); - - self.app_challenges - .write() - .await - .insert(sender_id, challenge.clone()); - self.app_sessions - .write() - .await - .insert(sender_id, (user_id, app_identifier.clone())); - - if let Some(pub_key) = iota_util::crypto_helper::load_public_key(&app_public_key) { - let conf = CONFIG.read().await; - let priv_k_str = conf.get_private_key().unwrap_or_default(); - let pub_k_str = conf.get_public_key().unwrap_or_default(); - drop(conf); - - if let Some(priv_key) = iota_util::crypto_helper::load_secret_key(&priv_k_str) { - let encrypted_challenge = - SecurePayload::new(challenge.as_bytes(), DataFormat::Raw, priv_key) - .unwrap() - .encrypt_x448(pub_key) - .unwrap() - .export(DataFormat::Base64); - - let res = CommunicationValue::new(CommunicationType::app_challenge) - .with_id(cv.get_id()) - .with_receiver(sender_id) - .add_data(DataTypes::public_key, DataValue::Str(pub_k_str)) - .add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge)); - - self.send_message(&res).await; - return; - } - } - } - - let res = CommunicationValue::new(CommunicationType::error) - .with_id(cv.get_id()) - .with_receiver(sender_id); - self.send_message(&res).await; - return; - } - - if cv.is_type(CommunicationType::app_challenge_response) { - let sender_id = cv.get_sender(); - let mut challenges = self.app_challenges.write().await; - if let Some(expected) = challenges.remove(&sender_id) { - if let DataValue::Str(response) = cv.get_data(DataTypes::challenge) { - if expected == *response { - let res = - CommunicationValue::new(CommunicationType::app_identification_response) - .with_id(cv.get_id()) - .with_receiver(sender_id); - self.send_message(&res).await; - return; - } - } - } - let res = CommunicationValue::new(CommunicationType::error) - .with_id(cv.get_id()) - .with_receiver(sender_id); - self.send_message(&res).await; - return; - } - if cv.is_type(CommunicationType::save_app_data) { let sender_id = cv.get_sender(); - let app_data = cv + let _app_data = cv .get_data(DataTypes::app_data) .as_str() .unwrap_or("") .to_string(); - let sessions = self.app_sessions.read().await; - if let Some((user_id, app_identifier)) = sessions.get(&sender_id) { - iota_storage::users::user_manager::save_app_data( - *user_id, - app_identifier, - &app_data, - ); - } - let res = CommunicationValue::new(CommunicationType::save_app_data) .with_id(cv.get_id()) .with_receiver(sender_id); @@ -529,13 +148,7 @@ impl OmikronConnection { if cv.is_type(CommunicationType::load_app_data) { let sender_id = cv.get_sender(); - let mut app_data = String::new(); - - let sessions = self.app_sessions.read().await; - if let Some((user_id, app_identifier)) = sessions.get(&sender_id) { - app_data = - iota_storage::users::user_manager::load_app_data(*user_id, app_identifier); - } + let app_data = String::new(); let res = CommunicationValue::new(CommunicationType::load_app_data) .with_id(cv.get_id()) @@ -660,16 +273,6 @@ impl OmikronConnection { return; } - if cv.is_type(CommunicationType::identification_response) { - if let Some(_accepted) = cv.get_data(DataTypes::accepted).as_bool() { - let mut state = self.state.write().await; - if let ConnectionState::Connected { identified: _ } = *state { - *state = ConnectionState::Connected { identified: true }; - } - } - return; - } - // ************************************************ // // Direct messages // // ************************************************ // @@ -1334,11 +937,6 @@ impl OmikronConnection { if let Some(sender) = self.sender.write().await.take() { sender.close(); } - self.fail_all_waiting_tasks(format!( - "Send failed: connection closed (connection_id={})", - self.connection_id - )) - .await; return Err("connection closed".to_string()); } @@ -1350,11 +948,6 @@ impl OmikronConnection { } if let Err(e) = sender_clone.send(cv).await { - self.fail_all_waiting_tasks(format!( - "Send failed: {} (connection_id={})", - e, self.connection_id - )) - .await; return Err(e.to_string()); } @@ -1364,133 +957,40 @@ impl OmikronConnection { } } - async fn fail_all_waiting_tasks(&self, reason: String) { - let keys: Vec = WAITING_TASKS.iter().map(|entry| *entry.key()).collect(); - - for key in keys { - if let Some((_, waiting_task)) = WAITING_TASKS.remove(&key) { - let response = CommunicationValue::new(CommunicationType::error) - .with_id(key) - .add_data(DataTypes::message, DataValue::Str(reason.clone())); - let _ = (waiting_task.task)(OMIKRON_CONNECTION.clone(), response); - } - } - } - - pub async fn is_connected(&self) -> bool { - self.state.read().await.is_connected() - } - - pub async fn is_identified(&self) -> bool { - self.state.read().await.is_identified() - } - pub async fn await_response( - &self, + self: Arc, cv: &CommunicationValue, timeout_duration: Option, ) -> Result { let (tx, mut rx) = mpsc::channel(1); let msg_id = cv.get_id(); - WAITING_TASKS.insert( + let task_tx = tx.clone(); + self.waiting_tasks.insert( msg_id, - WaitingTask { - task: Box::new(move |_, response_cv| { - let inner_tx = tx.clone(); - tokio::spawn(async move { - let _ = inner_tx.send(response_cv).await; - }); - true - }), - inserted_at: Instant::now(), - }, + Box::new(move |_, response_cv| { + let inner_tx = task_tx.clone(); + tokio::spawn(async move { + let _ = inner_tx.send(response_cv).await; + }); + true + }), ); - if let Err(send_err) = self.send_message_result(cv).await { - WAITING_TASKS.remove(&msg_id); - return Err(format!( - "Request send failed (msg_id={}, reason={})", - msg_id, send_err - )); - } + self.send_message(cv).await; let timeout = timeout_duration.unwrap_or(Duration::from_secs(10)); match tokio::time::timeout(timeout, rx.recv()).await { - Ok(Some(response_cv)) => { - if response_cv.is_type(CommunicationType::error) { - let reason = response_cv - .get_data(DataTypes::message) - .as_str() - .unwrap_or("connection error") - .to_string(); - Err(format!( - "Request failed due to disconnect (msg_id={}, reason={})", - msg_id, reason - )) - } else { - Ok(response_cv) - } - } - Ok(_) => { - WAITING_TASKS.remove(&msg_id); - Err("Channel closed while awaiting response".to_string()) - } + Ok(Some(response_cv)) => Ok(response_cv), + Ok(_) => Err("Failed to receive response, channel was closed.".to_string()), Err(_) => { - let waiting_tasks_len = WAITING_TASKS.len(); - WAITING_TASKS.remove(&msg_id); + self.waiting_tasks.remove(&msg_id); Err(format!( - "Request timed out (msg_id={}, timeout={}s, connected={}, waiting_tasks={})", - msg_id, - timeout.as_secs(), - self.is_connected().await, - waiting_tasks_len + "Request timed out after {} seconds.", + timeout.as_secs() )) } } } - - pub async fn await_connection(&self, timeout_duration: Option) -> Result<(), String> { - if self.state.read().await.is_connected() { - return Ok(()); - } - - let timeout = timeout_duration.unwrap_or(CONNECTION_TIMEOUT); - let start = Instant::now(); - - loop { - if self.state.read().await.is_connected() { - return Ok(()); - } - - if start.elapsed() >= timeout { - return Err(format!( - "Connection not established within {} seconds", - timeout.as_secs() - )); - } - - sleep(Duration::from_millis(100)).await; - } - } -} - -// ============================================================================ -// Global Instance -// ============================================================================ - -pub static OMIKRON_CONNECTION: LazyLock> = LazyLock::new(|| { - let conn = Arc::new(OmikronConnection::new()); - - start_task_cleanup_loop(); - - conn -}); - -pub async fn get_omikron_connection() -> Arc { - let conn = OMIKRON_CONNECTION.clone(); - - conn.connect().await; - conn } diff --git a/client/src/lib.rs b/client/src/lib.rs index 228eaa9..f3e6b6c 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -1 +1,2 @@ mod client_connection; +pub use client_connection::ClientConnection; diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 11376fc..3698da9 100755 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -18,7 +18,7 @@ use tokio::sync::{Mutex, RwLock, mpsc, watch}; use tokio::task::JoinHandle; use tokio::time::sleep; use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue}; -use ttp_native::{Receiver, Sender}; +use ttp_native::{Policy, Receiver, SendMode, Sender}; use uuid::Uuid; // ============================================================================ @@ -218,9 +218,26 @@ impl OmikronConnection { let addr_str = format!("https://{}:{}/ws/iota/", 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), + 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_t!("omikron_connection_success"); diff --git a/other_iota/Cargo.toml b/other-iota/Cargo.toml similarity index 98% rename from other_iota/Cargo.toml rename to other-iota/Cargo.toml index b930ed5..a1d34f7 100644 --- a/other_iota/Cargo.toml +++ b/other-iota/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "communities" +name = "other-iota" version = "0.1.0" edition = "2024" diff --git a/other_iota/src/lib.rs b/other-iota/src/lib.rs similarity index 100% rename from other_iota/src/lib.rs rename to other-iota/src/lib.rs