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

Merge pull request #25 from nymtech/feature/presence_with_clients

Browse Source 
Feature/presence with clients
This commit is contained in:
Dave Hrycyszyn
2019-12-19 13:58:06 +00:00
committed by GitHub
parent ac8f04959e cbac85fbf4
commit b9747669d3
5 changed files with 95 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+4 -2
View File
@@ -1,4 +1,6 @@
/target
/targetString
**/*.rs.bk
/*/target
/test_inbox
/test_inbox
.idea
target
src/main.rs
-8
View File
@@ -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()
}
src/provider/client_handling/mod.rs
+17 -20
View File
@@ -52,14 +52,14 @@ impl From<io::Error> for ClientProcessingError {
#[derive(Debug)]
pub(crate) struct ClientProcessingData {
store_dir: PathBuf,
registered_clients_ledger: ClientLedger,
registered_clients_ledger: Arc<FMutex<ClientLedger>>,
secret_key: Scalar,
}
impl ClientProcessingData {
pub(crate) fn new(
store_dir: PathBuf,
registered_clients_ledger: ClientLedger,
registered_clients_ledger: Arc<FMutex<ClientLedger>>,
secret_key: Scalar,
) -> Self {
ClientProcessingData {
@@ -69,17 +69,17 @@ impl ClientProcessingData {
}
}
pub(crate) fn add_arc_futures_mutex(self) -> Arc<FMutex<Self>> {
Arc::new(FMutex::new(self))
pub(crate) fn add_arc(self) -> Arc<Self> {
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<FMutex<ClientProcessingData>>,
processing_data: Arc<ClientProcessingData>,
) -> Result<Vec<u8>, 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<FMutex<ClientProcessingData>>,
processing_data: Arc<ClientProcessingData>,
) -> Result<PullResponse, ClientProcessingError> {
// 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<FMutex<ClientProcessingData>>,
processing_data: Arc<ClientProcessingData>,
) -> Result<RegisterResponse, ClientProcessingError> {
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))
src/provider/mod.rs
+36 -7
View File
@@ -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<FMutex<Self>> {
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<MixProviderClient> {
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<FMutex<ClientProcessingData>>,
processing_data: Arc<ClientProcessingData>,
) {
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<FMutex<ClientLedger>>,
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());
});
src/provider/presence.rs
+38 -19
View File
@@ -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<FMutex<ClientLedger>>,
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<FMutex<ClientLedger>>,
) -> 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;
}
}
}