From eb266f59f1941908b5ea959e7fe994f233940c3c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Fri, 7 Aug 2020 17:16:10 +0100 Subject: [PATCH] Removed client list from topology (#301) * Removed client list from topology * Formatting --- clients/native/src/client/mod.rs | 23 ++++----------- clients/native/src/client/topology_control.rs | 28 +------------------ clients/native/src/websocket/handler.rs | 18 ------------ clients/native/src/websocket/types.rs | 1 - .../models/src/presence/gateways.rs | 20 ------------- .../src/requests/presence_gateways_post.rs | 1 - common/topology/src/gateway.rs | 10 ------- gateway/src/node/mod.rs | 2 +- gateway/src/node/presence/mod.rs | 16 ++--------- gateway/src/node/storage/ledger.rs | 18 ------------ 10 files changed, 10 insertions(+), 127 deletions(-) diff --git a/clients/native/src/client/mod.rs b/clients/native/src/client/mod.rs index 743bc59b07..aac39271cb 100644 --- a/clients/native/src/client/mod.rs +++ b/clients/native/src/client/mod.rs @@ -256,18 +256,13 @@ impl NymClient { fn start_websocket_listener( &self, - topology_accessor: TopologyAccessor, buffer_requester: ReceivedBufferRequestSender, msg_input: InputMessageSender, ) { info!("Starting websocket listener..."); - let websocket_handler = websocket::Handler::new( - msg_input, - buffer_requester, - self.as_mix_recipient(), - topology_accessor, - ); + let websocket_handler = + websocket::Handler::new(msg_input, buffer_requester, self.as_mix_recipient()); websocket::Listener::new(self.config.get_listening_port()) .start(self.runtime.handle(), websocket_handler); @@ -362,18 +357,12 @@ impl NymClient { sphinx_message_sender.clone(), ); - self.start_cover_traffic_stream( - ack_key, - shared_topology_accessor.clone(), - sphinx_message_sender, - ); + self.start_cover_traffic_stream(ack_key, shared_topology_accessor, sphinx_message_sender); match self.config.get_socket_type() { - SocketType::WebSocket => self.start_websocket_listener( - shared_topology_accessor, - received_buffer_request_sender, - input_sender, - ), + SocketType::WebSocket => { + self.start_websocket_listener(received_buffer_request_sender, input_sender) + } SocketType::None => { // if we did not start the socket, it means we're running (supposedly) in the native mode // and hence we should announce 'ourselves' to the buffer diff --git a/clients/native/src/client/topology_control.rs b/clients/native/src/client/topology_control.rs index f3160fae8e..f1ba0af906 100644 --- a/clients/native/src/client/topology_control.rs +++ b/clients/native/src/client/topology_control.rs @@ -25,7 +25,7 @@ use std::time::Duration; use tokio::runtime::Handle; use tokio::sync::{RwLock, RwLockReadGuard}; use tokio::task::JoinHandle; -use topology::{gateway, NymTopology}; +use topology::NymTopology; // I'm extremely curious why compiler NEVER complained about lack of Debug here before #[derive(Debug)] @@ -118,17 +118,6 @@ impl TopologyAccessor { self.inner.write().await.update(new_topology); } - // pub(crate) async fn get_gateway_socket_url(&self, id: &str) -> Option { - // match &self.inner.read().await.0 { - // None => None, - // Some(ref topology) => topology - // .gateways() - // .iter() - // .find(|gateway| gateway.identity_key == id) - // .map(|gateway| gateway.client_listener.clone()), - // } - // } - // only used by the client at startup to get a slightly more reasonable error message // (currently displays as unused because healthchecker is disabled due to required changes) pub(crate) async fn is_routable(&self) -> bool { @@ -137,21 +126,6 @@ impl TopologyAccessor { Some(ref topology) => topology.can_construct_path_through(DEFAULT_NUM_MIX_HOPS), } } - - pub(crate) async fn get_all_clients(&self) -> Option> { - // TODO: this will need to be modified to instead return pairs (provider, client) - match &self.inner.read().await.0 { - None => None, - Some(ref topology) => Some( - topology - .gateways() - .iter() - .flat_map(|gateway| gateway.registered_clients.iter()) - .cloned() - .collect::>(), - ), - } - } } pub(crate) struct TopologyRefresherConfig { diff --git a/clients/native/src/websocket/handler.rs b/clients/native/src/websocket/handler.rs index 346d85c824..e479bee751 100644 --- a/clients/native/src/websocket/handler.rs +++ b/clients/native/src/websocket/handler.rs @@ -18,7 +18,6 @@ use crate::client::{ received_buffer::{ ReceivedBufferMessage, ReceivedBufferRequestSender, ReconstructedMessagesReceiver, }, - topology_control::TopologyAccessor, }; use futures::channel::mpsc; use futures::{SinkExt, StreamExt}; @@ -47,7 +46,6 @@ pub(crate) struct Handler { msg_input: InputMessageSender, buffer_requester: ReceivedBufferRequestSender, self_full_address: Recipient, - topology_accessor: TopologyAccessor, socket: Option>, received_response_type: ReceivedResponseType, } @@ -59,7 +57,6 @@ impl Clone for Handler { msg_input: self.msg_input.clone(), buffer_requester: self.buffer_requester.clone(), self_full_address: self.self_full_address.clone(), - topology_accessor: self.topology_accessor.clone(), socket: None, received_response_type: Default::default(), } @@ -79,13 +76,11 @@ impl Handler { msg_input: InputMessageSender, buffer_requester: ReceivedBufferRequestSender, self_full_address: Recipient, - topology_accessor: TopologyAccessor, ) -> Self { Handler { msg_input, buffer_requester, self_full_address, - topology_accessor, socket: None, received_response_type: Default::default(), } @@ -114,18 +109,6 @@ impl Handler { None } - async fn handle_text_get_clients(&mut self) -> ServerResponse { - match self.topology_accessor.get_all_clients().await { - Some(clients) => { - let client_keys = clients.into_iter().map(|client| client.pub_key).collect(); - ServerResponse::GetClients { - clients: client_keys, - } - } - None => ServerResponse::new_error("invalid network topology"), - } - } - fn handle_text_self_address(&self) -> ServerResponse { ServerResponse::SelfAddress { address: self.self_full_address.to_string(), @@ -147,7 +130,6 @@ impl Handler { ClientRequest::Send { message, recipient } => self .handle_text_send(message, recipient) .map(|resp| resp.into()), - ClientRequest::GetClients => Some(self.handle_text_get_clients().await.into()), ClientRequest::SelfAddress => Some(self.handle_text_self_address().into()), }, } diff --git a/clients/native/src/websocket/types.rs b/clients/native/src/websocket/types.rs index e1a72cd33d..16c42fd0fd 100644 --- a/clients/native/src/websocket/types.rs +++ b/clients/native/src/websocket/types.rs @@ -21,7 +21,6 @@ use tokio_tungstenite::tungstenite::protocol::Message; #[serde(tag = "type", rename_all = "camelCase")] pub enum ClientRequest { Send { message: String, recipient: String }, - GetClients, SelfAddress, } diff --git a/common/client-libs/directory-client/models/src/presence/gateways.rs b/common/client-libs/directory-client/models/src/presence/gateways.rs index 9d1839e00d..fd9718514e 100644 --- a/common/client-libs/directory-client/models/src/presence/gateways.rs +++ b/common/client-libs/directory-client/models/src/presence/gateways.rs @@ -50,7 +50,6 @@ pub struct GatewayPresence { pub mixnet_listener: String, pub identity_key: String, pub sphinx_key: String, - pub registered_clients: Vec, pub last_seen: u64, pub version: String, } @@ -73,27 +72,8 @@ impl TryInto for GatewayPresence { mixnet_listener: resolved_mix_hostname.unwrap(), identity_key: identity::PublicKey::from_base58_string(self.identity_key)?, sphinx_key: encryption::PublicKey::from_base58_string(self.sphinx_key)?, - registered_clients: self - .registered_clients - .into_iter() - .map(|c| c.into()) - .collect(), last_seen: self.last_seen, version: self.version, }) } } - -#[derive(Clone, Debug, Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct GatewayClient { - pub pub_key: String, -} - -impl Into for GatewayClient { - fn into(self) -> topology::gateway::Client { - topology::gateway::Client { - pub_key: self.pub_key, - } - } -} diff --git a/common/client-libs/directory-client/src/requests/presence_gateways_post.rs b/common/client-libs/directory-client/src/requests/presence_gateways_post.rs index 80ef05c4c1..bd12a83f6d 100644 --- a/common/client-libs/directory-client/src/requests/presence_gateways_post.rs +++ b/common/client-libs/directory-client/src/requests/presence_gateways_post.rs @@ -92,7 +92,6 @@ mod presence_gateways_post_request { mixnet_listener: "foo.com".to_string(), identity_key: "def".to_string(), sphinx_key: "abc".to_string(), - registered_clients: vec![], last_seen: 0, version: "0.1.0".to_string(), } diff --git a/common/topology/src/gateway.rs b/common/topology/src/gateway.rs index c7ea3abfc1..46e02f5e34 100644 --- a/common/topology/src/gateway.rs +++ b/common/topology/src/gateway.rs @@ -31,20 +31,10 @@ pub struct Node { pub mixnet_listener: SocketAddr, pub identity_key: identity::PublicKey, pub sphinx_key: encryption::PublicKey, // TODO: or nymsphinx::PublicKey? both are x25519 - pub registered_clients: Vec, pub last_seen: u64, pub version: String, } -impl Node { - pub fn has_client(&self, client_pub_key: String) -> bool { - self.registered_clients - .iter() - .find(|client| client.pub_key == client_pub_key) - .is_some() - } -} - impl filter::Versioned for Node { fn version(&self) -> String { self.version.clone() diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index 69b0269245..e0a2dc3fa5 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -127,7 +127,7 @@ impl Gateway { self.encryption_keys.public_key().to_base58_string(), self.config.get_presence_sending_delay(), ); - presence::Notifier::new(notifier_config, self.registered_clients_ledger.clone()).start(); + presence::Notifier::new(notifier_config).start(); } async fn wait_for_interrupt(&self) { diff --git a/gateway/src/node/presence/mod.rs b/gateway/src/node/presence/mod.rs index 1173f393c7..7a6e32a453 100644 --- a/gateway/src/node/presence/mod.rs +++ b/gateway/src/node/presence/mod.rs @@ -13,8 +13,7 @@ // limitations under the License. use crate::built_info; -use crate::node::storage::ClientLedger; -use directory_client::presence::gateways::{GatewayClient, GatewayPresence}; +use directory_client::presence::gateways::GatewayPresence; use directory_client::DirectoryClient; use log::{error, trace}; use std::time::Duration; @@ -55,7 +54,6 @@ impl NotifierConfig { pub(crate) struct Notifier { location: String, net_client: directory_client::Client, - client_ledger: ClientLedger, sending_delay: Duration, client_listener: String, mixnet_listener: String, @@ -64,14 +62,13 @@ pub(crate) struct Notifier { } impl Notifier { - pub(crate) fn new(config: NotifierConfig, client_ledger: ClientLedger) -> Notifier { + pub(crate) fn new(config: NotifierConfig) -> Notifier { let directory_client_cfg = directory_client::Config { base_url: config.directory_server, }; let net_client = directory_client::Client::new(directory_client_cfg); Notifier { - client_ledger, net_client, location: config.location, client_listener: config.clients_announce_host, @@ -83,21 +80,12 @@ impl Notifier { } async fn make_presence(&self) -> GatewayPresence { - let client_keys = self.client_ledger.current_clients().unwrap(); - let registered_clients = client_keys - .into_iter() - .map(|key_bytes| GatewayClient { - pub_key: key_bytes.to_base58_string(), - }) - .collect(); - GatewayPresence { location: self.location.clone(), client_listener: self.client_listener.clone(), mixnet_listener: self.mixnet_listener.clone(), identity_key: self.identity.clone(), sphinx_key: self.sphinx_key.clone(), - registered_clients, last_seen: 0, version: built_info::PKG_VERSION.to_string(), } diff --git a/gateway/src/node/storage/ledger.rs b/gateway/src/node/storage/ledger.rs index 9da2d5beff..41cf310a8c 100644 --- a/gateway/src/node/storage/ledger.rs +++ b/gateway/src/node/storage/ledger.rs @@ -155,22 +155,4 @@ impl ClientLedger { self.db.flush().unwrap(); removal_result } - - pub(crate) fn current_clients( - &self, - ) -> Result, ClientLedgerError> { - let clients = self.db.iter().keys(); - - let mut client_vec = Vec::new(); - for client in clients { - match client { - Err(e) => return Err(ClientLedgerError::DbWriteError(e)), - Ok(client_entry) => { - client_vec.push(self.read_destination_address_bytes(client_entry)) - } - } - } - - Ok(client_vec) - } }