[Add] client side identification

src/omega/omega_connection.rs

@@ -12,7 +12,7 @@ use dashmap::DashMap;
 use epsilon_core::{CommunicationType, CommunicationValue, DataTypes, DataValue, rand_u32};
 use epsilon_native::{Receiver, Sender};
 use once_cell::sync::Lazy;
-use std::{collections::HashMap, env, net::SocketAddr, sync::Arc, time::Duration};
+use std::{collections::HashMap, env, sync::Arc, time::Duration};
 use tokio::{
     sync::{Mutex, RwLock, mpsc, watch},
     task::JoinHandle,
@@ -250,9 +250,6 @@ impl OmegaConnection {
             addr_str
         );
 
-        // Reset reconnect delay on successful connection
-        let reconnect_delay = RECONNECT_DELAY;
-
         // Store sender
         let sender_arc = Arc::new(sender);
         *self.sender.write().await = Some(sender_arc.clone());

src/rho/client_connection.rs

@@ -76,7 +76,7 @@ impl ClientConnection {
         if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) {
             log_cv_out!(PrintType::Client, &cv);
         }
-        self.sender.send(&cv).await;
+        let _ = self.sender.send(&cv).await;
     }
 
     /// Handle incoming message from client

src/rho/connection.rs

@@ -61,7 +61,7 @@ impl GeneralConnection {
         loop {
             let cv = match self.receiver.receive().await {
                 Ok(v) => v,
-                Err(e) => {
+                Err(_) => {
                     break;
                 }
             };
@@ -104,7 +104,7 @@ impl GeneralConnection {
 
             let response_cv = match response_cv {
                 Ok(r) => r,
-                Err(e) => {
+                Err(_) => {
                     return;
                 }
             };
@@ -140,6 +140,64 @@ impl GeneralConnection {
                     .export(DataFormat::Base64);
 
             let response = CommunicationValue::new(CommunicationType::challenge)
+                .with_id(cv.get_id())
+                .add_data(
+                    DataTypes::public_key,
+                    DataValue::Str(public_key_to_base64(&get_public_key())),
+                )
+                .add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
+
+            let _ = self.sender.send(&response).await;
+        } else if let DataValue::Number(user_id) = cv.get_data(DataTypes::user_id) {
+            *self.id.write().await = *user_id as u64;
+            *self.connection_kind.write().await = Some(ConnectionKind::Client);
+
+            let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_user_data)
+                .add_data(DataTypes::user_id, DataValue::Number(*user_id));
+
+            let response_cv = get_omega_connection()
+                .await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
+                .await;
+
+            let response_cv = match response_cv {
+                Ok(r) => r,
+                Err(_) => {
+                    return;
+                }
+            };
+
+            let base64_pub = response_cv
+                .get_data(DataTypes::public_key)
+                .as_str()
+                .unwrap_or("");
+
+            let pub_key = match load_public_key(base64_pub) {
+                Some(pk) => pk,
+                None => {
+                    return;
+                }
+            };
+
+            *self.pub_key.write().await = Some(pub_key.as_bytes().to_vec());
+
+            let challenge: String = rand::thread_rng()
+                .sample_iter(&Alphanumeric)
+                .take(32)
+                .map(char::from)
+                .collect();
+
+            *self.challenge.write().await = challenge.clone();
+            *self.identified.write().await = true;
+
+            let encrypted_challenge =
+                SecurePayload::new(challenge.as_bytes(), DataFormat::Raw, get_private_key())
+                    .unwrap()
+                    .encrypt_x448(pub_key)
+                    .unwrap()
+                    .export(DataFormat::Base64);
+
+            let response = CommunicationValue::new(CommunicationType::challenge)
+                .with_id(cv.get_id())
                 .add_data(
                     DataTypes::public_key,
                     DataValue::Str(public_key_to_base64(&get_public_key())),
@@ -202,10 +260,18 @@ impl GeneralConnection {
 
         match kind {
             ConnectionKind::Client => {
+                let notify = CommunicationValue::new(CommunicationType::user_connected)
+                    .add_data(DataTypes::user_id, DataValue::Number(id as i64));
+                get_omega_connection().send_message(&notify).await;
+
                 let client = ClientConnection::from_general(self.clone(), id).await;
                 client.start();
             }
             ConnectionKind::Iota => {
+                let notify = CommunicationValue::new(CommunicationType::iota_connected)
+                    .add_data(DataTypes::iota_id, DataValue::Number(id as i64));
+                get_omega_connection().send_message(&notify).await;
+
                 let iota = IotaConnection::from_general(self.clone(), id).await;
 
                 let rho = Arc::new(RhoConnection::new(iota.clone(), Vec::new()).await);

src/rho/iota_connection.rs

@@ -3,8 +3,6 @@ use crate::calls::call_manager;
 use crate::log_cv_in;
 use crate::log_cv_out;
 use crate::log_err;
-use crate::log_in;
-use crate::log_out;
 use crate::omega::omega_connection::get_omega_connection;
 use crate::rho::connection::GeneralConnection;
 use crate::util::logger::PrintType;