Calling & Config

Cargo.lock

@@ -34,6 +34,7 @@ dependencies = [
  "sys-info",
  "sysinfo",
  "tokio",
+ "tokio-rustls",
  "tokio-stream",
  "tokio-util",
  "tokio_websocket_server",

Cargo.toml

@@ -32,6 +32,7 @@ sha2 = "*"
 sys-info = "*"
 sysinfo = "0.30"
 tokio = { version = "*", features = ["full"] }
+tokio-rustls = { version = "*" }
 tokio-stream = "*"
 tokio-util = { version = "*", features = ["full"] }
 tokio_websocket_server = "0.1.0"

src/auth/auth_connector.rs

@@ -70,7 +70,7 @@ pub async fn get_iota_id(user_id: Uuid) -> Option<Uuid> {
     let client = client();
     let res = client
         .get(&url)
-        .header("Authorization", CONFIG.lock().await.omikron_id.to_string())
+        .header("Authorization", CONFIG.read().await.omikron_id.to_string())
         .header("Content-Type", "application/json")
         .send()
         .await
@@ -96,7 +96,7 @@ pub async fn is_private_key_valid(user_id: Uuid, pk_hash: &str) -> bool {
     let client = client();
     let res = client
         .get(&url)
-        .header("Authorization", CONFIG.lock().await.omikron_id.to_string())
+        .header("Authorization", CONFIG.read().await.omikron_id.to_string())
         .header("PrivateKeyHash", pk_hash)
         .header("Accept", "application/json")
         .send()

src/calls/call_connection.rs

@@ -59,7 +59,7 @@ impl CallConnection {
         if let Some(sender) = *self.user_id.read().await {
             cv = cv.with_sender(sender);
         }
-        if !cv.is_type(CommunicationType::identification) && !cv.is_type(CommunicationType::ping) {
+        if !cv.is_type(CommunicationType::ping) {
             line(PrintType::CallIn, &cv.to_json().to_string());
         }
 
@@ -101,10 +101,6 @@ impl CallConnection {
         }
 
         let group = call_manager::get_or_create_group(cid, secret_sha).await;
-        {
-            group.lock().await.add_member(uid, self.tx.clone());
-        }
-
         // Build broadcast
         let broadcast = CommunicationValue::new(CommunicationType::client_connected)
             .with_id(cv.get_id().clone())
@@ -116,6 +112,12 @@ impl CallConnection {
                 .await
                 .broadcast(&broadcast.to_json().to_string());
         }
+
+        // Add member to group
+        {
+            group.lock().await.add_member(uid, self.tx.clone());
+        }
+
         // Build response
         let mut response = CommunicationValue::new(CommunicationType::identification_response)
             .with_id(cv.get_id().clone());
@@ -129,7 +131,6 @@ impl CallConnection {
                 let _ = users.insert(&caller.user_id.to_string(), user_info);
             }
         }
-
         response = response.add_data(DataTypes::about, users);
         self.send_message(&response).await;
     }
@@ -237,6 +238,12 @@ impl CallConnection {
         let (uid, cid) = { (*self.user_id.read().await, *self.call_id.read().await) };
         if let (Some(uid), Some(cid)) = (uid, cid) {
             if let Some(group) = call_manager::get_group(cid).await {
+                group.lock().await.broadcast(
+                    &CommunicationValue::new(CommunicationType::client_disconnected)
+                        .add_data_str(DataTypes::user_id, uid.to_string())
+                        .to_json()
+                        .to_string(),
+                );
                 group.lock().await.remove_member(uid);
             }
             call_manager::remove_inactive().await;
@@ -245,5 +252,7 @@ impl CallConnection {
         let mut session = self.sender.write().await;
         let _ = session.close(None).await;
     }
-    pub async fn handle_close(&self) {}
+    pub async fn handle_close(&self) {
+        self.close().await;
+    }
 }

src/data/communication.rs

@@ -187,6 +187,7 @@ pub enum CommunicationType {
     iota_closed,
     client_changed,
     client_connected,
+    client_disconnected,
     client_closed,
     public_key,
     private_key,
@@ -237,6 +238,7 @@ impl CommunicationType {
             "iotaclosed" => CommunicationType::iota_closed,
             "clientchanged" => CommunicationType::client_changed,
             "clientconnected" => CommunicationType::client_connected,
+            "clientdisconnected" => CommunicationType::client_disconnected,
             "clientclosed" => CommunicationType::client_closed,
             "publickey" => CommunicationType::public_key,
             "privatekey" => CommunicationType::private_key,

src/main.rs

@@ -19,19 +19,22 @@ use crate::{
     calls::call_connection::CallConnection,
     omega::omega_connection::OmegaConnection,
     rho::{client_connection::ClientConnection, iota_connection::IotaConnection},
+    util::config_util::CONFIG,
     util::print::{PrintType, line, line_err, print_start_message},
 };
-#[tokio::main]
 
+#[tokio::main]
 async fn main() {
     print_start_message();
+
     tokio::spawn(async move {
         OmegaConnection::new().connect().await;
     });
-    let listener = TcpListener::bind("0.0.0.0:959").await.unwrap();
+    let address = format!("{}:{}", &CONFIG.read().await.ip, &CONFIG.read().await.port);
+    let listener = TcpListener::bind(&address).await.unwrap();
     line(
-        PrintType::OmegaIn,
+        PrintType::General,
-        "WebSocket server listening on 0.0.0.0:959",
+        &format!("WebSocket server listening on {}", &address),
     );
 
     while let Ok((stream, _)) = listener.accept().await {

src/omega/omega_connection.rs

@@ -57,7 +57,7 @@ impl OmegaConnection {
                     let identify_msg = CommunicationValue::new(CommunicationType::identification)
                         .add_data(
                             DataTypes::uuid,
-                            JsonValue::String(CONFIG.lock().await.omikron_id.to_string()),
+                            JsonValue::String(CONFIG.read().await.omikron_id.to_string()),
                         );
                     self.send_message(&identify_msg).await;
 

src/util/config_util.rs

@@ -1,10 +1,7 @@
 use crate::util::file_util::load_file;
-use crate::util::print::PrintType;
-use crate::util::print::line;
-use crate::util::print::line_err;
-use futures::lock::Mutex;
 use once_cell::sync::Lazy;
 use serde::{Deserialize, Serialize};
+use tokio::sync::RwLock;
 use uuid::Uuid;
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Config {
@@ -20,9 +17,9 @@ pub struct Config {
 impl Default for Config {
     fn default() -> Self {
         Self {
-            omega_server: "omega.tensamin.methanium.net".into(),
+            omega_server: "omega.tensamin.net".into(),
-            auth_server: "auth.tensamin.methanium.net".into(),
+            auth_server: "auth.tensamin.net".into(),
-            omikron_id: Uuid::parse_str("a9e92dd6-08a6-4765-abf1-9fa39d0a99f9").unwrap_or_default(),
+            omikron_id: Uuid::parse_str("00000000-0000-0000-0000-000000000000").unwrap_or_default(),
             keep_people_stored_for: 90,
             max_data: 1000 * 1000 * 1000 * 8,
             ip: "0.0.0.0".into(),
@@ -31,7 +28,7 @@ impl Default for Config {
     }
 }
 
-pub static CONFIG: Lazy<Mutex<Config>> = Lazy::new(|| Mutex::new(Config::load()));
+pub static CONFIG: Lazy<RwLock<Config>> = Lazy::new(|| RwLock::new(Config::load()));
 
 impl Config {
     pub fn load() -> Self {
@@ -41,18 +38,21 @@ impl Config {
         }
 
         let json = json::parse(&content).unwrap();
-        line(PrintType::ClientIn, &format!("{:?}", json));
-        line(PrintType::ClientIn, &format!("{:?}", json["omikron_id"]));
         Self {
-            omega_server: json["omega_server"].as_str().unwrap_or_default().into(),
+            omega_server: json["omega_server"]
-            auth_server: json["auth_server"].as_str().unwrap_or_default().into(),
+                .as_str()
+                .unwrap_or("omega.tensamin.net")
+                .into(),
+            auth_server: json["auth_server"]
+                .as_str()
+                .unwrap_or("auth.tensamin.net")
+                .into(),
             omikron_id: Uuid::parse_str(json["omikron_id"].as_str().unwrap_or_default())
                 .unwrap_or_default(),
-            keep_people_stored_for: json["keep_people_stored_for"].as_i64().unwrap_or_default()
+            keep_people_stored_for: json["keep_people_stored_for"].as_i64().unwrap_or(90) as i32,
-                as i32,
+            max_data: json["max_data"].as_u64().unwrap_or(8000000000),
-            max_data: json["max_data"].as_u64().unwrap_or_default(),
+            ip: json["ip"].as_str().unwrap_or("0.0.0.0").into(),
-            ip: json["ip"].as_str().unwrap_or_default().into(),
+            port: json["port"].as_u64().unwrap_or(959) as u16,
-            port: json["port"].as_u64().unwrap_or_default() as u16,
         }
     }
 }