tensamin/omikron
Watch
6
0
Fork
You've already forked omikron
 0
Code Issues Pull requests 3 Projects Releases Packages Wiki Activity Actions

Calling Logging fixing

Browse source
This commit is contained in:
Alex Emmet 2025-11-13 23:13:55 +00:00
commit 3cb2f0ec5e
12 changed files with 118 additions and 112 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/auth/auth_connector.rs
View file

@ -132,7 +132,7 @@ pub async fn get_public_key(user_id: Uuid) -> Option<String> {
.ok()?;
let body = res.text().await.ok()?;
let mut cv = CommunicationValue::from_json(&body);
let cv = CommunicationValue::from_json(&body);
if cv.comm_type != CommunicationType::message_send {
return None;

18
src/calls/call_connection.rs
View file

@ -1,10 +1,7 @@
use async_tungstenite::{
WebSocketReceiver, WebSocketSender, WebSocketStream, tungstenite::Message,
};
use futures::SinkExt;
use async_tungstenite::{WebSocketReceiver, WebSocketSender, tungstenite::Message};
use std::sync::Arc;
use tokio::sync::{
Mutex, RwLock,
RwLock,
mpsc::{UnboundedSender, unbounded_channel},
};
use tokio_util::compat::Compat;
@ -101,6 +98,15 @@ impl CallConnection {
}
let group = call_manager::get_or_create_group(cid, secret_sha).await;
if let None = group {
self.send_message(&CommunicationValue::new(
CommunicationType::error_invalid_secret,
))
.await;
return;
}
let group = group.unwrap();
// Build broadcast
let broadcast = CommunicationValue::new(CommunicationType::client_connected)
.with_id(cv.get_id().clone())
@ -113,7 +119,6 @@ impl CallConnection {
.broadcast(&broadcast.to_json().to_string());
}
// Add member to group
{
group.lock().await.add_member(uid, self.tx.clone());
}
@ -244,6 +249,7 @@ impl CallConnection {
.to_json()
.to_string(),
);
group.lock().await.get_member(uid).await;
group.lock().await.remove_member(uid);
}
call_manager::remove_inactive().await;

29
src/calls/call_group.rs
View file

@ -1,24 +1,30 @@
use crate::calls::caller::Caller;
use std::collections::HashMap;
use std::{collections::HashMap, sync::Arc};
use tokio::sync::mpsc::UnboundedSender;
use tungstenite::Utf8Bytes;
use uuid::Uuid;
pub struct CallGroup {
pub call_id: Uuid,
pub callers: HashMap<Uuid, Caller>,
pub callers: HashMap<Uuid, Arc<Caller>>,
pub secret_hash: String,
}
impl CallGroup {
pub fn new(call_id: Uuid) -> Self {
pub fn new(secret_hash: String) -> Self {
Self {
call_id,
callers: HashMap::new(),
secret_hash: secret_hash,
}
}
pub fn add_member(&mut self, user_id: Uuid, tx: UnboundedSender<Utf8Bytes>) {
self.callers.insert(user_id, Caller { user_id, tx });
self.callers
.insert(user_id, Arc::new(Caller::new(user_id, tx)));
}
pub fn disconnect_member(&mut self, user_id: Uuid) {
if let Some(caller) = self.callers.get(&user_id) {
caller.disconnect();
}
}
pub fn remove_member(&mut self, user_id: Uuid) {
@ -26,7 +32,12 @@ impl CallGroup {
}
pub fn is_empty(&self) -> bool {
self.callers.is_empty()
for caller in self.callers.values() {
if !caller.is_connected() {
return false;
}
}
true
}
pub fn send_to(&self, user_id: &Uuid, message: &str) {
@ -41,7 +52,7 @@ impl CallGroup {
}
}
pub fn caller_state_mut(&mut self, user_id: &Uuid) -> Option<&mut Caller> {
self.callers.get_mut(user_id)
pub fn caller_state_mut(&mut self, user_id: &Uuid) -> Option<&Arc<Caller>> {
self.callers.get(user_id)
}
}

17
src/calls/call_manager.rs
View file

@ -4,12 +4,8 @@ use once_cell::sync::Lazy;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tokio::sync::mpsc::UnboundedSender;
use tungstenite::Utf8Bytes;
use uuid::Uuid;
pub type Tx = UnboundedSender<Utf8Bytes>;
pub static CALL_GROUPS: Lazy<RwLock<Mutex<HashMap<Uuid, Arc<Mutex<CallGroup>>>>>> =
Lazy::new(|| RwLock::new(Mutex::new(HashMap::new())));
@ -22,14 +18,18 @@ pub async fn get_group(call_id: Uuid) -> Option<Arc<Mutex<CallGroup>>> {
.get_mut(&call_id)
.cloned()
}
pub async fn get_or_create_group(call_id: Uuid, secret: &str) -> Arc<Mutex<CallGroup>> {
pub async fn get_or_create_group(call_id: Uuid, secret: &str) -> Option<Arc<Mutex<CallGroup>>> {
let g = CALL_GROUPS.write().await;
if let Some(group) = g.lock().await.get_mut(&call_id) {
group.clone()
if group.lock().await.secret_hash.eq(secret) {
Some(group.clone())
} else {
None
}
} else {
let cg = Arc::new(Mutex::new(CallGroup::new(call_id)));
let cg = Arc::new(Mutex::new(CallGroup::new(secret.to_string())));
g.lock().await.insert(call_id, cg.clone());
cg
Some(cg)
}
}
@ -45,7 +45,6 @@ pub async fn remove_inactive() {
.unwrap()
.lock()
.await
.callers
.is_empty()
{
rem.push(cg.clone());

43
src/calls/caller.rs
View file

@ -1,14 +1,49 @@
use std::sync::Arc;
use futures::lock::Mutex;
use tokio::sync::mpsc::UnboundedSender;
use tungstenite::Utf8Bytes;
use uuid::Uuid;
#[derive(Debug, Clone)]
pub struct Caller {
pub user_id: Uuid,
pub tx: UnboundedSender<Utf8Bytes>,
pub user_id: Mutex<Uuid>,
pub tx: Mutex<Option<UnboundedSender<Utf8Bytes>>>,
pub user_state: Mutex<CallUserState>,
pub streaming: Mutex<bool>,
}
#[derive(Clone)]
pub enum CallUserState {
Active,
Muted,
Deafed,
Disconnected,
}
impl Caller {
pub fn new(user_id: Uuid, tx: UnboundedSender<Utf8Bytes>) -> Self {
Self {
user_id: Mutex::new(user_id),
tx: Mutex::new(Some(tx)),
user_state: Mutex::new(CallUserState::Active),
streaming: Mutex::new(false),
}
}
pub fn send(&self, msg: impl Into<Utf8Bytes>) {
let _ = self.tx.send(msg.into());
if let Some(tx) = &self.tx.lock().await {
let _ = tx.send(msg.into());
}
}
pub fn disconnect(self: Arc<Self>) {
self.user_state = CallUserState::Disconnected;
self.tx = None;
}
pub fn is_connected(&self) -> bool {
if let CallUserState::Disconnected = self.user_state {
false
} else {
true
}
}
}

14
src/data/communication.rs
View file

@ -6,6 +6,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
use uuid::Uuid;
#[derive(Eq, Hash, PartialEq, Clone, Debug)]
#[allow(non_camel_case_types, dead_code)]
pub enum DataTypes {
error_type,
accepted_ids,
@ -153,10 +154,14 @@ impl DataTypes {
}
#[derive(PartialEq, Clone, Debug)]
#[allow(non_camel_case_types, dead_code)]
pub enum CommunicationType {
error,
error_invalid_user_id,
error_not_found,
error_no_iota,
error_invalid_challenge,
error_invalid_secret,
error_invalid_private_key,
success,
message,
@ -267,6 +272,7 @@ pub struct CommunicationValue {
pub data: HashMap<DataTypes, JsonValue>,
}
#[allow(dead_code)]
impl CommunicationValue {
pub fn new(comm_type: CommunicationType) -> Self {
Self {
@ -366,14 +372,6 @@ impl CommunicationValue {
data,
}
}
pub fn ack_message(message_id: Uuid, sender: Uuid) -> CommunicationValue {
let mut cv = CommunicationValue::new(CommunicationType::message).with_id(message_id);
let s = sender;
cv = cv.add_data(DataTypes::send_time, JsonValue::String(s.to_string()));
cv
}
pub fn forward_to_other_iota(original: &mut CommunicationValue) -> CommunicationValue {
let receiver = Uuid::from_str(
&*original

7
src/main.rs
View file

@ -7,13 +7,10 @@ mod util;
use async_tungstenite::accept_hdr_async;
use futures::StreamExt;
use std::{sync::Arc, time::Duration};
use std::sync::Arc;
use tokio::net::TcpListener;
use tokio_util::compat::TokioAsyncReadCompatExt;
use tungstenite::{
Message, Utf8Bytes,
handshake::server::{Request, Response},
};
use tungstenite::handshake::server::{Request, Response};
use crate::{
calls::call_connection::CallConnection,

6
src/omega/omega_connection.rs
View file

@ -3,7 +3,7 @@ use std::time::Duration;
use crate::data::communication::{CommunicationType, CommunicationValue};
use crate::util::print::PrintType;
use crate::util::print::line_err;
use crate::util::print::{line, line_err};
use crate::{
data::{
communication::DataTypes,
@ -14,10 +14,9 @@ use crate::{
};
use async_tungstenite::tungstenite::protocol::Message;
use dashmap::DashMap;
use futures::{SinkExt, StreamExt};
use futures::StreamExt;
use json::JsonValue;
use once_cell::sync::Lazy;
use tokio::io::unix::AsyncFd;
use tokio::sync::Mutex;
use tokio::time::sleep;
use tokio_util::compat::Compat;
@ -141,6 +140,7 @@ impl OmegaConnection {
pub async fn send_message(&self, cv: &CommunicationValue) {
let mut guard = self.ws_stream.lock().await;
if let Some(ws) = guard.as_mut() {
line(PrintType::OmegaOut, &cv.to_json().to_string());
let _ = ws
.send(Message::Text(Utf8Bytes::from(cv.to_json().to_string())))
.await;

42
src/rho/client_connection.rs
View file

@ -1,14 +1,13 @@
use async_tungstenite::tungstenite::Message;
use async_tungstenite::{WebSocketReceiver, WebSocketSender};
use async_tungstenite::{WebSocketStream, tungstenite::Message};
use futures::SinkExt;
use std::sync::{Arc, Weak};
use tokio::sync::Mutex;
use tokio::sync::RwLock;
use tokio_util::compat::Compat;
use tungstenite::Utf8Bytes;
use uuid::Uuid;
use super::{rho_connection::RhoConnection, rho_manager};
use crate::calls::call_manager;
use crate::util::print::PrintType;
use crate::util::print::line;
use crate::util::print::line_err;
@ -154,7 +153,7 @@ impl ClientConnection {
}
/// Handle identification message
async fn handle_identification(&self, sarc: Arc<ClientConnection>, mut cv: CommunicationValue) {
async fn handle_identification(&self, sarc: Arc<ClientConnection>, cv: CommunicationValue) {
// Extract user ID
let user_id = match cv.get_data(DataTypes::user_id) {
Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
@ -228,7 +227,7 @@ impl ClientConnection {
}
/// Handle ping message
async fn handle_ping(&self, mut cv: CommunicationValue) {
async fn handle_ping(&self, cv: CommunicationValue) {
// Update our ping if provided
if let Some(last_ping) = cv.get_data(DataTypes::last_ping) {
if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
@ -253,9 +252,9 @@ impl ClientConnection {
}
/// Handle client status change
async fn handle_client_changed(&self, mut cv: CommunicationValue) {
async fn handle_client_changed(&self, cv: CommunicationValue) {
if let Some(user_id) = self.get_user_id().await {
if let Some(status_str) = cv.get_data(DataTypes::user_state) {
if let Some(_status_str) = cv.get_data(DataTypes::user_state) {
// Parse user status - this would need to be implemented properly
let user_status = UserStatus::online; // placeholder
if let Some(rho_conn) = self.get_rho_connection().await {
@ -359,7 +358,7 @@ impl ClientConnection {
}
/// Handle get call request
async fn handle_get_call(&self, mut cv: CommunicationValue) {
async fn handle_get_call(&self, cv: CommunicationValue) {
let user_id = match self.get_user_id().await {
Some(id) => id,
None => return,
@ -396,24 +395,25 @@ impl ClientConnection {
.with_id(cv.get_id())
.with_receiver(user_id);
// Placeholder call group logic
// if let Some(call_group) = CallManager::get_call_group(call_id, &call_secret_sha.unwrap(), true).await {
// response = response
// .add_data_str(DataTypes::call_state, call_group.call_state.to_string())
// .add_data_str(DataTypes::start_date, call_group.started_at.to_string());
//
// if call_group.ended_at != 0 {
// response = response.add_data_str(DataTypes::end_date, call_group.ended_at.to_string());
// }
// } else {
response = response.add_data_str(DataTypes::call_state, "DESTROYED".to_string());
// }
if let Some(_call_group) = call_manager::get_group(call_id).await {
/*response = response
.add_data_str(DataTypes::call_state, call_group.call_state.to_string())
.add_data_str(DataTypes::start_date, call_group.started_at.to_string());
if call_group.lock(). != 0 {
response =
response.add_data_str(DataTypes::end_date, call_group.ended_at.to_string());
}
*/
} else {
response = response.add_data_str(DataTypes::call_state, "DESTROYED".to_string());
}
self.send_message(&response).await;
}
/// Forward message to Iota
async fn forward_to_iota(&self, mut cv: CommunicationValue) {
async fn forward_to_iota(&self, cv: CommunicationValue) {
if let Some(user_id) = self.get_user_id().await {
if let Some(rho_conn) = self.get_rho_connection().await {
let updated_cv = cv.with_sender(user_id);

16
src/rho/iota_connection.rs
View file

@ -1,19 +1,15 @@
use crate::util::print::PrintType;
use crate::util::print::line;
use crate::util::print::line_err;
use ansi_term::Color;
use async_tungstenite::WebSocketReceiver;
use async_tungstenite::WebSocketSender;
use async_tungstenite::tungstenite::protocol::WebSocketConfig;
use async_tungstenite::{WebSocketStream, tungstenite::Message};
use futures::FutureExt;
use futures::SinkExt;
use async_tungstenite::tungstenite::Message;
use json::JsonValue;
use std::{
collections::HashMap,
sync::{Arc, Weak},
};
use tokio::sync::{Mutex, RwLock};
use tokio::sync::RwLock;
use tokio_util::compat::Compat;
use tungstenite::Utf8Bytes;
use uuid::Uuid;
@ -280,12 +276,12 @@ impl IotaConnection {
}
/// Handle GET_CHATS message
async fn handle_get_chats(&self, mut cv: CommunicationValue) {
async fn handle_get_chats(&self, cv: CommunicationValue) {
let receiver_id = cv.get_receiver();
let mut interested_ids: Vec<Uuid> = Vec::new();
let interested_ids: Vec<Uuid> = Vec::new();
// Process contacts and add call information
if let Some(contacts_data) = cv.get_data(DataTypes::user_ids) {
if let Some(_contacts_data) = cv.get_data(DataTypes::user_ids) {
// Parse contacts JSON array and enrich with call data
// This would need proper JSON parsing implementation
// For now, placeholder logic:
@ -321,7 +317,7 @@ impl IotaConnection {
async fn forward_to_client(&self, cv: CommunicationValue) {
if let Some(rho_conn) = self.get_rho_connection().await {
let updated_cv = cv.with_sender(self.get_iota_id().await);
let receiver_id = updated_cv.get_receiver();
let _receiver_id = updated_cv.get_receiver();
rho_conn.message_to_client(updated_cv).await;
}
}

3
src/rho/rho_connection.rs
View file

@ -4,9 +4,6 @@ use crate::data::{
user::UserStatus,
};
use crate::omega::omega_connection::OmegaConnection;
use crate::util::print::PrintType;
use crate::util::print::line;
use crate::util::print::line_err;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

33
src/rho/rho_manager.rs
View file

@ -68,36 +68,3 @@ pub async fn connection_count() -> usize {
let connections = RHO_CONNECTIONS.read().await;
connections.len()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::rho::iota_connection::IotaConnection;
use std::sync::Arc;
use tokio::sync::Mutex;
#[tokio::test]
async fn test_add_and_get_rho() {
// Clear any existing connections
{
let mut connections = RHO_CONNECTIONS.write().await;
connections.clear();
}
let iota_id = Uuid::new_v4();
let user_ids = vec![Uuid::new_v4(), Uuid::new_v4()];
// Skip this test due to WebSocket complexity - would require proper mock setup
return;
// This test would need proper WebSocket stream mocking:
// let mock_session = create_mock_websocket_stream();
// let iota_conn = IotaConnection::new_with_ids(iota_id, user_ids.clone(), mock_session);
// let rho_conn = Arc::new(RhoConnection::new(iota_conn, user_ids.clone()));
// Test assertions would go here:
// add_rho(Arc::clone(&rho_conn)).await;
// assert!(contains_iota(iota_id).await);
// etc.
}
}