From 5915e459611bf2b60b034c63489916baa6d403f7 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 10:22:24 +0000 Subject: [PATCH 01/11] Made code compile + moved address to provider client constructor --- src/clients/mod.rs | 10 ++++++++-- src/clients/provider.rs | 22 +++++++++++++--------- 2 files changed, 21 insertions(+), 11 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index 39975b339b..dd818526df 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -130,14 +130,14 @@ impl NymClient { } async fn start_provider_polling(provider_address: SocketAddrV4) { - let provider_client = ProviderClient::new(); + let provider_client = ProviderClient::new(provider_address); loop { println!("[FETCH MSG] - Polling provider..."); let delay_duration = Duration::from_secs_f64(FETCH_MESSAGES_DELAY); tokio::time::delay_for(delay_duration).await; provider_client - .retrieve_messages(provider_address) + .retrieve_messages() .await .unwrap(); } @@ -146,6 +146,12 @@ impl NymClient { pub fn start(self, socket_address: SocketAddr) -> Result<(), Box> { println!("starting nym client"); + + // TODO: registration here + + return Ok(()); + // don't start anything, just register + let (mix_tx, mix_rx) = mpsc::unbounded(); let mut rt = Runtime::new()?; let topology = get_topology(self.is_local); diff --git a/src/clients/provider.rs b/src/clients/provider.rs index ff2fdb6bac..5991b42288 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -1,29 +1,33 @@ use sfw_provider_requests::requests::{ProviderRequest, PullRequest}; use sfw_provider_requests::responses::{ProviderResponse, PullResponse}; -use std::net::Shutdown; +use std::net::{Shutdown, SocketAddr}; use std::net::SocketAddrV4; use tokio::prelude::*; -use tokio::time::Duration; +use std::time::Duration; -pub struct ProviderClient {} +pub struct ProviderClient { + address: SocketAddrV4, +} impl ProviderClient { - pub fn new() -> Self { - ProviderClient {} + pub fn new(address: SocketAddrV4) -> Self { + ProviderClient { + address + } } pub async fn retrieve_messages( &self, - provider: SocketAddrV4, ) -> Result<(), Box> { - let address = [42; 32]; - let pull_request = PullRequest::new(address); + let destination_address = [42u8; 32]; + let dummy_auth_token_to_make_it_compile = [0u8; 32]; + let pull_request = PullRequest::new(destination_address, dummy_auth_token_to_make_it_compile); let bytes = pull_request.to_bytes(); // DH temporary: the provider's client port is not in the topology, but we can't change that // right now without messing up the existing Go mixnet. So I'm going to hardcode this // for the moment until the Go mixnet goes away. - let provider_socket = SocketAddrV4::new(*provider.ip(), 9000); + let provider_socket = SocketAddrV4::new(*self.address.ip(), 9000); println!("Provider: {:?}", provider_socket); let mut socket = tokio::net::TcpStream::connect(provider_socket).await?; From 812a0fe1446023d989898e6f3a86f91d3a29a421 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 10:25:32 +0000 Subject: [PATCH 02/11] Deprecated run command --- src/commands/run.rs | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/src/commands/run.rs b/src/commands/run.rs index 552a6469a6..e239986209 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -5,12 +5,13 @@ use clap::ArgMatches; pub fn execute(matches: &ArgMatches) { println!("{}", banner()); - - let is_local = matches.is_present("local"); - let id = matches.value_of("id").unwrap().to_string(); - println!("Starting client..."); - - let keypair = pemstore::read_keypair_from_disk(id); - let client = NymClient::new(keypair.public_bytes(), is_local); - client.start("127.0.0.1:9000".parse().unwrap()).unwrap(); + panic!("For time being this command is deprecated! Please use 'websocket' instead"); + // + // let is_local = matches.is_present("local"); + // let id = matches.value_of("id").unwrap().to_string(); + // println!("Starting client..."); + // + // let keypair = pemstore::read_keypair_from_disk(id); + // let client = NymClient::new(keypair.public_bytes(), is_local); + // client.start("127.0.0.1:9000".parse().unwrap()).unwrap(); } From 6726329e2c0af1ba7743addfaacb227ff0e2033f Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 10:26:53 +0000 Subject: [PATCH 03/11] Optional auth token for client constructor --- src/clients/mod.rs | 9 ++++++++- src/commands/tcpsocket.rs | 2 +- src/commands/websocket.rs | 4 +++- 3 files changed, 12 insertions(+), 3 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index dd818526df..a9b08bf7e3 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -14,6 +14,7 @@ use std::net::SocketAddr; use std::net::SocketAddrV4; use std::time::Duration; use tokio::runtime::Runtime; +use sfw_provider_requests::requests::AuthToken; pub mod directory; pub mod mix; @@ -65,13 +66,14 @@ pub struct NymClient { input_rx: mpsc::UnboundedReceiver, is_local: bool, + auth_token: Option } #[derive(Debug)] pub struct InputMessage(pub Destination, pub Vec); impl NymClient { - pub fn new(address: DestinationAddressBytes, is_local: bool) -> Self { + pub fn new(address: DestinationAddressBytes, is_local: bool, auth_token: Option) -> Self { let (input_tx, input_rx) = mpsc::unbounded::(); NymClient { @@ -79,6 +81,7 @@ impl NymClient { input_tx, input_rx, is_local, + auth_token, } } @@ -147,6 +150,10 @@ impl NymClient { println!("starting nym client"); + match self.auth_token { + None => println!("Need to register!"), + Some(token) => println!("Already got the token! - {:?}", token), + } // TODO: registration here return Ok(()); diff --git a/src/commands/tcpsocket.rs b/src/commands/tcpsocket.rs index 1e3a21f1d1..c6163a37ac 100644 --- a/src/commands/tcpsocket.rs +++ b/src/commands/tcpsocket.rs @@ -26,7 +26,7 @@ pub fn execute(matches: &ArgMatches) { .expect("Failed to extract the socket address from the iterator"); let keypair = pemstore::read_keypair_from_disk(id); - let _client = NymClient::new(keypair.public_bytes(), is_local); + let _client = NymClient::new(keypair.public_bytes(), is_local, None); // Question: should we be passing the client into the TCP socket somehow next? tcp::start(socket_address); diff --git a/src/commands/websocket.rs b/src/commands/websocket.rs index d26ba5c988..9150a69baa 100644 --- a/src/commands/websocket.rs +++ b/src/commands/websocket.rs @@ -25,7 +25,9 @@ pub fn execute(matches: &ArgMatches) { .expect("Failed to extract the socket address from the iterator"); let keypair = pemstore::read_keypair_from_disk(id); - let client = NymClient::new(keypair.public_bytes(), is_local); + // TODO: reading auth_token from disk (if exists); + let auth_token = None; + let client = NymClient::new(keypair.public_bytes(), is_local, auth_token); client.start(socket_address).unwrap(); } From 7a57078c65ff9c4f923e31b3bef948c2fecb9646 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 10:30:27 +0000 Subject: [PATCH 04/11] Moved socket address from start method to new --- src/clients/mod.rs | 9 +++++---- src/commands/tcpsocket.rs | 7 ++++++- src/commands/websocket.rs | 4 ++-- 3 files changed, 13 insertions(+), 7 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index a9b08bf7e3..d5734f9b5f 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -64,7 +64,7 @@ pub struct NymClient { pub input_tx: mpsc::UnboundedSender, // to be used by "send" function or socket, etc input_rx: mpsc::UnboundedReceiver, - + socket_listening_address: SocketAddr, is_local: bool, auth_token: Option } @@ -73,13 +73,14 @@ pub struct NymClient { pub struct InputMessage(pub Destination, pub Vec); impl NymClient { - pub fn new(address: DestinationAddressBytes, is_local: bool, auth_token: Option) -> Self { + pub fn new(address: DestinationAddressBytes, socket_listening_address: SocketAddr, is_local: bool, auth_token: Option) -> Self { let (input_tx, input_rx) = mpsc::unbounded::(); NymClient { address, input_tx, input_rx, + socket_listening_address, is_local, auth_token, } @@ -146,7 +147,7 @@ impl NymClient { } } - pub fn start(self, socket_address: SocketAddr) -> Result<(), Box> { + pub fn start(self) -> Result<(), Box> { println!("starting nym client"); @@ -185,7 +186,7 @@ impl NymClient { )); let provider_polling_future = rt.spawn(NymClient::start_provider_polling(provider_address)); - let websocket_future = rt.spawn(ws::start_websocket(socket_address, self.input_tx)); + let websocket_future = rt.spawn(ws::start_websocket(self.socket_listening_address, self.input_tx)); rt.block_on(async { let future_results = join5( diff --git a/src/commands/tcpsocket.rs b/src/commands/tcpsocket.rs index c6163a37ac..4ba652bcd6 100644 --- a/src/commands/tcpsocket.rs +++ b/src/commands/tcpsocket.rs @@ -26,7 +26,12 @@ pub fn execute(matches: &ArgMatches) { .expect("Failed to extract the socket address from the iterator"); let keypair = pemstore::read_keypair_from_disk(id); - let _client = NymClient::new(keypair.public_bytes(), is_local, None); + let _client = NymClient::new( + keypair.public_bytes(), + socket_address.clone(), + is_local, + None, + ); // Question: should we be passing the client into the TCP socket somehow next? tcp::start(socket_address); diff --git a/src/commands/websocket.rs b/src/commands/websocket.rs index 9150a69baa..36a88ea9ed 100644 --- a/src/commands/websocket.rs +++ b/src/commands/websocket.rs @@ -27,7 +27,7 @@ pub fn execute(matches: &ArgMatches) { let keypair = pemstore::read_keypair_from_disk(id); // TODO: reading auth_token from disk (if exists); let auth_token = None; - let client = NymClient::new(keypair.public_bytes(), is_local, auth_token); + let client = NymClient::new(keypair.public_bytes(), socket_address, is_local, auth_token); - client.start(socket_address).unwrap(); + client.start().unwrap(); } From ef0e1d081d5cc160ffb746ff5d031579c170907b Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 10:58:47 +0000 Subject: [PATCH 05/11] Included address and auth token to provider client --- src/clients/mod.rs | 26 +++++++++++++---------- src/clients/provider.rs | 47 +++++++++++++++++++++++++++++++++-------- 2 files changed, 53 insertions(+), 20 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index d5734f9b5f..0b18df060d 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -133,8 +133,8 @@ impl NymClient { } } - async fn start_provider_polling(provider_address: SocketAddrV4) { - let provider_client = ProviderClient::new(provider_address); + async fn start_provider_polling(provider_client: ProviderClient) { +// let provider_client = ProviderClient::new(provider_address); loop { println!("[FETCH MSG] - Polling provider..."); @@ -150,6 +150,18 @@ impl NymClient { pub fn start(self) -> Result<(), Box> { println!("starting nym client"); + let topology = get_topology(self.is_local); + let provider_address: SocketAddrV4 = topology + .mix_provider_nodes + .first() + .unwrap() + .host + .parse() + .unwrap(); + + + + let provider_client = ProviderClient::new(provider_address, self.address, self.auth_token); match self.auth_token { None => println!("Need to register!"), @@ -162,14 +174,6 @@ impl NymClient { let (mix_tx, mix_rx) = mpsc::unbounded(); let mut rt = Runtime::new()?; - let topology = get_topology(self.is_local); - let provider_address: SocketAddrV4 = topology - .mix_provider_nodes - .first() - .unwrap() - .host - .parse() - .unwrap(); let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx)); let loop_cover_traffic_future = rt.spawn(NymClient::start_loop_cover_traffic_stream( @@ -185,7 +189,7 @@ impl NymClient { topology.clone(), )); - let provider_polling_future = rt.spawn(NymClient::start_provider_polling(provider_address)); + let provider_polling_future = rt.spawn(NymClient::start_provider_polling(provider_client)); let websocket_future = rt.spawn(ws::start_websocket(self.socket_listening_address, self.input_tx)); rt.block_on(async { diff --git a/src/clients/provider.rs b/src/clients/provider.rs index 5991b42288..1e0f978077 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -1,33 +1,56 @@ -use sfw_provider_requests::requests::{ProviderRequest, PullRequest}; +use sfw_provider_requests::requests::{ProviderRequest, PullRequest, AuthToken, RegisterRequest}; use sfw_provider_requests::responses::{ProviderResponse, PullResponse}; use std::net::{Shutdown, SocketAddr}; use std::net::SocketAddrV4; use tokio::prelude::*; use std::time::Duration; +use sphinx::route::DestinationAddressBytes; +use futures::io::Error; + +#[derive(Debug)] +pub enum ProviderClientError { + EmptyAuthTokenError, + NetworkError, +} + +impl From for ProviderClientError { + fn from(_: Error) -> Self { + use ProviderClientError::*; + + NetworkError + } +} + pub struct ProviderClient { - address: SocketAddrV4, + provider_network_address: SocketAddrV4, + our_address: DestinationAddressBytes, + auth_token: Option, } impl ProviderClient { - pub fn new(address: SocketAddrV4) -> Self { + pub fn new(provider_network_address: SocketAddrV4, our_address: DestinationAddressBytes, auth_token: Option) -> Self { ProviderClient { - address + provider_network_address, + our_address, + auth_token, } } pub async fn retrieve_messages( &self, - ) -> Result<(), Box> { - let destination_address = [42u8; 32]; - let dummy_auth_token_to_make_it_compile = [0u8; 32]; - let pull_request = PullRequest::new(destination_address, dummy_auth_token_to_make_it_compile); + ) -> Result<(), ProviderClientError> { + if self.auth_token.is_none() { + return Err(ProviderClientError::EmptyAuthTokenError) + } + + let pull_request = PullRequest::new(self.our_address, self.auth_token.unwrap()); let bytes = pull_request.to_bytes(); // DH temporary: the provider's client port is not in the topology, but we can't change that // right now without messing up the existing Go mixnet. So I'm going to hardcode this // for the moment until the Go mixnet goes away. - let provider_socket = SocketAddrV4::new(*self.address.ip(), 9000); + let provider_socket = SocketAddrV4::new(*self.provider_network_address.ip(), 9000); println!("Provider: {:?}", provider_socket); let mut socket = tokio::net::TcpStream::connect(provider_socket).await?; @@ -51,4 +74,10 @@ impl ProviderClient { } Ok(()) } + + pub async fn register(&mut self) -> Result> { + let register_request = RegisterRequest::new(self.our_address); + + Ok([0;32]) + } } From d08e51461b0ba98e1c81465e5ecb10073691b9a0 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 11:04:13 +0000 Subject: [PATCH 06/11] Overwriting provider address at 'new' --- src/clients/provider.rs | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/src/clients/provider.rs b/src/clients/provider.rs index 1e0f978077..6802669772 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -30,8 +30,13 @@ pub struct ProviderClient { impl ProviderClient { pub fn new(provider_network_address: SocketAddrV4, our_address: DestinationAddressBytes, auth_token: Option) -> Self { + // DH temporary: the provider's client port is not in the topology, but we can't change that + // right now without messing up the existing Go mixnet. So I'm going to hardcode this + // for the moment until the Go mixnet goes away. + let provider_socket = SocketAddrV4::new(*provider_network_address.ip(), 9000); + ProviderClient { - provider_network_address, + provider_network_address: provider_socket, our_address, auth_token, } @@ -47,13 +52,7 @@ impl ProviderClient { let pull_request = PullRequest::new(self.our_address, self.auth_token.unwrap()); let bytes = pull_request.to_bytes(); - // DH temporary: the provider's client port is not in the topology, but we can't change that - // right now without messing up the existing Go mixnet. So I'm going to hardcode this - // for the moment until the Go mixnet goes away. - let provider_socket = SocketAddrV4::new(*self.provider_network_address.ip(), 9000); - println!("Provider: {:?}", provider_socket); - - let mut socket = tokio::net::TcpStream::connect(provider_socket).await?; + let mut socket = tokio::net::TcpStream::connect(self.provider_network_address).await?; println!("keep alive: {:?}", socket.keepalive()); socket.set_keepalive(Some(Duration::from_secs(2))).unwrap(); socket.write_all(&bytes[..]).await?; From bc143c16c0014e7cab622a5a87de502bb39d2564 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 11:11:07 +0000 Subject: [PATCH 07/11] Moved sending bytes to provider to separate method --- src/clients/provider.rs | 36 ++++++++++++++++++++++-------------- 1 file changed, 22 insertions(+), 14 deletions(-) diff --git a/src/clients/provider.rs b/src/clients/provider.rs index 6802669772..9836d7f242 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -42,6 +42,26 @@ impl ProviderClient { } } + pub async fn send_request(&self, bytes: Vec) -> Result, ProviderClientError> { + let mut socket = tokio::net::TcpStream::connect(self.provider_network_address).await?; + println!("keep alive: {:?}", socket.keepalive()); + socket.set_keepalive(Some(Duration::from_secs(2))).unwrap(); + socket.write_all(&bytes[..]).await?; + if let Err(e) = socket.shutdown(Shutdown::Write) { + // TODO: make it a silent log once we have a proper logging library +// eprintln!("failed to close write part of the socket; err = {:?}", e) + } + + let mut response = Vec::new(); + socket.read_to_end(&mut response).await?; + if let Err(e) = socket.shutdown(Shutdown::Read) { + // TODO: make it a silent log once we have a proper logging library +// eprintln!("failed to close read part of the socket; err = {:?}", e) + } + + Ok(response) + } + pub async fn retrieve_messages( &self, ) -> Result<(), ProviderClientError> { @@ -52,21 +72,9 @@ impl ProviderClient { let pull_request = PullRequest::new(self.our_address, self.auth_token.unwrap()); let bytes = pull_request.to_bytes(); - let mut socket = tokio::net::TcpStream::connect(self.provider_network_address).await?; - println!("keep alive: {:?}", socket.keepalive()); - socket.set_keepalive(Some(Duration::from_secs(2))).unwrap(); - socket.write_all(&bytes[..]).await?; - if let Err(e) = socket.shutdown(Shutdown::Write) { - eprintln!("failed to close write part of the socket; err = {:?}", e) - } - - let mut response = Vec::new(); - socket.read_to_end(&mut response).await?; - if let Err(e) = socket.shutdown(Shutdown::Read) { - eprintln!("failed to close read part of the socket; err = {:?}", e) - } - + let response = self.send_request(bytes).await?; println!("Received the following response: {:?}", response); + let parsed_response = PullResponse::from_bytes(&response).unwrap(); for message in parsed_response.messages { println!("Received: {:?}", String::from_utf8(message).unwrap()) From 2b89be303137d7be399b4456f2c06d4d3f1ca0e8 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 11:20:52 +0000 Subject: [PATCH 08/11] Basic structure for client registration message send --- src/clients/provider.rs | 36 ++++++++++++++++++++++++++++++++---- 1 file changed, 32 insertions(+), 4 deletions(-) diff --git a/src/clients/provider.rs b/src/clients/provider.rs index 9836d7f242..6b951dd0f4 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -1,5 +1,5 @@ use sfw_provider_requests::requests::{ProviderRequest, PullRequest, AuthToken, RegisterRequest}; -use sfw_provider_requests::responses::{ProviderResponse, PullResponse}; +use sfw_provider_requests::responses::{ProviderResponse, PullResponse, RegisterResponse, ProviderResponseError}; use std::net::{Shutdown, SocketAddr}; use std::net::SocketAddrV4; use tokio::prelude::*; @@ -9,8 +9,13 @@ use futures::io::Error; #[derive(Debug)] pub enum ProviderClientError { + ClientAlreadyRegisteredError, EmptyAuthTokenError, NetworkError, + + InvalidRequestError, + InvalidResponseError, + InvalidResponseLengthError, } impl From for ProviderClientError { @@ -21,6 +26,16 @@ impl From for ProviderClientError { } } +impl From for ProviderClientError { + fn from(err: ProviderResponseError) -> Self { + use ProviderClientError::*; + match err { + ProviderResponseError::MarshalError => InvalidRequestError, + ProviderResponseError::UnmarshalError => InvalidResponseError, + ProviderResponseError::UnmarshalErrorInvalidLength => InvalidResponseLengthError, + } + } +} pub struct ProviderClient { provider_network_address: SocketAddrV4, @@ -74,16 +89,29 @@ impl ProviderClient { let response = self.send_request(bytes).await?; println!("Received the following response: {:?}", response); - - let parsed_response = PullResponse::from_bytes(&response).unwrap(); + + let parsed_response = PullResponse::from_bytes(&response)?; for message in parsed_response.messages { println!("Received: {:?}", String::from_utf8(message).unwrap()) } + + // TODO: make it return the actual messages instead Ok(()) } - pub async fn register(&mut self) -> Result> { + pub async fn register(&mut self) -> Result { + if self.auth_token.is_some() { + return Err(ProviderClientError::ClientAlreadyRegisteredError); + } + let register_request = RegisterRequest::new(self.our_address); + let bytes = register_request.to_bytes(); + + let response = self.send_request(bytes).await?; + println!("Received the following response: {:?}", response); + + let parsed_response = RegisterResponse::from_bytes(&response)?; + println!("parsed register: {:?}", parsed_response); Ok([0;32]) } From 0fe7276d447ef33f5885c465235aa3fccd111cd0 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 11:22:46 +0000 Subject: [PATCH 09/11] Method to update auth token of provider client + removed mut from register --- src/clients/provider.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/clients/provider.rs b/src/clients/provider.rs index 6b951dd0f4..cf986041ed 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -57,6 +57,10 @@ impl ProviderClient { } } + pub fn update_token(&mut self, auth_token: AuthToken) { + self.auth_token = Some(auth_token) + } + pub async fn send_request(&self, bytes: Vec) -> Result, ProviderClientError> { let mut socket = tokio::net::TcpStream::connect(self.provider_network_address).await?; println!("keep alive: {:?}", socket.keepalive()); @@ -99,7 +103,7 @@ impl ProviderClient { Ok(()) } - pub async fn register(&mut self) -> Result { + pub async fn register(&self) -> Result { if self.auth_token.is_some() { return Err(ProviderClientError::ClientAlreadyRegisteredError); } From 441a6c3e1b7ed9c2554f78b7cbe765c5dbe1d5ca Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 14:25:29 +0000 Subject: [PATCH 10/11] Client sending registration request on startup --- src/clients/mod.rs | 25 +++++++++++++++++-------- src/clients/provider.rs | 14 +++----------- 2 files changed, 20 insertions(+), 19 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index 0b18df060d..ff58ece3cf 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -140,15 +140,19 @@ impl NymClient { println!("[FETCH MSG] - Polling provider..."); let delay_duration = Duration::from_secs_f64(FETCH_MESSAGES_DELAY); tokio::time::delay_for(delay_duration).await; - provider_client + let messages = provider_client .retrieve_messages() .await .unwrap(); + for message in messages { + println!("Retrieved: {:?}", String::from_utf8(message).unwrap()) + } } } pub fn start(self) -> Result<(), Box> { println!("starting nym client"); + let mut rt = Runtime::new()?; let topology = get_topology(self.is_local); let provider_address: SocketAddrV4 = topology @@ -161,19 +165,24 @@ impl NymClient { - let provider_client = ProviderClient::new(provider_address, self.address, self.auth_token); + let mut provider_client = ProviderClient::new(provider_address, self.address, self.auth_token); + + rt.block_on(async { + match self.auth_token { + None => { + let auth_token = provider_client.register().await.unwrap(); + provider_client.update_token(auth_token); + println!("Obtained new token! - {:?}", auth_token); + }, + Some(token) => println!("Already got the token! - {:?}", token), + } + }); - match self.auth_token { - None => println!("Need to register!"), - Some(token) => println!("Already got the token! - {:?}", token), - } - // TODO: registration here return Ok(()); // don't start anything, just register let (mix_tx, mix_rx) = mpsc::unbounded(); - let mut rt = Runtime::new()?; let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx)); let loop_cover_traffic_future = rt.spawn(NymClient::start_loop_cover_traffic_stream( diff --git a/src/clients/provider.rs b/src/clients/provider.rs index cf986041ed..28c028596d 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -83,7 +83,7 @@ impl ProviderClient { pub async fn retrieve_messages( &self, - ) -> Result<(), ProviderClientError> { + ) -> Result>, ProviderClientError> { if self.auth_token.is_none() { return Err(ProviderClientError::EmptyAuthTokenError) } @@ -95,12 +95,7 @@ impl ProviderClient { println!("Received the following response: {:?}", response); let parsed_response = PullResponse::from_bytes(&response)?; - for message in parsed_response.messages { - println!("Received: {:?}", String::from_utf8(message).unwrap()) - } - - // TODO: make it return the actual messages instead - Ok(()) + Ok(parsed_response.messages) } pub async fn register(&self) -> Result { @@ -112,11 +107,8 @@ impl ProviderClient { let bytes = register_request.to_bytes(); let response = self.send_request(bytes).await?; - println!("Received the following response: {:?}", response); - let parsed_response = RegisterResponse::from_bytes(&response)?; - println!("parsed register: {:?}", parsed_response); - Ok([0;32]) + Ok(parsed_response.auth_token) } } From a9584453097a97d005d0890d931db1ef1e074f1f Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 18 Dec 2019 14:44:07 +0000 Subject: [PATCH 11/11] Got rid of compiler warnings --- src/clients/mod.rs | 8 ++------ src/clients/provider.rs | 11 ++++++----- src/commands/run.rs | 6 +++--- 3 files changed, 11 insertions(+), 14 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index ff58ece3cf..fd5de14d11 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -14,7 +14,7 @@ use std::net::SocketAddr; use std::net::SocketAddrV4; use std::time::Duration; use tokio::runtime::Runtime; -use sfw_provider_requests::requests::AuthToken; +use sfw_provider_requests::AuthToken; pub mod directory; pub mod mix; @@ -155,6 +155,7 @@ impl NymClient { let mut rt = Runtime::new()?; let topology = get_topology(self.is_local); + // this is temporary and assumes there exists only a single provider. let provider_address: SocketAddrV4 = topology .mix_provider_nodes .first() @@ -163,8 +164,6 @@ impl NymClient { .parse() .unwrap(); - - let mut provider_client = ProviderClient::new(provider_address, self.address, self.auth_token); rt.block_on(async { @@ -179,9 +178,6 @@ impl NymClient { }); - return Ok(()); - // don't start anything, just register - let (mix_tx, mix_rx) = mpsc::unbounded(); let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx)); diff --git a/src/clients/provider.rs b/src/clients/provider.rs index 28c028596d..eabed21ef4 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -1,11 +1,12 @@ -use sfw_provider_requests::requests::{ProviderRequest, PullRequest, AuthToken, RegisterRequest}; +use sfw_provider_requests::requests::{ProviderRequest, PullRequest, RegisterRequest}; use sfw_provider_requests::responses::{ProviderResponse, PullResponse, RegisterResponse, ProviderResponseError}; -use std::net::{Shutdown, SocketAddr}; +use std::net::Shutdown; use std::net::SocketAddrV4; use tokio::prelude::*; use std::time::Duration; use sphinx::route::DestinationAddressBytes; use futures::io::Error; +use sfw_provider_requests::AuthToken; #[derive(Debug)] pub enum ProviderClientError { @@ -66,14 +67,14 @@ impl ProviderClient { println!("keep alive: {:?}", socket.keepalive()); socket.set_keepalive(Some(Duration::from_secs(2))).unwrap(); socket.write_all(&bytes[..]).await?; - if let Err(e) = socket.shutdown(Shutdown::Write) { + if let Err(_e) = socket.shutdown(Shutdown::Write) { // TODO: make it a silent log once we have a proper logging library // eprintln!("failed to close write part of the socket; err = {:?}", e) } let mut response = Vec::new(); socket.read_to_end(&mut response).await?; - if let Err(e) = socket.shutdown(Shutdown::Read) { + if let Err(_e) = socket.shutdown(Shutdown::Read) { // TODO: make it a silent log once we have a proper logging library // eprintln!("failed to close read part of the socket; err = {:?}", e) } @@ -85,7 +86,7 @@ impl ProviderClient { &self, ) -> Result>, ProviderClientError> { if self.auth_token.is_none() { - return Err(ProviderClientError::EmptyAuthTokenError) + return Err(ProviderClientError::EmptyAuthTokenError); } let pull_request = PullRequest::new(self.our_address, self.auth_token.unwrap()); diff --git a/src/commands/run.rs b/src/commands/run.rs index e239986209..932da4a190 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -1,9 +1,9 @@ use crate::banner; -use crate::clients::NymClient; -use crate::persistence::pemstore; +//use crate::clients::NymClient; +//use crate::persistence::pemstore; use clap::ArgMatches; -pub fn execute(matches: &ArgMatches) { +pub fn execute(_matches: &ArgMatches) { println!("{}", banner()); panic!("For time being this command is deprecated! Please use 'websocket' instead"); //