diff --git a/Cargo.lock b/Cargo.lock index 7634a80..5a5541b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1090,7 +1090,7 @@ dependencies = [ [[package]] name = "epsilon-core" version = "0.1.0" -source = "git+https://github.com/Tensamin/Epsilon.git#44e3a3ec0ee6ac4c00f8cfb7c20c3299c7ae89c5" +source = "git+https://github.com/Tensamin/Epsilon.git#a175705a731a124bca7507ea897d29f536bb15dc" dependencies = [ "byteorder", "quinn", @@ -1102,7 +1102,7 @@ dependencies = [ [[package]] name = "epsilon-native" version = "0.1.0" -source = "git+https://github.com/Tensamin/Epsilon.git#44e3a3ec0ee6ac4c00f8cfb7c20c3299c7ae89c5" +source = "git+https://github.com/Tensamin/Epsilon.git#a175705a731a124bca7507ea897d29f536bb15dc" dependencies = [ "anyhow", "async-trait", diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index 7c15ba8..e0cb2ba 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -46,9 +46,9 @@ impl AnonymousClientConnection { pub fn start(self: Arc) { let self_clone = self.clone(); tokio::spawn(async move { - while let Ok(cv) = self_clone.receiver.receive().await { + /*while let Ok(cv) = self_clone.receiver.receive().await { self_clone.clone().handle_message(cv).await; - } + }*/ }); } diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index d7a407b..d3efdb2 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -213,7 +213,7 @@ impl OmegaConnection { let addr_str = format!("https://{}:{}", self.host, self.port); - let (sender, receiver) = epsilon_native::client::connect(&addr_str, None) + let (sender, mut receiver) = epsilon_native::client::connect(&addr_str, None) .await .map_err(|e| format!("Connection failed: {}", e))?; @@ -231,7 +231,7 @@ impl OmegaConnection { // Start read loop let read_self = self.clone(); let read_handle = tokio::spawn(async move { - read_self.read_loop(receiver).await; + read_self.read_loop(&mut receiver).await; }); // Send identification @@ -275,7 +275,7 @@ impl OmegaConnection { let identify_msg = CommunicationValue::new(CommunicationType::identification) .with_id(id) - .add_data(DataTypes::omikron, DataValue::Number(omikron_id)); + .add_data(DataTypes::omikron_id, DataValue::Number(omikron_id)); WAITING_TASKS.insert( id, @@ -404,17 +404,17 @@ impl OmegaConnection { // Read Loop & Heartbeat // ------------------------------------------------------------------------- - async fn read_loop(self: Arc, receiver: Receiver) { + async fn read_loop(self: Arc, receiver: &mut Receiver) { loop { match receiver.receive().await { Ok(cv) => { + log_cv_in!(&cv); + if cv.is_type(CommunicationType::pong) || cv.is_type(CommunicationType::ping) { self.handle_pong(&cv).await; continue; } - log_cv_in!(&cv); - let msg_id = cv.get_id(); if let Some((_, task)) = WAITING_TASKS.remove(&msg_id) { if (task.task)(self.clone(), cv) { @@ -474,9 +474,7 @@ impl OmegaConnection { // ------------------------------------------------------------------------- pub async fn send_message(&self, cv: &CommunicationValue) { - if !cv.is_type(CommunicationType::ping) { - log_cv_out!(cv); - } + log_cv_out!(cv); let sender_guard = self.sender.read().await; if let Some(sender) = sender_guard.as_ref() { diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 13f4e8f..33ddb5a 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -42,9 +42,9 @@ impl ClientConnection { pub fn start(self: Arc) { let self_clone = self.clone(); tokio::spawn(async move { - while let Ok(cv) = self_clone.receiver.receive().await { + /*while let Ok(cv) = self_clone.receiver.receive().await { self_clone.clone().handle_message(cv).await; - } + }*/ }); } diff --git a/src/rho/connection.rs b/src/rho/connection.rs index 21bd7fe..a7c2e1b 100644 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -53,7 +53,7 @@ impl GeneralConnection { impl GeneralConnection { pub async fn handle(self: Arc) { loop { - let cv = match self.receiver.receive().await { + /*let cv = match self.receiver.receive().await { Ok(v) => v, Err(_) => break, }; @@ -70,7 +70,7 @@ impl GeneralConnection { if self.migrate().await { break; - } + }*/ } } async fn handle_identification(self: &Arc, cv: CommunicationValue) { diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index cc7ff64..cea0e88 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -13,6 +13,7 @@ use epsilon_core::DataTypes; use epsilon_core::DataValue; use epsilon_native::Receiver; use epsilon_native::Sender; +use std::collections::BTreeMap; use std::{ collections::HashMap, sync::{Arc, Weak}, @@ -53,9 +54,9 @@ impl IotaConnection { pub fn start(self: Arc) { let self_clone = self.clone(); tokio::spawn(async move { - while let Ok(cv) = self_clone.receiver.receive().await { + /*while let Ok(cv) = self_clone.receiver.receive().await { self_clone.clone().handle_message(cv).await; - } + }*/ }); } @@ -243,7 +244,7 @@ impl IotaConnection { let admin = call_self.has_admin(); // Build call container - let mut call_map: HashMap = HashMap::new(); + let mut call_map: BTreeMap = BTreeMap::new(); call_map.insert(DataTypes::call_id, DataValue::Str(call.call_id.to_string())); @@ -278,7 +279,7 @@ impl IotaConnection { if let DataValue::Array(users) = cv.get_data(DataTypes::user_ids) { for user_val in users { if let DataValue::Container(entries) = user_val { - let mut user_map: HashMap = + let mut user_map: BTreeMap = entries.iter().cloned().collect(); // extract user_id diff --git a/src/rho/server.rs b/src/rho/server.rs index 987e8bc..df72b06 100644 --- a/src/rho/server.rs +++ b/src/rho/server.rs @@ -1,32 +1,17 @@ -use crate::{log, rho::connection::GeneralConnection, util::file_util::load_file_vec}; +use crate::{ + log, + rho::connection::GeneralConnection, + util::{file_util::load_file_vec, logger::PrintType}, +}; use epsilon_native::Host; pub async fn start(port: u16) -> Result<(), Box> { - let cert_pem = load_file_vec("certs", "cert.pem") - .map_err(|e| format!("Failed to load certificate: {}", e))?; - let key_pem = load_file_vec("certs", "key.pem") - .map_err(|e| format!("Failed to load private key: {}", e))?; + let cert_pem = load_file_vec("certs", "cert.pem").expect("Error loading Pemfile"); -<<<<<<< HEAD - let key_pem = load_file_vec("certs", "key.pem").unwrap(); - let cert_pem = load_file_vec("certs", "cert.pem").unwrap(); + let key_pem = load_file_vec("certs", "key.pem").expect("Error loading Keyfile"); - let mut host: Host = epsilon_native::host(port, cert_pem, key_pem).await.unwrap(); - tokio::spawn(async move { - while let Some((sender, receiver)) = host.next().await { - tokio::spawn(async move { - GeneralConnection::new(sender, receiver).handle().await; - }); - } - }); -======= let mut host: Host = epsilon_native::host(port, cert_pem, key_pem).await?; - log!( - 0, - crate::util::logger::PrintType::Omikron, - "Webtransport Server listening on port {}", - port - ); + log!(0, PrintType::General, "Server listening on port {}", port); while let Some((sender, receiver)) = host.next().await { tokio::spawn(async move { @@ -36,5 +21,4 @@ pub async fn start(port: u16) -> Result<(), Box> { } Ok(()) ->>>>>>> b2e6e903f789a81dd3629c21c08f03bbb430280b } diff --git a/src/util/logger.rs b/src/util/logger.rs index d9c40cb..ecfa334 100644 --- a/src/util/logger.rs +++ b/src/util/logger.rs @@ -1,5 +1,5 @@ use std::{ - collections::HashMap, + collections::BTreeMap, fs::{self, OpenOptions}, io::Write, path::Path, @@ -175,7 +175,7 @@ pub fn format_cv(cv: &CommunicationValue) -> String { let comm_type = cv.get_type().to_string(); parts.push(format!("{}", comm_type)); - let data: &HashMap = cv.get_data_container(); + let data: &BTreeMap = cv.get_data_container(); let formated_data = format_data_container(data.iter().map(|(k, v)| (k.clone(), v.clone())).collect());