GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Tidy up

Browse Source
This commit is contained in:
Jon Häggblad
2023-08-22 10:30:12 +02:00
parent 0b7b10e591
commit 33e69f9eb9
7 changed files with 50 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/client-core/src/client/topology_control/nym_api_provider.rs
-4
View File
@@ -19,7 +19,6 @@ pub(crate) struct NymApiTopologyProvider {
impl NymApiTopologyProvider {
pub(crate) fn new(mut nym_api_urls: Vec<Url>, client_version: String) -> Self {
println!("new");
nym_api_urls.shuffle(&mut thread_rng());
NymApiTopologyProvider {
@@ -33,7 +32,6 @@ impl NymApiTopologyProvider {
}
fn use_next_nym_api(&mut self) {
println!("using next nym api");
if self.nym_api_urls.len() == 1 {
warn!("There's only a single nym API available - it won't be possible to use a different one");
return;
@@ -63,7 +61,6 @@ impl NymApiTopologyProvider {
}
async fn get_current_compatible_topology(&mut self) -> Option<NymTopology> {
println!("getting current compatible topology");
let mixnodes = match self.validator_client.get_cached_active_mixnodes().await {
Err(err) => {
error!("failed to get network mixnodes - {err}");
@@ -112,7 +109,6 @@ impl NymApiTopologyProvider {
#[async_trait]
impl TopologyProvider for NymApiTopologyProvider {
async fn get_new_topology(&mut self) -> Option<NymTopology> {
println!("get_new_topology");
self.get_current_compatible_topology().await
}
}
common/client-libs/gateway-client/src/client.rs
+1 -4
View File
@@ -81,10 +81,7 @@ impl<C, St> fmt::Debug for GatewayClient<C, St> {
.field("shared_key", &self.shared_key)
// .field("connection", &self.connection)
.field("packet_router", &self.packet_router)
.field(
"response_timeout_duration",
&self.response_timeout_duration,
)
.field("response_timeout_duration", &self.response_timeout_duration)
// .field("bandwidth_controller", &self.bandwidth_controller)
.field(
"should_reconnect_on_failure",
gateway/src/node/client_handling/active_clients.rs
+16 -56
View File
@@ -1,15 +1,16 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::node::client_handling::websocket::message_receiver::MixMessageSender;
use dashmap::DashMap;
use nym_sphinx::DestinationAddressBytes;
use futures::channel::{mpsc, oneshot};
use nym_sphinx::DestinationAddressBytes;
use std::sync::Arc;
use super::websocket::message_receiver::{IsActiveRequestSender, MixMessageSender};
#[derive(Clone)]
pub(crate) struct ActiveClientsStore(
Arc<DashMap<DestinationAddressBytes, (MixMessageSender, mpsc::UnboundedSender<oneshot::Sender<bool>>)>>,
Arc<DashMap<DestinationAddressBytes, (MixMessageSender, IsActiveRequestSender)>>,
);
impl ActiveClientsStore {
@@ -24,13 +25,16 @@ impl ActiveClientsStore {
/// # Arguments
///
/// * `client`: address of the client for which to obtain the handle.
pub(crate) fn get(&self, client: DestinationAddressBytes) -> Option<MixMessageSender> {
pub(crate) fn get(
&self,
client: DestinationAddressBytes,
) -> Option<(MixMessageSender, IsActiveRequestSender)> {
let entry = self.0.get(&client)?;
let handle = &entry.value().0;
let handle = entry.value();
// if the entry is stale, remove it from the map
// if handle.is_valid() {
if !handle.is_closed() {
if !handle.0.is_closed() {
Some(handle.clone())
} else {
// drop the reference to the map to prevent deadlocks
@@ -40,54 +44,6 @@ impl ActiveClientsStore {
}
}
// pub(crate) fn get_is_recently_active(&self, client: DestinationAddressBytes) -> Option<RecentlyActive> {
// let entry = self.0.get(&client)?;
// let is_recently_active = &entry.value().1;
//
// // if the entry is stale, remove it from the map
// // if handle.is_valid() {
// if !entry.value().0.is_closed() {
// Some(is_recently_active.clone())
// } else {
// // drop the reference to the map to prevent deadlocks
// drop(entry);
// self.0.remove(&client);
// None
// }
// }
pub(crate) fn get_is_active_sender(&self, client: DestinationAddressBytes) -> Option<mpsc::UnboundedSender<oneshot::Sender<bool>>> {
let entry = self.0.get(&client)?;
let is_active_sender = &entry.value().1;
// if the entry is stale, remove it from the map
// if handle.is_valid() {
if !entry.value().0.is_closed() {
Some(is_active_sender.clone())
} else {
// drop the reference to the map to prevent deadlocks
drop(entry);
self.0.remove(&client);
None
}
}
//
// pub(crate) fn register_activity(&self, client: DestinationAddressBytes) {
// if let Some(mut entry) = self.0.get_mut(&client) {
// entry.value_mut().1 = RecentlyActive::Yes;
// }
// }
// pub(crate) fn reset_activity(&self, client: DestinationAddressBytes) {
// if let Some(mut entry) = self.0.get_mut(&client) {
// if entry.value_mut().1 == RecentlyActive::Yes {
// entry.value_mut().1 = RecentlyActive::YesButReset;
// } else if entry.value_mut().1 == RecentlyActive::YesButReset {
// entry.value_mut().1 = RecentlyActive::No;
// }
// }
// }
/// Indicates particular client has disconnected from the gateway and its handle should get removed.
///
/// # Arguments
@@ -103,7 +59,12 @@ impl ActiveClientsStore {
///
/// * `client`: address of the client for which to insert the handle.
/// * `handle`: the sender channel for all mix packets to be pushed back onto the websocket
pub(crate) fn insert(&self, client: DestinationAddressBytes, handle: MixMessageSender, is_active_sender: mpsc::UnboundedSender<oneshot::Sender<bool>>) {
pub(crate) fn insert(
&self,
client: DestinationAddressBytes,
handle: MixMessageSender,
is_active_sender: mpsc::UnboundedSender<oneshot::Sender<bool>>,
) {
self.0.insert(client, (handle, is_active_sender));
}
@@ -111,5 +72,4 @@ impl ActiveClientsStore {
pub(crate) fn size(&self) -> usize {
self.0.len()
}
}
gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs
+21 -27
View File
@@ -2,10 +2,11 @@
// SPDX-License-Identifier: Apache-2.0
use crate::node::client_handling::websocket::connection_handler::{ClientDetails, FreshHandler};
use crate::node::client_handling::websocket::message_receiver::MixMessageReceiver;
use crate::node::client_handling::websocket::message_receiver::{
IsActiveRequestReceiver, MixMessageReceiver,
};
use crate::node::storage::error::StorageError;
use crate::node::storage::Storage;
use futures::channel::{mpsc, oneshot};
use futures::StreamExt;
use log::*;
use nym_gateway_requests::iv::IVConversionError;
@@ -98,7 +99,7 @@ pub(crate) struct AuthenticatedHandler<R, S, St> {
inner: FreshHandler<R, S, St>,
client: ClientDetails,
mix_receiver: MixMessageReceiver,
is_active_request_receiver: mpsc::UnboundedReceiver<oneshot::Sender<bool>>,
is_active_request_receiver: IsActiveRequestReceiver,
}
// explicitly remove handle from the global store upon being dropped
@@ -129,7 +130,7 @@ where
fresh: FreshHandler<R, S, St>,
client: ClientDetails,
mix_receiver: MixMessageReceiver,
is_active_request_receiver: mpsc::UnboundedReceiver<oneshot::Sender<bool>>,
is_active_request_receiver: IsActiveRequestReceiver,
) -> Self {
AuthenticatedHandler {
inner: fresh,
@@ -139,12 +140,6 @@ where
}
}
// fn register_activity(&self) {
// self.inner
// .active_clients_store
// .register_activity(self.client.address)
// }
/// Explicitly removes handle from the global store.
fn disconnect(self) {
self.inner
@@ -321,7 +316,6 @@ where
///
/// * `bin_msg`: raw message to handle.
async fn handle_binary(&self, bin_msg: Vec<u8>) -> Message {
println!("handle_binary");
// this function decrypts the request and checks the MAC
match BinaryRequest::try_from_encrypted_tagged_bytes(bin_msg, &self.client.shared_keys) {
Err(e) => {
@@ -346,7 +340,6 @@ where
///
/// * `raw_request`: raw message to handle.
async fn handle_text(&mut self, raw_request: String) -> Message {
println!("handle_text");
match ClientControlRequest::try_from(raw_request) {
Err(e) => RequestHandlingError::InvalidTextRequest(e).into_error_message(),
Ok(request) => match request {
@@ -363,6 +356,21 @@ where
}
}
/// Handles pong message received from the client.
/// If the client is still active, the handler that requested the ping will receive a reply.
async fn handle_pong(&mut self, msg: Vec<u8>) {
if let Ok(msg) = msg.try_into() {
let msg = u64::from_be_bytes(msg);
debug!("Received pong from client: {}", msg);
if let Some(tx) = self.inner.is_active_ping_pending_replies.remove(&msg) {
debug!("Reporting back");
if let Err(err) = tx.send(true) {
warn!("Failed to send pong reply back to the requesting handler: {err}");
}
}
}
}
/// Attempts to handle websocket message received from the connected client.
///
/// # Arguments
@@ -377,21 +385,7 @@ where
Message::Binary(bin_msg) => Some(self.handle_binary(bin_msg).await),
Message::Text(text_msg) => Some(self.handle_text(text_msg).await),
Message::Pong(msg) => {
// println!("1 Received pong from client: {}", String::from_utf8_lossy(&s));
// let ss = nym_sphinx::DestinationAddressBytes::try_from_byte_slice(&s).expect("JON");
// let msg = msg.try_into();
// let msg = u64::from_be_bytes(msg.try_into().expect("JON"));
if let Ok(msg) = msg.try_into() {
let msg = u64::from_be_bytes(msg);
// WIP(JON): for testing
if true {
println!("Received pong from client: {}", msg);
if let Some(tx) = self.inner.is_active_ping_pending_replies.remove(&msg) {
println!("Reporting back");
tx.send(true).expect("JON");
}
}
}
self.handle_pong(msg).await;
None
}
_ => None,
gateway/src/node/client_handling/websocket/connection_handler/fresh.rs
+4 -3
View File
@@ -6,6 +6,7 @@ use crate::node::client_handling::websocket::connection_handler::coconut::Coconu
use crate::node::client_handling::websocket::connection_handler::{
AuthenticatedHandler, ClientDetails, InitialAuthResult, SocketStream,
};
use crate::node::client_handling::websocket::message_receiver::IsActiveResultSender;
use crate::node::storage::error::StorageError;
use crate::node::storage::Storage;
use futures::channel::oneshot;
@@ -81,7 +82,7 @@ pub(crate) struct FreshHandler<R, S, St> {
// Occasionally the handler is requested to ping the connected client for confirm that it's
// active, such as when a duplicate connection is detected. This hashmap stores the oneshot
// senders that are used to return the result of the ping to the handler requesting the ping.
pub(crate) is_active_ping_pending_replies: HashMap<u64, oneshot::Sender<bool>>,
pub(crate) is_active_ping_pending_replies: HashMap<u64, IsActiveResultSender>,
}
impl<R, S, St> FreshHandler<R, S, St>
@@ -426,14 +427,14 @@ where
let encrypted_address = EncryptedAddressBytes::try_from_base58_string(enc_address)?;
let iv = IV::try_from_base58_string(iv)?;
if let Some(mut tx) = self.active_clients_store.get_is_active_sender(address) {
if let Some((__, mut is_active_request_tx)) = self.active_clients_store.get(address) {
log::warn!("Detected duplicate connection for client: {}", address);
// Ask the other connection to ping if they are still active.
// Use a oneshot channel to return the result to us
let (ping_result_sender, ping_result_receiver) = oneshot::channel::<bool>();
log::debug!("Asking other connection to ping the connection client");
tx.send(ping_result_sender).await.ok();
is_active_request_tx.send(ping_result_sender).await.ok();
// Wait for the reply
match tokio::time::timeout(Duration::from_millis(1000), ping_result_receiver).await {
gateway/src/node/client_handling/websocket/message_receiver.rs
+7 -1
View File
@@ -1,7 +1,13 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use futures::channel::mpsc;
use futures::channel::{mpsc, oneshot};
pub(crate) type MixMessageSender = mpsc::UnboundedSender<Vec<Vec<u8>>>;
pub(crate) type MixMessageReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
// Channels used for one handler to requester another handler to check that the client is still
// active. The result is then passed back to the requesting handler in the oneshot channel.
pub(crate) type IsActiveRequestSender = mpsc::UnboundedSender<IsActiveResultSender>;
pub(crate) type IsActiveRequestReceiver = mpsc::UnboundedReceiver<IsActiveResultSender>;
pub(crate) type IsActiveResultSender = oneshot::Sender<bool>;
gateway/src/node/mixnet_handling/receiver/connection_handler.rs
+1 -1
View File
@@ -70,7 +70,7 @@ impl<St: Storage> ConnectionHandler<St> {
}
fn update_clients_store_cache_entry(&mut self, client_address: DestinationAddressBytes) {
if let Some(client_sender) = self.active_clients_store.get(client_address) {
if let Some((client_sender, _)) = self.active_clients_store.get(client_address) {
self.clients_store_cache
.insert(client_address, client_sender);
}