Omikron Connection

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/server/omikron_connection.rs

@@ -1,5 +1,11 @@
 use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes};
+use crate::sql::iota_omikron_tracker::{
+    track_iota_omikron, untrack_by_omikron as untrack_iota_by_omikron, untrack_iota,
+};
 use crate::sql::sql::get_omikron_by_id;
+use crate::sql::user_online_tracker::{
+    track_user_omikron, untrack_by_omikron as untrack_user_by_omikron, untrack_user,
+};
 use crate::util::crypto_helper::encrypt;
 use crate::util::logger::PrintType;
 use crate::{get_private_key, log_out};
@@ -204,6 +210,53 @@ impl OmikronConnection {
             self.close().await;
             return;
         }
+
+        let omikron_id = self.get_user_id().await;
+        if cv.is_type(CommunicationType::user_connected) {
+            if let Some(user_id) = cv.get_data(DataTypes::user_id).and_then(|v| v.as_i64()) {
+                track_user_omikron(user_id, omikron_id).await;
+            }
+            return;
+        }
+        if cv.is_type(CommunicationType::user_disconnected) {
+            if let Some(user_id) = cv.get_data(DataTypes::user_id).and_then(|v| v.as_i64()) {
+                untrack_user(user_id).await;
+            }
+            return;
+        }
+        if cv.is_type(CommunicationType::iota_connected) {
+            if let Some(iota_id) = cv.get_data(DataTypes::iota_id).and_then(|v| v.as_i64()) {
+                track_iota_omikron(iota_id, omikron_id).await;
+            }
+            return;
+        }
+        if cv.is_type(CommunicationType::iota_disconnected) {
+            if let Some(iota_id) = cv.get_data(DataTypes::iota_id).and_then(|v| v.as_i64()) {
+                untrack_iota(iota_id).await;
+            }
+            return;
+        }
+        if cv.is_type(CommunicationType::sync_client_iota_status) {
+            if let Some(json::JsonValue::Array(user_ids)) =
+                cv.get_data(DataTypes::user_ids).cloned()
+            {
+                for user_id_json in user_ids {
+                    if let Some(user_id) = user_id_json.as_i64() {
+                        track_user_omikron(user_id, omikron_id).await;
+                    }
+                }
+            }
+            if let Some(json::JsonValue::Array(iota_ids)) =
+                cv.get_data(DataTypes::iota_ids).cloned()
+            {
+                for iota_id_json in iota_ids {
+                    if let Some(iota_id) = iota_id_json.as_i64() {
+                        track_iota_omikron(iota_id, omikron_id).await;
+                    }
+                }
+            }
+            return;
+        }
     }
 
     async fn send_error_response(&self, message_id: &Uuid, error_type: CommunicationType) {
@@ -216,7 +269,11 @@ impl OmikronConnection {
     }
     pub async fn handle_close(self: Arc<Self>) {
         if self.is_identified().await {
-            if self.get_user_id().await != 0 {}
+            let omikron_id = self.get_user_id().await;
+            if omikron_id != 0 {
+                untrack_iota_by_omikron(omikron_id).await;
+                untrack_user_by_omikron(omikron_id).await;
+            }
         }
     }
 

src/sql/iota_omikron_tracker.rs

@@ -20,3 +20,8 @@ pub async fn untrack_iota(iota: i64) {
     let mut c = IOTA_OMIKRON_MAP.write().await;
     c.remove(&iota);
 }
+
+pub async fn untrack_by_omikron(omikron: i64) {
+    let mut c = IOTA_OMIKRON_MAP.write().await;
+    c.retain(|_, v| *v != omikron);
+}

src/sql/mod.rs

@@ -1,2 +1,3 @@
 pub mod iota_omikron_tracker;
 pub mod sql;
+pub mod user_online_tracker;

src/sql/user_online_tracker.rs

@@ -0,0 +1,27 @@
+use once_cell::sync::Lazy;
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
+static USER_OMIKRON_MAP: Lazy<Arc<RwLock<HashMap<i64, i64>>>> =
+    Lazy::new(|| Arc::new(RwLock::new(HashMap::new())));
+
+pub async fn track_user_omikron(user: i64, omikron: i64) {
+    let mut c = USER_OMIKRON_MAP.write().await;
+    c.insert(user, omikron);
+}
+
+pub async fn get_omikron_for_user(user: i64) -> Option<i64> {
+    let c = USER_OMIKRON_MAP.read().await;
+    c.get(&user).cloned()
+}
+
+pub async fn untrack_user(user: i64) {
+    let mut c = USER_OMIKRON_MAP.write().await;
+    c.remove(&user);
+}
+
+pub async fn untrack_by_omikron(omikron: i64) {
+    let mut c = USER_OMIKRON_MAP.write().await;
+    c.retain(|_, v| *v != omikron);
+}