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

Made sockets more generic to work on any NymTopology object

Browse Source
This commit is contained in:
Jedrzej Stuczynski
2020-01-23 16:55:39 +00:00
parent 35451a05de
commit 6f69641443
2 changed files with 58 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nym-client/src/sockets/tcp.rs
+29 -23
View File
@@ -1,17 +1,16 @@
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 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;
@@ -129,15 +128,24 @@ impl ClientRequest {
ServerResponse::Fetch { messages }
}
async fn handle_get_clients(topology: &Topology) -> ServerResponse {
println!("get client handle");
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 {
@@ -203,9 +211,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 {
@@ -218,7 +226,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
@@ -228,27 +236,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");
println!("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.
@@ -287,12 +293,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?;