Omega Connection Logging system

.gitignore

@@ -16,3 +16,4 @@ target
 
 # Added by cargo
 /target
+/logs

src/data/communication.rs

@@ -16,6 +16,7 @@ pub enum DataTypes {
     iota_id,
     user_id,
     user_ids,
+    iota_ids,
     user_state,
     user_states,
     user_pings,
@@ -75,7 +76,7 @@ pub enum DataTypes {
     challenge,
     community_title,
     communities,
-
+    rho_connections,
     user,
 }
 
@@ -92,6 +93,7 @@ impl DataTypes {
             "iotaid" => DataTypes::iota_id,
             "userid" => DataTypes::user_id,
             "userids" => DataTypes::user_ids,
+            "iotaids" => DataTypes::iota_ids,
             "userstate" => DataTypes::user_state,
             "userstates" => DataTypes::user_states,
             "userpings" => DataTypes::user_pings,
@@ -151,7 +153,7 @@ impl DataTypes {
             "challenge" => DataTypes::challenge,
             "communitytitle" => DataTypes::community_title,
             "communities" => DataTypes::communities,
-
+            "rhoconnections" => DataTypes::rho_connections,
             "user" => DataTypes::user,
             _ => DataTypes::error_type, // fallback if unknown
         }
@@ -202,8 +204,6 @@ pub enum CommunicationType {
     pong,
     add_chat,
     send_chat,
-    iota_connected,
-    iota_closed,
     client_changed,
     client_connected,
     client_disconnected,
@@ -224,6 +224,13 @@ pub enum CommunicationType {
     function,
     update,
     create_user,
+    rho_update,
+
+    user_connected,
+    user_disconnected,
+    iota_connected,
+    iota_disconnected,
+    sync_client_iota_status,
 }
 impl CommunicationType {
     pub fn parse(p0: String) -> CommunicationType {
@@ -280,8 +287,6 @@ impl CommunicationType {
             "pong" => CommunicationType::pong,
             "addchat" => CommunicationType::add_chat,
             "sendchat" => CommunicationType::send_chat,
-            "iotaconnected" => CommunicationType::iota_connected,
-            "iotaclosed" => CommunicationType::iota_closed,
             "clientchanged" => CommunicationType::client_changed,
             "clientconnected" => CommunicationType::client_connected,
             "clientdisconnected" => CommunicationType::client_disconnected,
@@ -292,6 +297,13 @@ impl CommunicationType {
             "webrtcice" => CommunicationType::webrtc_ice,
             "startstream" => CommunicationType::start_stream,
             "endstream" => CommunicationType::end_stream,
+            "rhoupdate" => CommunicationType::rho_update,
+
+            "iotaconnected" => CommunicationType::iota_connected,
+            "iotadisconnected" => CommunicationType::iota_disconnected,
+            "userconnected" => CommunicationType::user_connected,
+            "userdisconnected" => CommunicationType::user_disconnected,
+            "syncclientiotastatus" => CommunicationType::sync_client_iota_status,
 
             _ => CommunicationType::error,
         }

src/omega/omega_connection.rs

@@ -1,5 +1,5 @@
 use async_tungstenite::{
-    WebSocketReceiver, WebSocketSender, WebSocketStream,
+    WebSocketReceiver, WebSocketSender,
     stream::Stream,
     tokio::{TokioAdapter, connect_async},
     tungstenite::protocol::Message,
@@ -23,8 +23,8 @@ use crate::{
         communication::{CommunicationType, CommunicationValue, DataTypes},
         user::UserStatus,
     },
-    get_private_key, log_in, log_out,
+    get_private_key, log, log_in, log_out,
-    rho::rho_manager,
+    rho::rho_manager::{self, RHO_CONNECTIONS},
     util::logger::PrintType,
 };
 use crate::{auth::crypto_helper::secret_key_to_base64, log_err};
@@ -46,6 +46,7 @@ static OMEGA_CONNECTION: Lazy<Arc<OmegaConnection>> = Lazy::new(|| {
     conn
 });
 
+#[allow(dead_code)]
 pub fn get_omega_connection() -> Arc<OmegaConnection> {
     OMEGA_CONNECTION.clone()
 }
@@ -119,6 +120,7 @@ impl OmegaConnection {
                     let cloned_self = self.clone();
                     tokio::spawn(async move {
                         let id = Uuid::new_v4();
+
                         let identify_msg =
                             CommunicationValue::new(CommunicationType::identification)
                                 .with_id(id)
@@ -182,7 +184,7 @@ impl OmegaConnection {
                                         let response_id = response_msg.get_id();
                                         WAITING_TASKS.insert(
                                             response_id,
-                                            Box::new(|_self, final_cv| {
+                                            Box::new(|selfc, final_cv| {
                                                 if !final_cv
                                                     .is_type(CommunicationType::identification_response)
                                                 {
@@ -193,7 +195,30 @@ impl OmegaConnection {
                                                     return false;
                                                 }
 
-                                                log_err!(
+                                                tokio::spawn(async move {
+                                                    let mut connected_iota_ids: Vec<JsonValue> = Vec::new();
+                                                    let mut connected_user_ids: Vec<JsonValue> = Vec::new();
+                                                    let rho_connections_reader = RHO_CONNECTIONS.read().await;
+
+                                                    for iota_id in rho_connections_reader.keys() {
+                                                        connected_iota_ids.push(JsonValue::from(*iota_id));
+                                                    }
+
+                                                    for rho in rho_connections_reader.values() {
+                                                        for client_conn in rho.get_client_connections().await {
+                                                            connected_user_ids.push(JsonValue::from(client_conn.get_user_id().await));
+                                                        }
+                                                    }
+
+                                                    drop(rho_connections_reader);
+
+                                                    let sync_msg = CommunicationValue::new(CommunicationType::sync_client_iota_status)
+                                                        .add_data(DataTypes::iota_ids, JsonValue::Array(connected_iota_ids))
+                                                        .add_data(DataTypes::user_ids, JsonValue::Array(connected_user_ids));
+
+                                                    selfc.send_message(&sync_msg).await;
+                                                });
+                                                log!(
                                                     PrintType::Omega,
                                                     "Successfully identified with Omega.",
                                                 );
@@ -313,30 +338,30 @@ impl OmegaConnection {
         }
     }
 
-    pub async fn connect_iota(iota_id: i64, user_ids: Vec<i64>) {
+    pub async fn connect_iota(iota_id: i64, _user_ids: Vec<i64>) {
-        let user_ids_str = user_ids
-            .iter()
-            .map(|id| id.to_string())
-            .collect::<Vec<_>>()
-            .join(",");
         let cv = CommunicationValue::new(CommunicationType::iota_connected)
-            .add_data(DataTypes::iota_id, JsonValue::from(iota_id.to_string()))
+            .add_data(DataTypes::iota_id, JsonValue::from(iota_id));
-            .add_data(DataTypes::user_ids, JsonValue::from(user_ids_str));
         OmegaConnection::send_global(cv).await;
     }
 
     pub async fn close_iota(iota_id: i64) {
-        let cv = CommunicationValue::new(CommunicationType::iota_closed)
+        let cv = CommunicationValue::new(CommunicationType::iota_disconnected)
-            .add_data(DataTypes::iota_id, JsonValue::from(iota_id.to_string()));
+            .add_data(DataTypes::iota_id, JsonValue::from(iota_id));
         OmegaConnection::send_global(cv).await;
     }
 
-    pub async fn client_changed(iota_id: i64, user_id: i64, state: UserStatus) {
+    pub async fn client_changed(_iota_id: i64, user_id: i64, state: UserStatus) {
-        let cv = CommunicationValue::new(CommunicationType::client_changed)
+        let msg_type = match state {
-            .add_data(DataTypes::iota_id, JsonValue::from(iota_id))
+            UserStatus::iota_offline => Some(CommunicationType::user_disconnected),
-            .add_data(DataTypes::user_id, JsonValue::from(user_id))
+            UserStatus::user_offline => Some(CommunicationType::user_disconnected),
-            .add_data(DataTypes::user_state, JsonValue::from(state.to_string()));
+            _ => Some(CommunicationType::user_connected),
-        OmegaConnection::send_global(cv).await;
+        };
+
+        if let Some(t) = msg_type {
+            let cv =
+                CommunicationValue::new(t).add_data(DataTypes::user_id, JsonValue::from(user_id));
+            OmegaConnection::send_global(cv).await;
+        }
     }
 
     pub async fn user_states(user_id: i64, user_ids: Vec<i64>) {

src/rho/rho_manager.rs

@@ -1,6 +1,5 @@
 use super::rho_connection::RhoConnection;
 use crate::log_in;
-use crate::log_out;
 use crate::util::logger::PrintType;
 use std::{
     collections::HashMap,