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

Old Call Logic

Browse source
This commit is contained in:
Alex Emmet 2025-11-24 18:00:48 +00:00
commit 7493bd4cb7
8 changed files with 1023 additions and 375 deletions
Download patch file Download diff file Expand all files Collapse all files

1217
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

5
Cargo.toml
View file

@ -5,7 +5,7 @@ edition = "2024"
[dependencies]
ansi_term = "*"
async-tungstenite = { version = "*" }
async-tungstenite = { version = "0.32.0", features = ["futures-03-sink", "futures-util", "handshake", "__rustls-tls", "async-native-tls", "async-std", "async-std-runtime", "async-tls", "gio", "gio-runtime", "glib", "openssl", "real-async-native-tls", "real-async-tls", "real-native-tls", "real-tokio-native-tls", "real-tokio-openssl", "real-tokio-rustls", "rustls-native-certs", "rustls-pki-types", "tokio", "tokio-native-tls", "tokio-openssl", "tokio-runtime", "tokio-rustls-manual-roots", "tokio-rustls-native-certs", "tokio-rustls-webpki-roots", "url", "verbose-logging", "webpki-roots" ] }
axum = "*"
base64 = "0.22.1"
bytes = "*"
@ -18,7 +18,7 @@ futures = "*"
futures-util = "*"
hex = "*"
http = "*"
hyper = { version = "*", features = ["full"] }
hyper = { version = "1.8.1", features = ["full"] }
json = "*"
loom = "0.7.2"
once_cell = "1.21.3"
@ -36,6 +36,7 @@ tokio-rustls = { version = "*" }
tokio-stream = "*"
tokio-util = { version = "*", features = ["full"] }
tokio_websocket_server = "0.1.0"
tower = "0.5.2"
tracing = "*"
tracing-subscriber = "*"
tungstenite = "*"

44
src/calls/call_connection.rs
View file

