From 88d6fb4e22a378c27bbe673cbe2819bde0f32958 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bogdan-=C8=98tefan=20Neac=C5=9Fu?= Date: Mon, 16 Dec 2024 13:57:34 +0200 Subject: [PATCH] Add fd callback to client core (#5230) * Add fd callback to client core * Include in sdk * Fix clippy many args * Method in builder * Replace Box with Arc --- .../client-core/src/client/base_client/mod.rs | 23 ++++++++++++++++ .../gateway-client/src/client/mod.rs | 17 ++++++++++++ nym-api/src/network_monitor/monitor/sender.rs | 1 + sdk/rust/nym-sdk/src/mixnet/client.rs | 27 +++++++++++++++++++ 4 files changed, 68 insertions(+) diff --git a/common/client-core/src/client/base_client/mod.rs b/common/client-core/src/client/base_client/mod.rs index 856ddd9497..839df3ec90 100644 --- a/common/client-core/src/client/base_client/mod.rs +++ b/common/client-core/src/client/base_client/mod.rs @@ -188,6 +188,9 @@ pub struct BaseClientBuilder<'a, C, S: MixnetClientStorage> { user_agent: Option, setup_method: GatewaySetup, + + #[cfg(unix)] + connection_fd_callback: Option>, } impl<'a, C, S> BaseClientBuilder<'a, C, S> @@ -210,6 +213,8 @@ where shutdown: None, user_agent: None, setup_method: GatewaySetup::MustLoad { gateway_id: None }, + #[cfg(unix)] + connection_fd_callback: None, } } @@ -261,6 +266,15 @@ where Ok(self) } + #[cfg(unix)] + pub fn with_connection_fd_callback( + mut self, + callback: Arc, + ) -> Self { + self.connection_fd_callback = Some(callback); + self + } + // note: do **NOT** make this method public as its only valid usage is from within `start_base` // because it relies on the crypto keys being already loaded fn mix_address(details: &InitialisationResult) -> Recipient { @@ -352,6 +366,7 @@ where controller.start_with_shutdown(shutdown) } + #[allow(clippy::too_many_arguments)] async fn start_gateway_client( config: &Config, initialisation_result: InitialisationResult, @@ -359,6 +374,7 @@ where details_store: &S::GatewaysDetailsStore, packet_router: PacketRouter, stats_reporter: ClientStatsSender, + #[cfg(unix)] connection_fd_callback: Option>, shutdown: TaskClient, ) -> Result, ClientCoreError> where @@ -401,6 +417,8 @@ where packet_router, bandwidth_controller, stats_reporter, + #[cfg(unix)] + connection_fd_callback, shutdown, ) }; @@ -462,6 +480,7 @@ where details_store: &S::GatewaysDetailsStore, packet_router: PacketRouter, stats_reporter: ClientStatsSender, + #[cfg(unix)] connection_fd_callback: Option>, mut shutdown: TaskClient, ) -> Result, ClientCoreError> where @@ -493,6 +512,8 @@ where details_store, packet_router, stats_reporter, + #[cfg(unix)] + connection_fd_callback, shutdown, ) .await?; @@ -772,6 +793,8 @@ where &details_store, gateway_packet_router, stats_reporter.clone(), + #[cfg(unix)] + self.connection_fd_callback, shutdown.fork("gateway_transceiver"), ) .await?; diff --git a/common/client-libs/gateway-client/src/client/mod.rs b/common/client-libs/gateway-client/src/client/mod.rs index 6cb7b83f02..481e57dd6f 100644 --- a/common/client-libs/gateway-client/src/client/mod.rs +++ b/common/client-libs/gateway-client/src/client/mod.rs @@ -101,6 +101,10 @@ pub struct GatewayClient { // currently unused (but populated) negotiated_protocol: Option, + // Callback on the fd as soon as the connection has been established + #[cfg(unix)] + connection_fd_callback: Option>, + /// Listen to shutdown messages and send notifications back to the task manager task_client: TaskClient, } @@ -116,6 +120,7 @@ impl GatewayClient { packet_router: PacketRouter, bandwidth_controller: Option>, stats_reporter: ClientStatsSender, + #[cfg(unix)] connection_fd_callback: Option>, task_client: TaskClient, ) -> Self { GatewayClient { @@ -131,6 +136,8 @@ impl GatewayClient { bandwidth_controller, stats_reporter, negotiated_protocol: None, + #[cfg(unix)] + connection_fd_callback, task_client, } } @@ -205,6 +212,12 @@ impl GatewayClient { }; self.connection = SocketState::Available(Box::new(ws_stream)); + + #[cfg(unix)] + if let (Some(callback), Some(fd)) = (self.connection_fd_callback.as_ref(), self.ws_fd()) { + callback.as_ref()(fd); + } + Ok(()) } @@ -1034,6 +1047,8 @@ impl GatewayClient { bandwidth_controller: None, stats_reporter: ClientStatsSender::new(None), negotiated_protocol: None, + #[cfg(unix)] + connection_fd_callback: None, task_client, } } @@ -1064,6 +1079,8 @@ impl GatewayClient { bandwidth_controller, stats_reporter, negotiated_protocol: self.negotiated_protocol, + #[cfg(unix)] + connection_fd_callback: self.connection_fd_callback, task_client, } } diff --git a/nym-api/src/network_monitor/monitor/sender.rs b/nym-api/src/network_monitor/monitor/sender.rs index c1140a55ac..172d6b3680 100644 --- a/nym-api/src/network_monitor/monitor/sender.rs +++ b/nym-api/src/network_monitor/monitor/sender.rs @@ -183,6 +183,7 @@ impl PacketSender { gateway_packet_router, Some(fresh_gateway_client_data.bandwidth_controller.clone()), nym_statistics_common::clients::ClientStatsSender::new(None), + None, task_client, ); diff --git a/sdk/rust/nym-sdk/src/mixnet/client.rs b/sdk/rust/nym-sdk/src/mixnet/client.rs index 118885a650..f223037572 100644 --- a/sdk/rust/nym-sdk/src/mixnet/client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/client.rs @@ -36,6 +36,7 @@ use nym_validator_client::{nyxd, QueryHttpRpcNyxdClient, UserAgent}; use rand::rngs::OsRng; use std::path::Path; use std::path::PathBuf; +use std::sync::Arc; use url::Url; use zeroize::Zeroizing; @@ -54,6 +55,7 @@ pub struct MixnetClientBuilder { custom_shutdown: Option, force_tls: bool, user_agent: Option, + connection_fd_callback: Option>, // TODO: incorporate it properly into `MixnetClientStorage` (I will need it in wasm anyway) gateway_endpoint_config_path: Option, @@ -93,6 +95,8 @@ impl MixnetClientBuilder { custom_gateway_transceiver: None, force_tls: false, user_agent: None, + #[cfg(unix)] + connection_fd_callback: None, }) } } @@ -120,6 +124,8 @@ where custom_shutdown: None, force_tls: false, user_agent: None, + #[cfg(unix)] + connection_fd_callback: None, gateway_endpoint_config_path: None, storage, } @@ -138,6 +144,8 @@ where custom_shutdown: self.custom_shutdown, force_tls: self.force_tls, user_agent: self.user_agent, + #[cfg(unix)] + connection_fd_callback: self.connection_fd_callback, gateway_endpoint_config_path: self.gateway_endpoint_config_path, storage, } @@ -237,6 +245,15 @@ where self } + #[must_use] + pub fn with_connection_fd_callback( + mut self, + connection_fd_callback: Arc, + ) -> Self { + self.connection_fd_callback = Some(connection_fd_callback); + self + } + /// Use custom mixnet sender that might not be the default websocket gateway connection. /// only for advanced use #[must_use] @@ -265,6 +282,7 @@ where client.wait_for_gateway = self.wait_for_gateway; client.force_tls = self.force_tls; client.user_agent = self.user_agent; + client.connection_fd_callback = self.connection_fd_callback; Ok(client) } @@ -314,6 +332,9 @@ where custom_shutdown: Option, user_agent: Option, + + /// Callback on the websocket fd as soon as the connection has been established + connection_fd_callback: Option>, } impl DisconnectedMixnetClient @@ -363,6 +384,7 @@ where force_tls: false, custom_shutdown: None, user_agent: None, + connection_fd_callback: None, }) } @@ -604,6 +626,11 @@ where base_builder = base_builder.with_gateway_transceiver(gateway_transceiver); } + #[cfg(unix)] + if let Some(connection_fd_callback) = self.connection_fd_callback { + base_builder = base_builder.with_connection_fd_callback(connection_fd_callback); + } + let started_client = base_builder.start_base().await?; self.state = BuilderState::Registered {}; let nym_address = started_client.address;