From 57658197f952610b9ff6304dd2496878cb28918e Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Tue, 6 Jan 2026 22:16:32 +0100 Subject: [PATCH] Omega Connection Logging system --- .gitignore | 1 + src/data/communication.rs | 24 +++++++++---- src/omega/omega_connection.rs | 67 ++++++++++++++++++++++++----------- src/rho/rho_manager.rs | 1 - 4 files changed, 65 insertions(+), 28 deletions(-) diff --git a/.gitignore b/.gitignore index dff8ceb..1baef92 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,4 @@ target # Added by cargo /target +/logs diff --git a/src/data/communication.rs b/src/data/communication.rs index 7af1340..58fd957 100644 --- a/src/data/communication.rs +++ b/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, } diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index 0fdcaaf..989dcc6 100644 --- a/src/omega/omega_connection.rs +++ b/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, - rho::rho_manager, + get_private_key, log, log_in, log_out, + 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> = Lazy::new(|| { conn }); +#[allow(dead_code)] pub fn get_omega_connection() -> Arc { 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 = Vec::new(); + let mut connected_user_ids: Vec = 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) { - let user_ids_str = user_ids - .iter() - .map(|id| id.to_string()) - .collect::>() - .join(","); + pub async fn connect_iota(iota_id: i64, _user_ids: Vec) { let cv = CommunicationValue::new(CommunicationType::iota_connected) - .add_data(DataTypes::iota_id, JsonValue::from(iota_id.to_string())) - .add_data(DataTypes::user_ids, JsonValue::from(user_ids_str)); + .add_data(DataTypes::iota_id, JsonValue::from(iota_id)); OmegaConnection::send_global(cv).await; } pub async fn close_iota(iota_id: i64) { - let cv = CommunicationValue::new(CommunicationType::iota_closed) - .add_data(DataTypes::iota_id, JsonValue::from(iota_id.to_string())); + let cv = CommunicationValue::new(CommunicationType::iota_disconnected) + .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) { - let cv = CommunicationValue::new(CommunicationType::client_changed) - .add_data(DataTypes::iota_id, JsonValue::from(iota_id)) - .add_data(DataTypes::user_id, JsonValue::from(user_id)) - .add_data(DataTypes::user_state, JsonValue::from(state.to_string())); - OmegaConnection::send_global(cv).await; + pub async fn client_changed(_iota_id: i64, user_id: i64, state: UserStatus) { + let msg_type = match state { + UserStatus::iota_offline => Some(CommunicationType::user_disconnected), + UserStatus::user_offline => Some(CommunicationType::user_disconnected), + _ => Some(CommunicationType::user_connected), + }; + + 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) { diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index fb9d74a..f2b41d3 100644 --- a/src/rho/rho_manager.rs +++ b/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,