Compare commits

...

1 Commits

Author SHA1 Message Date
Bogdan-Ștefan Neacşu fb3ec1e02e Connection fd callback before actual connection (#5494) 2025-02-27 10:53:04 +02:00
3 changed files with 50 additions and 12 deletions
@@ -204,15 +204,15 @@ impl<C, St> GatewayClient<C, St> {
"Attemting to establish connection to gateway at: {}", "Attemting to establish connection to gateway at: {}",
self.gateway_address self.gateway_address
); );
let (ws_stream, _) = connect_async(&self.gateway_address).await?; let (ws_stream, _) = connect_async(
&self.gateway_address,
#[cfg(unix)]
self.connection_fd_callback.clone(),
)
.await?;
self.connection = SocketState::Available(Box::new(ws_stream)); 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(()) Ok(())
} }
@@ -1,6 +1,11 @@
use crate::error::GatewayClientError; use crate::error::GatewayClientError;
use nym_http_api_client::HickoryDnsResolver; use nym_http_api_client::HickoryDnsResolver;
#[cfg(unix)]
use std::{
os::fd::{AsRawFd, RawFd},
sync::Arc,
};
use tokio::net::TcpStream; use tokio::net::TcpStream;
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream}; use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
use tungstenite::handshake::client::Response; use tungstenite::handshake::client::Response;
@@ -11,7 +16,10 @@ use std::net::SocketAddr;
#[cfg(not(target_arch = "wasm32"))] #[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn connect_async( pub(crate) async fn connect_async(
endpoint: &str, endpoint: &str,
#[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
) -> Result<(WebSocketStream<MaybeTlsStream<TcpStream>>, Response), GatewayClientError> { ) -> Result<(WebSocketStream<MaybeTlsStream<TcpStream>>, Response), GatewayClientError> {
use tokio::net::TcpSocket;
let resolver = HickoryDnsResolver::default(); let resolver = HickoryDnsResolver::default();
let uri = let uri =
Url::parse(endpoint).map_err(|_| GatewayClientError::InvalidUrl(endpoint.to_owned()))?; Url::parse(endpoint).map_err(|_| GatewayClientError::InvalidUrl(endpoint.to_owned()))?;
@@ -37,14 +45,41 @@ pub(crate) async fn connect_async(
} }
}; };
let stream = TcpStream::connect(&sock_addrs[..]).await.map_err(|error| { let mut stream = Err(GatewayClientError::NoEndpointForConnection {
GatewayClientError::NetworkConnectionFailed { address: endpoint.to_owned(),
address: endpoint.to_owned(), });
source: error.into(), for sock_addr in sock_addrs {
let socket = if sock_addr.is_ipv4() {
TcpSocket::new_v4()
} else {
TcpSocket::new_v6()
} }
})?; .map_err(|err| GatewayClientError::NetworkConnectionFailed {
address: endpoint.to_owned(),
source: err.into(),
})?;
tokio_tungstenite::client_async_tls(endpoint, stream) #[cfg(unix)]
if let Some(callback) = connection_fd_callback.as_ref() {
callback.as_ref()(socket.as_raw_fd());
}
match socket.connect(sock_addr).await {
Ok(s) => {
stream = Ok(s);
break;
}
Err(err) => {
stream = Err(GatewayClientError::NetworkConnectionFailed {
address: endpoint.to_owned(),
source: err.into(),
});
continue;
}
}
}
tokio_tungstenite::client_async_tls(endpoint, stream?)
.await .await
.map_err(|error| GatewayClientError::NetworkConnectionFailed { .map_err(|error| GatewayClientError::NetworkConnectionFailed {
address: endpoint.to_owned(), address: endpoint.to_owned(),
@@ -43,6 +43,9 @@ pub enum GatewayClientError {
#[error("connection failed: {address}: {source}")] #[error("connection failed: {address}: {source}")]
NetworkConnectionFailed { address: String, source: WsError }, NetworkConnectionFailed { address: String, source: WsError },
#[error("no socket address for endpoint: {address}")]
NoEndpointForConnection { address: String },
#[error("Invalid URL: {0}")] #[error("Invalid URL: {0}")]
InvalidUrl(String), InvalidUrl(String),