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

common/client-core/src/client/base_client/mod.rs

@@ -188,6 +188,9 @@ pub struct BaseClientBuilder<'a, C, S: MixnetClientStorage> {
     user_agent: Option<UserAgent>,
 
     setup_method: GatewaySetup,
+
+    #[cfg(unix)]
+    connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
 }
 
 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<dyn Fn(RawFd) + Send + Sync>,
+    ) -> 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<Arc<dyn Fn(RawFd) + Send + Sync>>,
         shutdown: TaskClient,
     ) -> Result<GatewayClient<C, S::CredentialStore>, 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<Arc<dyn Fn(RawFd) + Send + Sync>>,
         mut shutdown: TaskClient,
     ) -> Result<Box<dyn GatewayTransceiver + Send>, 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?;

common/client-libs/gateway-client/src/client/mod.rs

@@ -101,6 +101,10 @@ pub struct GatewayClient<C, St = EphemeralCredentialStorage> {
     // currently unused (but populated)
     negotiated_protocol: Option<u8>,
 
+    // Callback on the fd as soon as the connection has been established
+    #[cfg(unix)]
+    connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
+
     /// Listen to shutdown messages and send notifications back to the task manager
     task_client: TaskClient,
 }
@@ -116,6 +120,7 @@ impl<C, St> GatewayClient<C, St> {
         packet_router: PacketRouter,
         bandwidth_controller: Option<BandwidthController<C, St>>,
         stats_reporter: ClientStatsSender,
+        #[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
         task_client: TaskClient,
     ) -> Self {
         GatewayClient {
@@ -131,6 +136,8 @@ impl<C, St> GatewayClient<C, St> {
             bandwidth_controller,
             stats_reporter,
             negotiated_protocol: None,
+            #[cfg(unix)]
+            connection_fd_callback,
             task_client,
         }
     }
@@ -205,6 +212,12 @@ impl<C, St> GatewayClient<C, St> {
         };
 
         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<InitOnly, EphemeralCredentialStorage> {
             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<InitOnly, EphemeralCredentialStorage> {
             bandwidth_controller,
             stats_reporter,
             negotiated_protocol: self.negotiated_protocol,
+            #[cfg(unix)]
+            connection_fd_callback: self.connection_fd_callback,
             task_client,
         }
     }

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,
         );
 

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<S: MixnetClientStorage = Ephemeral> {
     custom_shutdown: Option<TaskClient>,
     force_tls: bool,
     user_agent: Option<UserAgent>,
+    connection_fd_callback: Option<Arc<dyn Fn(std::os::fd::RawFd) + Send + Sync>>,
 
     // TODO: incorporate it properly into `MixnetClientStorage` (I will need it in wasm anyway)
     gateway_endpoint_config_path: Option<PathBuf>,
@@ -93,6 +95,8 @@ impl MixnetClientBuilder<OnDiskPersistent> {
             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<dyn Fn(std::os::fd::RawFd) + Send + Sync>,
+    ) -> 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<TaskClient>,
 
     user_agent: Option<UserAgent>,
+
+    /// Callback on the websocket fd as soon as the connection has been established
+    connection_fd_callback: Option<Arc<dyn Fn(std::os::fd::RawFd) + Send + Sync>>,
 }
 
 impl<S> DisconnectedMixnetClient<S>
@@ -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;