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

Removed client list from topology (#301)

Browse Source 
* Removed client list from topology

* Formatting
This commit is contained in:
Jędrzej Stuczyński
2020-08-07 17:16:10 +01:00
committed by GitHub
parent 50805a9b0e
commit eb266f59f1
10 changed files with 10 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/native/src/client/mod.rs
+6 -17
View File
@@ -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
clients/native/src/client/topology_control.rs
+1 -27
View File
@@ -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<String> {
// 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<Vec<gateway::Client>> {
// 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::<Vec<_>>(),
),
}
}
}
pub(crate) struct TopologyRefresherConfig {
clients/native/src/websocket/handler.rs
-18
View File
@@ -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<WebSocketStream<TcpStream>>,
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()),
},
}
clients/native/src/websocket/types.rs
-1
View File
@@ -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,
}
common/client-libs/directory-client/models/src/presence/gateways.rs
-20
View File
@@ -50,7 +50,6 @@ pub struct GatewayPresence {
pub mixnet_listener: String,
pub identity_key: String,
pub sphinx_key: String,
pub registered_clients: Vec<GatewayClient>,
pub last_seen: u64,
pub version: String,
}
@@ -73,27 +72,8 @@ impl TryInto<topology::gateway::Node> 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<topology::gateway::Client> for GatewayClient {
fn into(self) -> topology::gateway::Client {
topology::gateway::Client {
pub_key: self.pub_key,
}
}
}
common/client-libs/directory-client/src/requests/presence_gateways_post.rs
-1
View File
@@ -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(),
}
common/topology/src/gateway.rs
-10
View File
@@ -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<Client>,
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()
gateway/src/node/mod.rs
+1 -1
View File
@@ -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) {
gateway/src/node/presence/mod.rs
+2 -14
View File
@@ -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(),
}
gateway/src/node/storage/ledger.rs
-18
View File
@@ -155,22 +155,4 @@ impl ClientLedger {
self.db.flush().unwrap();
removal_result
}
pub(crate) fn current_clients(
&self,
) -> Result<Vec<DestinationAddressBytes>, 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)
}
}