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

Merge pull request #94 from nymtech/feature/refreshing_topology

Browse Source 
Feature/refreshing topology
This commit is contained in:
Dave Hrycyszyn
2020-01-24 12:13:12 +00:00
committed by GitHub
parent dea363e813 143d45904c
commit 8510924ebb
9 changed files with 440 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/clients/directory-client/src/presence.rs
+144
View File
@@ -2,11 +2,18 @@ use crate::requests::presence_topology_get::PresenceTopologyGetRequester;
use crate::{Client, Config, DirectoryClient};
use log::*;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::convert::TryInto;
use std::io;
use std::net::ToSocketAddrs;
use topology::{CocoNode, MixNode, MixProviderNode, NymTopology};
// special version of 'PartialEq' that does not care about last_seen field (where applicable)
// or order of elements in vectors
trait PresenceEq {
fn presence_eq(&self, other: &Self) -> bool;
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CocoPresence {
@@ -93,6 +100,32 @@ pub struct MixProviderPresence {
pub version: String,
}
impl PresenceEq for MixProviderPresence {
fn presence_eq(&self, other: &Self) -> bool {
if self.registered_clients.len() != other.registered_clients.len() {
return false;
}
if self.client_listener != other.client_listener
|| self.mixnet_listener != other.mixnet_listener
|| self.pub_key != other.pub_key
|| self.version != other.version
{
return false;
}
let clients_self_set: HashSet<_> =
self.registered_clients.iter().map(|c| &c.pub_key).collect();
let clients_other_set: HashSet<_> = other
.registered_clients
.iter()
.map(|c| &c.pub_key)
.collect();
clients_self_set == clients_other_set
}
}
impl Into<topology::MixProviderNode> for MixProviderPresence {
fn into(self) -> topology::MixProviderNode {
topology::MixProviderNode {
@@ -149,6 +182,100 @@ impl From<topology::MixProviderClient> for MixProviderClient {
}
}
impl PresenceEq for Vec<CocoPresence> {
fn presence_eq(&self, other: &Self) -> bool {
if self.len() != other.len() {
return false;
}
// we can't take the whole thing into set as it does not implement 'Eq' and we can't
// derive it as we don't want to take 'last_seen' into consideration
let self_set: HashSet<_> = self
.iter()
.map(|c| (&c.host, &c.pub_key, &c.version))
.collect();
let other_set: HashSet<_> = other
.iter()
.map(|c| (&c.host, &c.pub_key, &c.version))
.collect();
self_set == other_set
}
}
impl PresenceEq for Vec<MixNodePresence> {
fn presence_eq(&self, other: &Self) -> bool {
if self.len() != other.len() {
return false;
}
// we can't take the whole thing into set as it does not implement 'Eq' and we can't
// derive it as we don't want to take 'last_seen' into consideration
let self_set: HashSet<_> = self
.iter()
.map(|m| (&m.host, &m.pub_key, &m.version, &m.layer))
.collect();
let other_set: HashSet<_> = other
.iter()
.map(|m| (&m.host, &m.pub_key, &m.version, &m.layer))
.collect();
self_set == other_set
}
}
impl PresenceEq for Vec<MixProviderPresence> {
fn presence_eq(&self, other: &Self) -> bool {
if self.len() != other.len() {
return false;
}
// we can't take the whole thing into set as it does not implement 'Eq' and we can't
// derive it as we don't want to take 'last_seen' into consideration.
// We also don't care about order of registered_clients
// since we're going to be getting rid of this very soon anyway, just clone registered
// clients vector and sort it
let self_set: HashSet<_> = self
.iter()
.map(|p| {
(
&p.client_listener,
&p.mixnet_listener,
&p.pub_key,
&p.version,
p.registered_clients
.iter()
.cloned()
.map(|c| c.pub_key)
.collect::<Vec<_>>()
.sort(),
)
})
.collect();
let other_set: HashSet<_> = other
.iter()
.map(|p| {
(
&p.client_listener,
&p.mixnet_listener,
&p.pub_key,
&p.version,
p.registered_clients
.iter()
.cloned()
.map(|c| c.pub_key)
.collect::<Vec<_>>()
.sort(),
)
})
.collect();
self_set == other_set
}
}
// Topology shows us the current state of the overall Nym network
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
@@ -158,6 +285,23 @@ pub struct Topology {
pub mix_provider_nodes: Vec<MixProviderPresence>,
}
impl PartialEq for Topology {
// we need a custom implementation as the order of nodes does not matter
// also we do not care about 'last_seen' when comparing topologies
fn eq(&self, other: &Self) -> bool {
if !self.coco_nodes.presence_eq(&other.coco_nodes)
|| !self.mix_nodes.presence_eq(&other.mix_nodes)
|| !self
.mix_provider_nodes
.presence_eq(&other.mix_provider_nodes)
{
return false;
}
true
}
}
impl NymTopology for Topology {
fn new(directory_server: String) -> Self {
debug!("Using directory server: {:?}", directory_server);
common/healthcheck/src/path_check.rs
+2 -2
View File
@@ -1,6 +1,6 @@
use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair, MixnetIdentityPublicKey};
use itertools::Itertools;
use log::{debug, error, trace, warn};
use log::{debug, error, info, trace, warn};
use mix_client::MixClient;
use provider_client::ProviderClient;
use sphinx::header::delays::Delay;
@@ -223,7 +223,7 @@ impl PathChecker {
debug!("sending test packet to {}", first_node_address);
match first_node_client.send(packet, first_node_address).await {
Err(err) => {
warn!("failed to send packet to {} - {}", first_node_address, err);
info!("failed to send packet to {} - {}", first_node_address, err);
if self
.paths_status
.insert(path_identifier, PathStatus::Unhealthy)
common/topology/src/lib.rs
+1 -1
View File
@@ -84,7 +84,7 @@ pub enum NymTopologyError {
MissingLayerError(Vec<u64>),
}
pub trait NymTopology: Sized {
pub trait NymTopology: Sized + PartialEq + std::fmt::Debug + Send + Sync {
fn new(directory_server: String) -> Self;
fn new_from_nodes(
mix_nodes: Vec<MixNode>,
nym-client/src/client/cover_traffic_stream.rs
+17 -11
View File
@@ -1,29 +1,35 @@
use crate::client::mix_traffic::MixMessage;
use crate::client::topology_control::TopologyInnerRef;
use crate::client::LOOP_COVER_AVERAGE_DELAY;
use futures::channel::mpsc;
use log::{info, trace};
use log::{info, trace, warn};
use sphinx::route::Destination;
use std::time::Duration;
use topology::NymTopology;
pub(crate) async fn start_loop_cover_traffic_stream<T>(
pub(crate) async fn start_loop_cover_traffic_stream<T: NymTopology>(
tx: mpsc::UnboundedSender<MixMessage>,
our_info: Destination,
topology: T,
) where
T: NymTopology,
{
topology_ctrl_ref: TopologyInnerRef<T>,
) {
info!("Starting loop cover traffic stream");
loop {
trace!("next cover message!");
let delay = mix_client::poisson::sample(LOOP_COVER_AVERAGE_DELAY);
let delay_duration = Duration::from_secs_f64(delay);
tokio::time::delay_for(delay_duration).await;
let cover_message = mix_client::packet::loop_cover_message(
our_info.address,
our_info.identifier,
&topology,
);
let read_lock = topology_ctrl_ref.read().await;
let topology = read_lock.topology.as_ref();
if topology.is_none() {
warn!("No valid topology detected - won't send any loop cover message this time");
continue;
}
let topology = topology.unwrap();
let cover_message =
mix_client::packet::loop_cover_message(our_info.address, our_info.identifier, topology);
// if this one fails, there's no retrying because it means that either:
// - we run out of memory
nym-client/src/client/mod.rs
+32 -70
View File
@@ -1,6 +1,6 @@
use crate::built_info;
use crate::client::mix_traffic::MixTrafficController;
use crate::client::received_buffer::ReceivedMessagesBuffer;
use crate::client::topology_control::TopologyInnerRef;
use crate::sockets::tcp;
use crate::sockets::ws;
use directory_client::presence::Topology;
@@ -18,6 +18,7 @@ mod mix_traffic;
mod provider_poller;
mod real_traffic_stream;
pub mod received_buffer;
pub mod topology_control;
// TODO: all of those constants should probably be moved to config file
const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5;
@@ -26,6 +27,8 @@ const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 0.5;
// seconds;
const FETCH_MESSAGES_DELAY: f64 = 1.0; // seconds;
const TOPOLOGY_REFRESH_RATE: f64 = 10.0; // seconds
pub enum SocketType {
TCP,
WebSocket,
@@ -46,13 +49,6 @@ pub struct NymClient {
socket_type: SocketType,
}
// TODO: this will be moved into module responsible for refreshing topology
#[derive(Debug)]
enum TopologyError {
HealthCheckError,
NoValidPathsError,
}
#[derive(Debug)]
pub struct InputMessage(pub Destination, pub Vec<u8>);
@@ -77,50 +73,16 @@ impl NymClient {
}
}
// TODO: this will be moved into module responsible for refreshing topology
async fn get_compatible_topology(&self) -> Result<Topology, TopologyError> {
let score_threshold = 0.0;
info!("Trying to obtain valid, healthy, topology");
let full_topology = Topology::new(self.directory.clone());
// run a healthcheck to determine healthy-ish nodes:
// this is a temporary solution as the healthcheck will eventually be moved to validators
let healthcheck_config = healthcheck::config::HealthCheck {
directory_server: self.directory.clone(),
// those are literally unrelevant when running single check
interval: 100000.0,
resolution_timeout: 5.0,
num_test_packets: 2,
};
let healthcheck = healthcheck::HealthChecker::new(healthcheck_config);
let healthcheck_result = healthcheck.do_check().await;
let healthcheck_scores = match healthcheck_result {
Err(err) => {
error!("Error while performing the healtcheck: {:?}", err);
return Err(TopologyError::HealthCheckError);
}
Ok(scores) => scores,
};
let healthy_topology =
healthcheck_scores.filter_topology_by_score(&full_topology, score_threshold);
// for time being assume same versioning, i.e. if client is running X.Y.Z,
// we're expecting mixes, providers and coconodes to also be running X.Y.Z
let versioned_healthy_topology = healthy_topology.filter_node_versions(
built_info::PKG_VERSION,
built_info::PKG_VERSION,
built_info::PKG_VERSION,
);
// make sure you can still send a packet through the network:
if !versioned_healthy_topology.can_construct_path_through() {
return Err(TopologyError::NoValidPathsError);
}
Ok(versioned_healthy_topology)
async fn get_provider_socket_address<T: NymTopology>(
&self,
topology_ctrl_ref: TopologyInnerRef<T>,
) -> SocketAddr {
// this is temporary and assumes there exists only a single provider.
topology_ctrl_ref.read().await.topology.as_ref().unwrap()
.get_mix_provider_nodes()
.first()
.expect("Could not get a provider from the initial network topology, are you using the right directory server?")
.client_listener
}
pub fn start(self) -> Result<(), Box<dyn std::error::Error>> {
@@ -144,20 +106,14 @@ impl NymClient {
let (received_messages_buffer_output_tx, received_messages_buffer_output_rx) =
mpsc::unbounded();
// get initial topology; already filtered by health and version
let initial_topology = match rt.block_on(self.get_compatible_topology()) {
Ok(topology) => topology,
Err(err) => {
panic!("Failed to obtain initial network topology: {:?}", err);
}
};
// TODO: when we switch to our graph topology, we need to remember to change 'Topology' type
let topology_controller = rt.block_on(topology_control::TopologyControl::<Topology>::new(
self.directory.clone(),
TOPOLOGY_REFRESH_RATE,
));
// this is temporary and assumes there exists only a single provider.
let provider_client_listener_address: SocketAddr = initial_topology
.get_mix_provider_nodes()
.first()
.expect("Could not get a provider from the supplied network topology, are you using the right directory server?")
.client_listener;
let provider_client_listener_address =
rt.block_on(self.get_provider_socket_address(topology_controller.get_inner_ref()));
let mut provider_poller = provider_poller::ProviderPoller::new(
poller_input_tx,
@@ -189,13 +145,13 @@ impl NymClient {
rt.spawn(cover_traffic_stream::start_loop_cover_traffic_stream(
mix_tx.clone(),
Destination::new(self.address, Default::default()),
initial_topology.clone(),
topology_controller.get_inner_ref(),
));
// cloning arguments required by OutQueueControl; required due to move
let topology_clone = initial_topology.clone();
let self_address = self.address;
let input_rx = self.input_rx;
let topology_ref = topology_controller.get_inner_ref();
// future constantly pumping traffic at some specified average rate
// if a real message is available on 'input_rx' that might have been received from say
@@ -206,7 +162,7 @@ impl NymClient {
mix_tx,
input_rx,
Destination::new(self_address, Default::default()),
topology_clone,
topology_ref,
)
.run_out_queue_control()
.await
@@ -225,7 +181,7 @@ impl NymClient {
self.input_tx,
received_messages_buffer_output_tx,
self.address,
initial_topology,
topology_controller.get_inner_ref(),
));
}
SocketType::TCP => {
@@ -234,12 +190,16 @@ impl NymClient {
self.input_tx,
received_messages_buffer_output_tx,
self.address,
initial_topology,
topology_controller.get_inner_ref(),
));
}
SocketType::None => (),
}
// future responsible for periodically polling directory server and updating
// the current global view of topology
let topology_refresher_future = rt.spawn(topology_controller.run_refresher());
rt.block_on(async {
let future_results = join!(
received_messages_buffer_controllers_future,
@@ -247,6 +207,7 @@ impl NymClient {
loop_cover_traffic_future,
out_queue_control_future,
provider_polling_future,
topology_refresher_future,
);
assert!(
@@ -255,6 +216,7 @@ impl NymClient {
&& future_results.2.is_ok()
&& future_results.3.is_ok()
&& future_results.4.is_ok()
&& future_results.5.is_ok()
);
});
nym-client/src/client/real_traffic_stream.rs
+45 -36
View File
@@ -1,38 +1,38 @@
use crate::client::mix_traffic::MixMessage;
use crate::client::topology_control::TopologyInnerRef;
use crate::client::{InputMessage, MESSAGE_SENDING_AVERAGE_DELAY};
use directory_client::presence::Topology;
use futures::channel::mpsc;
use futures::task::{Context, Poll};
use futures::{Future, Stream, StreamExt};
use log::{debug, info, trace};
use log::{info, trace, warn};
use sphinx::route::Destination;
use sphinx::SphinxPacket;
use std::net::SocketAddr;
use std::pin::Pin;
use std::time::Duration;
use tokio::time;
use topology::NymTopology;
// have a rather low value for test sake
const AVERAGE_PACKET_DELAY: f64 = 0.1;
pub(crate) struct OutQueueControl {
pub(crate) struct OutQueueControl<T: NymTopology> {
delay: time::Delay,
mix_tx: mpsc::UnboundedSender<MixMessage>,
input_rx: mpsc::UnboundedReceiver<InputMessage>,
our_info: Destination,
// due to pinning, DerefMut trait, futures, etc its way easier to
// just have concrete implementation here rather than generic NymTopology
// considering that it will be replaced with refreshing topology within few days anyway
topology: Topology,
topology_ctrl_ref: TopologyInnerRef<T>,
}
impl Stream for OutQueueControl {
type Item = (SocketAddr, SphinxPacket);
pub(crate) enum StreamMessage {
Cover,
Real(InputMessage),
}
impl<T: NymTopology> Stream for OutQueueControl<T> {
type Item = StreamMessage;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
// it is not yet time to return a message
if Future::poll(Pin::new(&mut self.delay), cx).is_pending() {
if Pin::new(&mut self.delay).poll(cx).is_pending() {
return Poll::Pending;
};
@@ -49,41 +49,26 @@ impl Stream for OutQueueControl {
self.delay.reset(next);
// decide what kind of message to send
match Stream::poll_next(Pin::new(&mut self.input_rx), cx) {
match Pin::new(&mut self.input_rx).poll_next(cx) {
// in the case our real message channel stream was closed, we should also indicate we are closed
// (and whoever is using the stream should panic)
Poll::Ready(None) => Poll::Ready(None),
// if there's an actual message - return it
Poll::Ready(Some(real_message)) => {
trace!("real message");
Poll::Ready(Some(mix_client::packet::encapsulate_message(
real_message.0,
real_message.1,
&self.topology,
AVERAGE_PACKET_DELAY,
)))
}
Poll::Ready(Some(real_message)) => Poll::Ready(Some(StreamMessage::Real(real_message))),
// otherwise construct a dummy one
_ => {
trace!("loop cover message");
Poll::Ready(Some(mix_client::packet::loop_cover_message(
self.our_info.address,
self.our_info.identifier,
&self.topology,
)))
}
Poll::Pending => Poll::Ready(Some(StreamMessage::Cover)),
}
}
}
impl OutQueueControl {
impl<T: NymTopology> OutQueueControl<T> {
pub(crate) fn new(
mix_tx: mpsc::UnboundedSender<MixMessage>,
input_rx: mpsc::UnboundedReceiver<InputMessage>,
our_info: Destination,
topology: Topology,
topology: TopologyInnerRef<T>,
) -> Self {
let initial_delay = time::delay_for(Duration::from_secs_f64(MESSAGE_SENDING_AVERAGE_DELAY));
OutQueueControl {
@@ -91,20 +76,44 @@ impl OutQueueControl {
mix_tx,
input_rx,
our_info,
topology,
topology_ctrl_ref: topology,
}
}
pub(crate) async fn run_out_queue_control(mut self) {
info!("starting out queue controller");
while let Some(next_message) = self.next().await {
debug!("created new message");
trace!("created new message");
let read_lock = self.topology_ctrl_ref.read().await;
let topology = read_lock.topology.as_ref();
if topology.is_none() {
warn!("No valid topology detected - won't send any loop cover or real message this time");
continue;
}
let topology = topology.unwrap();
let next_packet = match next_message {
StreamMessage::Cover => mix_client::packet::loop_cover_message(
self.our_info.address,
self.our_info.identifier,
topology,
),
StreamMessage::Real(real_message) => mix_client::packet::encapsulate_message(
real_message.0,
real_message.1,
topology,
AVERAGE_PACKET_DELAY,
),
};
// if this one fails, there's no retrying because it means that either:
// - we run out of memory
// - the receiver channel is closed
// in either case there's no recovery and we can only panic
self.mix_tx
.unbounded_send(MixMessage::new(next_message.0, next_message.1))
.unbounded_send(MixMessage::new(next_packet.0, next_packet.1))
.unwrap();
}
}
nym-client/src/client/topology_control.rs
+141
View File
@@ -0,0 +1,141 @@
use crate::built_info;
use log::{error, info, trace, warn};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock as FRwLock;
use topology::NymTopology;
const NODE_HEALTH_THRESHOLD: f64 = 0.0;
// auxiliary type for ease of use
pub type TopologyInnerRef<T> = Arc<FRwLock<Inner<T>>>;
pub(crate) struct TopologyControl<T: NymTopology> {
directory_server: String,
refresh_rate: f64,
inner: Arc<FRwLock<Inner<T>>>,
}
#[derive(Debug)]
enum TopologyError {
HealthCheckError,
NoValidPathsError,
}
impl<T: NymTopology> TopologyControl<T> {
pub(crate) async fn new(directory_server: String, refresh_rate: f64) -> Self {
let initial_topology = match Self::get_current_compatible_topology(directory_server.clone())
.await
{
Ok(topology) => Some(topology),
Err(err) => {
error!("Initial topology is invalid - {:?}. Right now it will be impossible to send any packets through the mixnet!", err);
None
}
};
TopologyControl {
directory_server,
refresh_rate,
inner: Arc::new(FRwLock::new(Inner::new(initial_topology))),
}
}
async fn get_current_compatible_topology(directory_server: String) -> Result<T, TopologyError> {
let full_topology = T::new(directory_server.clone());
// run a healthcheck to determine healthy-ish nodes:
// this is a temporary solution as the healthcheck will eventually be moved to validators
let healthcheck_config = healthcheck::config::HealthCheck {
directory_server,
// those are literally irrelevant when running single check
interval: 100000.0,
resolution_timeout: 5.0,
num_test_packets: 2,
};
let healthcheck = healthcheck::HealthChecker::new(healthcheck_config);
let healthcheck_result = healthcheck.do_check().await;
let healthcheck_scores = match healthcheck_result {
Err(err) => {
error!("Error while performing the healtcheck: {:?}", err);
return Err(TopologyError::HealthCheckError);
}
Ok(scores) => scores,
};
let healthy_topology =
healthcheck_scores.filter_topology_by_score(&full_topology, NODE_HEALTH_THRESHOLD);
let versioned_healthy_topology = healthy_topology.filter_node_versions(
built_info::PKG_VERSION,
built_info::PKG_VERSION,
built_info::PKG_VERSION,
);
// make sure you can still send a packet through the network:
if !versioned_healthy_topology.can_construct_path_through() {
return Err(TopologyError::NoValidPathsError);
}
Ok(versioned_healthy_topology)
}
pub(crate) fn get_inner_ref(&self) -> Arc<FRwLock<Inner<T>>> {
self.inner.clone()
}
async fn update_global_topology(&mut self, new_topology: Option<T>) {
// acquire write lock
let mut write_lock = self.inner.write().await;
write_lock.topology = new_topology;
}
async fn should_update_topology(&mut self, new_topology: &Option<T>) -> bool {
let read_lock = self.inner.read().await;
match new_topology {
// if new topology is invalid, we MUST update to it as it is impossible to send packets through
None => true,
Some(new_topology) => match &read_lock.topology {
None => true,
Some(old_topology) => new_topology != old_topology,
},
}
}
pub(crate) async fn run_refresher(mut self) {
info!("Starting topology refresher");
let delay_duration = Duration::from_secs_f64(self.refresh_rate);
loop {
trace!("Refreshing the topology");
let new_topology_res =
Self::get_current_compatible_topology(self.directory_server.clone()).await;
let new_topology = match new_topology_res {
Ok(topology) => Some(topology),
Err(err) => {
warn!("the obtained topology seems to be invalid - {:?}, it will be impossible to send packets through", err);
None
}
};
if self.should_update_topology(&new_topology).await {
info!("Detected changes in topology - updating global view!");
self.update_global_topology(new_topology).await;
}
tokio::time::delay_for(delay_duration).await;
}
}
}
pub struct Inner<T: NymTopology> {
pub topology: Option<T>,
}
impl<T: NymTopology> Inner<T> {
fn new(topology: Option<T>) -> Self {
Inner { topology }
}
}
nym-client/src/sockets/tcp.rs
+29 -22
View File
@@ -1,18 +1,17 @@
use crate::client::received_buffer::BufferResponse;
use crate::client::topology_control::TopologyInnerRef;
use crate::client::InputMessage;
use directory_client::presence::Topology;
use futures::channel::{mpsc, oneshot};
use futures::future::FutureExt;
use futures::io::Error;
use futures::SinkExt;
use log::*;
use sphinx::route::{Destination, DestinationAddressBytes};
use std::borrow::Borrow;
use std::convert::TryFrom;
use std::io;
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::prelude::*;
use topology::NymTopology;
const SEND_REQUEST_PREFIX: u8 = 1;
const FETCH_REQUEST_PREFIX: u8 = 2;
@@ -124,14 +123,24 @@ impl ClientRequest {
ServerResponse::Fetch { messages }
}
async fn handle_get_clients(topology: &Topology) -> ServerResponse {
let clients = topology
.mix_provider_nodes
.iter()
.flat_map(|provider| provider.registered_clients.iter())
.map(|client| base64::decode_config(&client.pub_key, base64::URL_SAFE).unwrap()) // TODO: this can potentially throw an error
.collect();
ServerResponse::GetClients { clients }
async fn handle_get_clients<T: NymTopology>(topology: &TopologyInnerRef<T>) -> ServerResponse {
let topology_data = &topology.read().await.topology;
match topology_data {
Some(topology) => {
let clients = topology
.get_mix_provider_nodes()
.iter()
.flat_map(|provider| provider.registered_clients.iter())
.filter_map(|client| {
base64::decode_config(&client.pub_key, base64::URL_SAFE).ok()
})
.collect();
ServerResponse::GetClients { clients }
}
None => ServerResponse::Error {
message: "Invalid network topology".to_string(),
},
}
}
async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse {
@@ -196,9 +205,9 @@ impl ServerResponse {
}
}
async fn handle_connection(
async fn handle_connection<T: NymTopology>(
data: &[u8],
request_handling_data: RequestHandlingData,
request_handling_data: RequestHandlingData<T>,
) -> Result<ServerResponse, TCPSocketError> {
let request = ClientRequest::try_from(data)?;
let response = match request {
@@ -211,7 +220,7 @@ async fn handle_connection(
}
ClientRequest::Fetch => ClientRequest::handle_fetch(request_handling_data.msg_query).await,
ClientRequest::GetClients => {
ClientRequest::handle_get_clients(request_handling_data.topology.borrow()).await
ClientRequest::handle_get_clients(&request_handling_data.topology).await
}
ClientRequest::OwnDetails => {
ClientRequest::handle_own_details(request_handling_data.self_address).await
@@ -221,27 +230,25 @@ async fn handle_connection(
Ok(response)
}
struct RequestHandlingData {
struct RequestHandlingData<T: NymTopology> {
msg_input: mpsc::UnboundedSender<InputMessage>,
msg_query: mpsc::UnboundedSender<BufferResponse>,
self_address: DestinationAddressBytes,
topology: Arc<Topology>,
topology: TopologyInnerRef<T>,
}
async fn accept_connection(
async fn accept_connection<T: 'static + NymTopology>(
mut socket: tokio::net::TcpStream,
msg_input: mpsc::UnboundedSender<InputMessage>,
msg_query: mpsc::UnboundedSender<BufferResponse>,
self_address: DestinationAddressBytes,
topology: Topology,
topology: TopologyInnerRef<T>,
) {
let address = socket
.peer_addr()
.expect("connected streams should have a peer address");
debug!("Peer address: {}", address);
let topology = Arc::new(topology);
let mut buf = [0u8; 2048];
// In a loop, read data from the socket and write the data back.
@@ -280,12 +287,12 @@ async fn accept_connection(
}
}
pub async fn start_tcpsocket(
pub async fn start_tcpsocket<T: 'static + NymTopology>(
address: SocketAddr,
message_tx: mpsc::UnboundedSender<InputMessage>,
received_messages_query_tx: mpsc::UnboundedSender<BufferResponse>,
self_address: DestinationAddressBytes,
topology: Topology,
topology: TopologyInnerRef<T>,
) -> Result<(), TCPSocketError> {
let mut listener = tokio::net::TcpListener::bind(address).await?;
nym-client/src/sockets/ws.rs
+29 -21
View File
@@ -1,6 +1,6 @@
use crate::client::received_buffer::BufferResponse;
use crate::client::topology_control::TopologyInnerRef;
use crate::client::InputMessage;
use directory_client::presence::Topology;
use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
use futures::channel::{mpsc, oneshot};
use futures::future::FutureExt;
@@ -12,20 +12,21 @@ use sphinx::route::{Destination, DestinationAddressBytes};
use std::convert::TryFrom;
use std::io;
use std::net::SocketAddr;
use topology::NymTopology;
use tungstenite::protocol::frame::coding::CloseCode;
use tungstenite::protocol::{CloseFrame, Message};
struct Connection {
struct Connection<T: NymTopology> {
address: SocketAddr,
msg_input: mpsc::UnboundedSender<InputMessage>,
msg_query: mpsc::UnboundedSender<BufferResponse>,
rx: UnboundedReceiver<Message>,
self_address: DestinationAddressBytes,
topology: Topology,
topology: TopologyInnerRef<T>,
tx: UnboundedSender<Message>,
}
impl Connection {
impl<T: NymTopology> Connection<T> {
async fn handle_text_message(&self, msg: String) -> ServerResponse {
debug!("Handling text message request");
trace!("Content: {:?}", msg.clone());
@@ -48,9 +49,7 @@ impl Connection {
ClientRequest::handle_send(message, recipient_address, self.msg_input.clone()).await
}
ClientRequest::Fetch => ClientRequest::handle_fetch(self.msg_query.clone()).await,
ClientRequest::GetClients => {
ClientRequest::handle_get_clients(self.topology.clone()).await
}
ClientRequest::GetClients => ClientRequest::handle_get_clients(&self.topology).await,
ClientRequest::OwnDetails => ClientRequest::handle_own_details(self.self_address).await,
}
}
@@ -212,7 +211,7 @@ impl ClientRequest {
Err(e) => {
return ServerResponse::Error {
message: e.to_string(),
}
};
}
Ok(hex) => hex,
};
@@ -269,14 +268,22 @@ impl ClientRequest {
ServerResponse::Fetch { messages }
}
async fn handle_get_clients(topology: Topology) -> ServerResponse {
let clients = topology
.mix_provider_nodes
.into_iter()
.flat_map(|provider| provider.registered_clients.into_iter())
.map(|client| client.pub_key)
.collect();
ServerResponse::GetClients { clients }
async fn handle_get_clients<T: NymTopology>(topology: &TopologyInnerRef<T>) -> ServerResponse {
let topology_data = &topology.read().await.topology;
match topology_data {
Some(topology) => {
let clients = topology
.get_mix_provider_nodes()
.iter()
.flat_map(|provider| provider.registered_clients.iter())
.map(|client| client.pub_key.clone())
.collect();
ServerResponse::GetClients { clients }
}
None => ServerResponse::Error {
message: "Invalid network topology".to_string(),
},
}
}
async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse {
@@ -306,14 +313,13 @@ impl Into<Message> for ServerResponse {
}
}
async fn accept_connection(
async fn accept_connection<T: 'static + NymTopology>(
stream: tokio::net::TcpStream,
msg_input: mpsc::UnboundedSender<InputMessage>,
msg_query: mpsc::UnboundedSender<BufferResponse>,
self_address: DestinationAddressBytes,
topology: Topology,
topology: TopologyInnerRef<T>,
) {
warn!("accept_connection");
let address = stream
.peer_addr()
.expect("connected streams should have a peer address");
@@ -337,6 +343,8 @@ async fn accept_connection(
msg_query,
self_address,
};
// TODO: make sure this actually doesn't leak memory...
tokio::spawn(conn.handle());
while let Some(message) = ws_stream.next().await {
@@ -385,12 +393,12 @@ async fn accept_connection(
}
}
pub async fn start_websocket(
pub async fn start_websocket<T: 'static + NymTopology>(
address: SocketAddr,
message_tx: mpsc::UnboundedSender<InputMessage>,
received_messages_query_tx: mpsc::UnboundedSender<BufferResponse>,
self_address: DestinationAddressBytes,
topology: Topology,
topology: TopologyInnerRef<T>,
) -> Result<(), WebSocketError> {
let mut listener = tokio::net::TcpListener::bind(address).await?;