@ -1,5 +1,5 @@
use async_tungstenite::{WebSocketReceiver, WebSocketSender, tungstenite::Message};
use std::sync::Arc;
use std::{any::Any, sync::Arc};
use tokio::sync::{
RwLock,
mpsc::{UnboundedSender, unbounded_channel},
@ -56,7 +56,21 @@ impl CallConnection {
if let Some(sender) = *self.user_id.read().await {
cv = cv.with_sender(sender);
}
if !cv.is_type(CommunicationType::ping) {
if cv.is_type(CommunicationType::webrtc_sdp)
| cv.is_type(CommunicationType::webrtc_ice)
| cv.is_type(CommunicationType::watch_stream)
{
line(
PrintType::CallIn,
&format!(
"{:?} : {} -> {} ",
&cv.type_id(),
&cv.get_sender(),
&cv.get_data(DataTypes::receiver_id).unwrap()
),
);
} else if !cv.is_type(CommunicationType::ping) {
line(PrintType::CallIn, &cv.to_json().to_string());
}
@ -116,7 +130,8 @@ impl CallConnection {
group
.lock()
.await
.broadcast(&broadcast.to_json().to_string());
.broadcast(&broadcast.to_json().to_string())
.await;
}
{
@ -167,7 +182,7 @@ impl CallConnection {
let mut bc = cv.clone();
bc = bc.add_data(DataTypes::sender_id, JsonValue::String(uid.to_string()));
group.broadcast(&bc.to_json().to_string());
group.broadcast(&bc.to_json().to_string()).await;
}
}
@ -187,7 +202,7 @@ impl CallConnection {
let _streaming = cv.comm_type == CommunicationType::start_stream;
let mut bc = cv.clone();
bc = bc.add_data(DataTypes::sender_id, JsonValue::String(uid.to_string()));
group.broadcast(&bc.to_json().to_string());
group.broadcast(&bc.to_json().to_string()).await;
}
}
@ -208,7 +223,8 @@ impl CallConnection {
group
.lock()
.await
.send_to(&receiver_id, &bc.to_json().to_string());
.send_to(&receiver_id, &bc.to_json().to_string())
.await;
line(
PrintType::CallOut,
&format!("Forwarded WebRTC message to receiver: {}", receiver_str),
@ -243,12 +259,16 @@ 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
.broadcast(
&CommunicationValue::new(CommunicationType::client_disconnected)
.add_data_str(DataTypes::user_id, uid.to_string())
.to_json()
.to_string(),
)
.await;
group.lock().await.disconnect_member(uid).await;
}
call_manager::remove_inactive().await;

10
src/calls/call_group.rs
View file

@ -35,22 +35,22 @@ impl CallGroup {
pub async fn is_empty(&self) -> bool {
for caller in self.callers.values() {
if !caller.is_connected().await {
if caller.is_connected().await {
return false;
}
}
true
}
pub fn send_to(&self, user_id: &Uuid, message: &str) {
pub async fn send_to(&self, user_id: &Uuid, message: &str) {
if let Some(caller) = self.callers.get(user_id) {
let _ = caller.send(Utf8Bytes::from(message.to_string()));
caller.send(Utf8Bytes::from(message.to_string())).await;
}
}
pub fn broadcast(&self, message: &str) {
pub async fn broadcast(&self, message: &str) {
for caller in self.callers.values() {
let _ = caller.send(Utf8Bytes::from(message.to_string()));
caller.send(Utf8Bytes::from(message.to_string())).await;
}
}

94
src/calls/call_manager.rs
View file

@ -1,57 +1,87 @@
use crate::calls::call_group::CallGroup;
use futures::lock::Mutex;
use futures::lock::Mutex; // Used only to protect CallGroup contents
use once_cell::sync::Lazy;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tokio::sync::RwLock; // Used to protect the global HashMap
use uuid::Uuid;
pub static CALL_GROUPS: Lazy<RwLock<Mutex<HashMap<Uuid, Arc<Mutex<CallGroup>>>>>> =
Lazy::new(|| RwLock::new(Mutex::new(HashMap::new())));
// FIX 1: Simplify the global state. Only use RwLock to protect the HashMap.
// The inner Arc<Mutex<CallGroup>> protects the contents of each group.
pub static CALL_GROUPS: Lazy<RwLock<HashMap<Uuid, Arc<Mutex<CallGroup>>>>> =
Lazy::new(|| RwLock::new(HashMap::new()));
/// Retrieves a CallGroup by ID, if it exists.
pub async fn get_group(call_id: Uuid) -> Option<Arc<Mutex<CallGroup>>> {
// Acquire read lock (very fast, non-blocking for other readers)
CALL_GROUPS
.read()
.await
.lock()
.await
.get_mut(&call_id)
.get(&call_id) // Use get(), as we only need to clone the Arc
.cloned()
}
/// Retrieves an existing group or creates a new one, performing a secret check.
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) {
if group.lock().await.secret_hash.eq(secret) {
Some(group.clone())
// Phase 1: Check existence under a READ lock.
{
let groups = CALL_GROUPS.read().await;
if let Some(group_arc) = groups.get(&call_id) {
// Found it. Release the global READ lock and check the secret.
// This await on the group's lock only blocks access to that *specific* group.
let group_lock = group_arc.lock().await;
if group_lock.secret_hash.eq(secret) {
return Some(group_arc.clone());
} else {
return None; // Secret mismatch
}
}
} // READ lock automatically released here.
// Phase 2: Not found, acquire a WRITE lock to create (only held for insertion).
let mut groups = CALL_GROUPS.write().await;
// FIX 2: Double-check (race condition prevention) - A group might have been created
// between the read lock release and the write lock acquisition.
if let Some(group_arc) = groups.get(&call_id) {
// Already created by another task, check secret again.
let group_lock = group_arc.lock().await;
if group_lock.secret_hash.eq(secret) {
return Some(group_arc.clone());
} else {
None
return None;
}
} else {
let cg = Arc::new(Mutex::new(CallGroup::new(secret.to_string())));
g.lock().await.insert(call_id, cg.clone());
Some(cg)
}
// Still not found, proceed with creation.
let new_group = Arc::new(Mutex::new(CallGroup::new(secret.to_string())));
groups.insert(call_id, new_group.clone());
// WRITE lock automatically released here.
Some(new_group)
}
/// Removes inactive groups.
pub async fn remove_inactive() {
// Collect keys to remove under a read lock first.
let groups_to_check: Vec<Uuid> = CALL_GROUPS.read().await.keys().cloned().collect();
let mut rem = Vec::new();
for cg in CALL_GROUPS.read().await.lock().await.keys() {
if CALL_GROUPS
.write()
.await
.lock()
.await
.get(cg)
.unwrap()
.lock()
.await
.is_empty()
.await
{
rem.push(cg.clone());
for call_id in groups_to_check {
// Retrieve the group Arc outside the global lock
if let Some(group_arc) = get_group(call_id).await {
// FIX 3: Check for emptiness outside the global lock
if group_arc.lock().await.is_empty().await {
rem.push(call_id);
}
}
}
for cg in rem {
CALL_GROUPS.write().await.lock().await.remove(&cg);
// Acquire write lock only to perform the removals.
if !rem.is_empty() {
let mut groups = CALL_GROUPS.write().await;
for call_id in rem {
groups.remove(&call_id);
}
}
}

11
src/data/communication.rs
View file

@ -11,6 +11,8 @@ pub enum DataTypes {
error_type,
accepted_ids,
uuid,
settings,
settings_name,
chat_partner_id,
iota_id,
user_id,
@ -85,6 +87,9 @@ impl DataTypes {
match normalized.as_str() {
"errortype" => DataTypes::error_type,
"chatpartnerid" => DataTypes::chat_partner_id,
"uuid" => DataTypes::uuid,
"settings" => DataTypes::settings,
"settingsname" => DataTypes::settings_name,
"iotaid" => DataTypes::iota_id,
"userid" => DataTypes::user_id,
"userids" => DataTypes::user_ids,
@ -164,6 +169,9 @@ pub enum CommunicationType {
error_invalid_secret,
error_invalid_private_key,
success,
settings_save,
settings_load,
settings_list,
message,
message_send,
message_live,
@ -214,6 +222,9 @@ impl CommunicationType {
match normalized.as_str() {
"error" => CommunicationType::error,
"settingssave" => CommunicationType::settings_save,
"settingsload" => CommunicationType::settings_load,
"settingslist" => CommunicationType::settings_list,
"success" => CommunicationType::success,
"message" => CommunicationType::message,
"messagelive" => CommunicationType::message_live,

2
src/rho/client_connection.rs
View file

@ -333,7 +333,7 @@ impl ClientConnection {
.with_sender(sender_id)
.add_data_str(DataTypes::call_id, call_id.to_string())
.add_data_str(DataTypes::receiver_id, receiver_id.to_string())
.add_data_str(DataTypes::sender_id, sender_id.to_string())
.add_data_str(DataTypes::sender_id, sender_id.to_string());
target_rho.message_to_client(distribute).await;

21
src/rho/iota_connection.rs
View file

@ -281,24 +281,19 @@ impl IotaConnection {
let interested_ids: Vec<Uuid> = Vec::new();
// Process contacts and add call information
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:
// Extract user IDs from contacts
// let contacts: Vec<ContactInfo> = parse_contacts(contacts_data);
// for contact in &contacts {
// interested_ids.push(contact.user_id);
// }
if let Some(contacts_data) = cv.get_data(DataTypes::user_ids) {
/*let contacts: Vec<(Uuid, String)> = parse_contacts(contacts_data);
for contact in &contacts {
interested_ids.push(contact.user_id);
}
// Get call invites for receiver
// let invites = CallManager::get_call_invites(receiver_id).await;
let invites = CallManager::get_call_invites(receiver_id).await;
// Enrich contacts with call information
// let enriched_contacts = enrich_with_calls(contacts, invites);
let enriched_contacts = enrich_with_calls(contacts, invites);
// cv = cv.add_data(DataTypes::user_ids, enriched_contacts.into());
cv = cv.add_data(DataTypes::user_ids, enriched_contacts.into());*/
}
// Notify OmegaConnection about user states