diff --git a/.gitignore b/.gitignore index c384ecb2ef..7fd7f19fc0 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,6 @@ -/target +/targetString **/*.rs.bk /*/target -/test_inbox \ No newline at end of file +/test_inbox +.idea +target \ No newline at end of file diff --git a/src/main.rs b/src/main.rs index 00662a0f59..249b26fe14 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,11 +1,9 @@ -use crate::provider::presence; use crate::provider::ServiceProvider; use clap::{App, Arg, ArgMatches, SubCommand}; use nym_client::identity::mixnet::KeyPair; use std::net::ToSocketAddrs; use std::path::PathBuf; use std::process; -use std::thread; pub mod provider; @@ -72,12 +70,6 @@ fn run(matches: &ArgMatches) { let config = new_config(matches); let provider = ServiceProvider::new(&config); - // Start sending presence notifications in a separate thread - thread::spawn(move || { - let notifier = presence::Notifier::new(&config); - notifier.run(); - }); - provider.start().unwrap() } diff --git a/src/provider/client_handling/mod.rs b/src/provider/client_handling/mod.rs index 6d22cad3d2..d252fc3cc2 100644 --- a/src/provider/client_handling/mod.rs +++ b/src/provider/client_handling/mod.rs @@ -52,14 +52,14 @@ impl From for ClientProcessingError { #[derive(Debug)] pub(crate) struct ClientProcessingData { store_dir: PathBuf, - registered_clients_ledger: ClientLedger, + registered_clients_ledger: Arc>, secret_key: Scalar, } impl ClientProcessingData { pub(crate) fn new( store_dir: PathBuf, - registered_clients_ledger: ClientLedger, + registered_clients_ledger: Arc>, secret_key: Scalar, ) -> Self { ClientProcessingData { @@ -69,17 +69,17 @@ impl ClientProcessingData { } } - pub(crate) fn add_arc_futures_mutex(self) -> Arc> { - Arc::new(FMutex::new(self)) + pub(crate) fn add_arc(self) -> Arc { + Arc::new(self) } } -pub(crate) struct ClientRequestProcessor(()); +pub(crate) struct ClientRequestProcessor; impl ClientRequestProcessor { pub(crate) async fn process_client_request( data: &[u8], - processing_data: Arc>, + processing_data: Arc, ) -> Result, ClientProcessingError> { let client_request = ProviderRequests::from_bytes(&data)?; println!("Received the following request: {:?}", client_request); @@ -100,20 +100,19 @@ impl ClientRequestProcessor { async fn process_pull_messages_request( req: PullRequest, - processing_data: Arc>, + processing_data: Arc, ) -> Result { // TODO: this lock is completely unnecessary as we're only reading the data. // Wait for https://github.com/nymtech/nym-sfw-provider/issues/19 to resolve. - let unlocked = processing_data.lock().await; + let unlocked_ledger = processing_data.registered_clients_ledger.lock().await; println!("Processing pull!"); - if unlocked.registered_clients_ledger.has_token(req.auth_token) { - let store_dir_clone = unlocked.store_dir.clone(); + if unlocked_ledger.has_token(req.auth_token) { // drop the mutex so that we could do IO without blocking others wanting to get the lock - drop(unlocked); + drop(unlocked_ledger); let retrieved_messages = ClientStorage::retrieve_client_files( req.destination_address, - store_dir_clone.as_path(), + processing_data.store_dir.as_path(), )?; Ok(PullResponse::new(retrieved_messages)) } else { @@ -123,22 +122,20 @@ impl ClientRequestProcessor { async fn register_new_client( req: RegisterRequest, - processing_data: Arc>, + processing_data: Arc, ) -> Result { println!("Processing register new client request!"); - let mut unlocked = processing_data.lock().await; + let mut unlocked_ledger = processing_data.registered_clients_ledger.lock().await; let auth_token = ClientRequestProcessor::generate_new_auth_token( req.destination_address.to_vec(), - unlocked.secret_key, + processing_data.secret_key, ); - if !unlocked.registered_clients_ledger.has_token(auth_token) { - unlocked - .registered_clients_ledger - .insert_token(auth_token, req.destination_address); + if !unlocked_ledger.has_token(auth_token) { + unlocked_ledger.insert_token(auth_token, req.destination_address); ClientRequestProcessor::create_storage_dir( req.destination_address, - unlocked.store_dir.as_path(), + processing_data.store_dir.as_path(), )?; } Ok(RegisterResponse::new(auth_token)) diff --git a/src/provider/mod.rs b/src/provider/mod.rs index 3bb6557a52..3b588fcb6d 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -5,6 +5,7 @@ use curve25519_dalek::montgomery::MontgomeryPoint; use curve25519_dalek::scalar::Scalar; use futures::io::Error; use futures::lock::Mutex as FMutex; +use nym_client::clients::directory::presence::MixProviderClient; use sfw_provider_requests::AuthToken; use sphinx::route::DestinationAddressBytes; use std::collections::HashMap; @@ -34,8 +35,8 @@ pub struct Config { } impl Config { - pub fn public_key_string(&self) -> String { - let key_bytes = self.public_key.to_bytes().to_vec(); + pub fn public_key_string(public_key: MontgomeryPoint) -> String { + let key_bytes = public_key.to_bytes().to_vec(); base64::encode_config(&key_bytes, base64::URL_SAFE) } } @@ -74,6 +75,10 @@ impl ClientLedger { ClientLedger(HashMap::new()) } + fn add_arc_futures_mutex(self) -> Arc> { + Arc::new(FMutex::new(self)) + } + fn has_token(&self, auth_token: AuthToken) -> bool { return self.0.contains_key(&auth_token); } @@ -86,6 +91,14 @@ impl ClientLedger { self.0.insert(auth_token, client_address) } + fn current_clients(&self) -> Vec { + self.0 + .iter() + .map(|(_, &v)| Config::public_key_string(MontgomeryPoint(v))) + .map(|pub_key| MixProviderClient { pub_key }) + .collect() + } + #[allow(dead_code)] fn load(_file: PathBuf) -> Self { unimplemented!() @@ -93,9 +106,11 @@ impl ClientLedger { } pub struct ServiceProvider { + directory_server: String, mix_network_address: SocketAddr, client_network_address: SocketAddr, secret_key: Scalar, + public_key: MontgomeryPoint, store_dir: PathBuf, registered_clients_ledger: ClientLedger, } @@ -106,9 +121,11 @@ impl ServiceProvider { mix_network_address: config.mix_socket_address, client_network_address: config.client_socket_address, secret_key: config.secret_key, + public_key: config.public_key, store_dir: PathBuf::from(config.store_dir.clone()), // TODO: load initial ledger from file registered_clients_ledger: ClientLedger::new(), + directory_server: config.directory_server.clone(), } } @@ -172,7 +189,7 @@ impl ServiceProvider { // TODO: FIGURE OUT HOW TO SET READ_DEADLINES IN TOKIO async fn process_client_socket_connection( mut socket: tokio::net::TcpStream, - processing_data: Arc>, + processing_data: Arc, ) { let mut buf = [0; 1024]; @@ -242,12 +259,12 @@ impl ServiceProvider { async fn start_client_listening( address: SocketAddr, store_dir: PathBuf, - client_ledger: ClientLedger, + client_ledger: Arc>, secret_key: Scalar, ) -> Result<(), ProviderError> { let mut listener = tokio::net::TcpListener::bind(address).await?; let processing_data = - ClientProcessingData::new(store_dir, client_ledger, secret_key).add_arc_futures_mutex(); + ClientProcessingData::new(store_dir, client_ledger, secret_key).add_arc(); loop { let (socket, _) = listener.accept().await?; @@ -268,6 +285,17 @@ impl ServiceProvider { let mut rt = Runtime::new()?; // let mut h = rt.handle(); + let initial_client_ledger = self.registered_clients_ledger; + let thread_shareable_ledger = initial_client_ledger.add_arc_futures_mutex(); + + let presence_notifier = presence::Notifier::new( + self.directory_server, + self.mix_network_address.clone(), + self.public_key, + thread_shareable_ledger.clone(), + ); + + let presence_future = rt.spawn(presence_notifier.run()); let mix_future = rt.spawn(ServiceProvider::start_mixnet_listening( self.mix_network_address, self.secret_key, @@ -276,12 +304,13 @@ impl ServiceProvider { let client_future = rt.spawn(ServiceProvider::start_client_listening( self.client_network_address, self.store_dir.clone(), - self.registered_clients_ledger, // we're just cloning the initial ledger state + thread_shareable_ledger, self.secret_key, )); // Spawn the root task rt.block_on(async { - let future_results = futures::future::join(mix_future, client_future).await; + let future_results = + futures::future::join3(mix_future, client_future, presence_future).await; assert!(future_results.0.is_ok() && future_results.1.is_ok()); }); diff --git a/src/provider/presence.rs b/src/provider/presence.rs index 5368214a5d..d1790b685a 100644 --- a/src/provider/presence.rs +++ b/src/provider/presence.rs @@ -1,45 +1,64 @@ -use crate::provider; +use crate::provider::{ClientLedger, Config}; +use curve25519_dalek::montgomery::MontgomeryPoint; +use futures::lock::Mutex as FMutex; use nym_client::clients::directory; -use nym_client::clients::directory::DirectoryClient; -use std::thread; -use std::time::Duration; - use nym_client::clients::directory::presence::MixProviderPresence; use nym_client::clients::directory::requests::presence_providers_post::PresenceMixProviderPoster; +use nym_client::clients::directory::DirectoryClient; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; pub struct Notifier { pub net_client: directory::Client, - presence: MixProviderPresence, + client_ledger: Arc>, + host: String, + pub_key: String, } impl Notifier { - pub fn new(config: &provider::Config) -> Notifier { + pub fn new( + directory_server_address: String, + host: SocketAddr, + pub_key: MontgomeryPoint, + client_ledger: Arc>, + ) -> Notifier { let directory_config = directory::Config { - base_url: config.directory_server.clone(), + base_url: directory_server_address, }; let net_client = directory::Client::new(directory_config); - let presence = MixProviderPresence { - host: config.mix_socket_address.to_string(), - pub_key: config.public_key_string(), - registered_clients: vec![], - }; + Notifier { net_client, - presence, + host: host.to_string(), + pub_key: Config::public_key_string(pub_key), + client_ledger, } } - pub fn notify(&self) { + async fn make_presence(&self) -> MixProviderPresence { + let unlocked_ledger = self.client_ledger.lock().await; + + MixProviderPresence { + host: self.host.clone(), + pub_key: self.pub_key.clone(), + registered_clients: unlocked_ledger.current_clients(), + } + } + + pub fn notify(&self, presence: MixProviderPresence) { self.net_client .presence_providers_post - .post(&self.presence) + .post(&presence) .unwrap(); } - pub fn run(&self) { + pub async fn run(self) { loop { - self.notify(); - thread::sleep(Duration::from_secs(5)); + let presence = self.make_presence().await; + self.notify(presence); + let delay_duration = Duration::from_secs(5); + tokio::time::delay_for(delay_duration).await; } } }