diff --git a/Cargo.lock b/Cargo.lock index d743298..4089ef5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7,6 +7,7 @@ name = "Iota" version = "0.1.0" dependencies = [ "aes-gcm", + "async-trait", "axum", "base64", "bytes", @@ -25,6 +26,7 @@ dependencies = [ "rand", "rand_core 0.6.4", "ratatui", + "reactive-rs", "reqwest", "rustls", "serde", @@ -105,6 +107,17 @@ dependencies = [ "libc", ] +[[package]] +name = "async-trait" +version = "0.1.89" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -1315,6 +1328,15 @@ dependencies = [ "hashbrown", ] +[[package]] +name = "mach" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "86dd2487cdfea56def77b88438a2c915fb45113c5319bfe7e14306ca4cd0b0e1" +dependencies = [ + "libc", +] + [[package]] name = "matchit" version = "0.8.4" @@ -1676,6 +1698,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "reactive-rs" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dee41fe079cafa6f6d12316ce44a1530aebcfb21b58e0face916dcbf19d47eab" +dependencies = [ + "slice-deque", +] + [[package]] name = "redox_syscall" version = "0.5.17" @@ -1991,6 +2022,17 @@ version = "0.4.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7a2ae44ef20feb57a68b23d846850f861394c2e02dc425a50098ae8c90267589" +[[package]] +name = "slice-deque" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d39fca478d10e201944a8e21f4393d6bfe38fa3b16a152050e4d097fe2bbf494" +dependencies = [ + "libc", + "mach", + "winapi", +] + [[package]] name = "smallvec" version = "1.15.1" diff --git a/Cargo.toml b/Cargo.toml index af7c669..a6a8a08 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ version = "0.1.0" edition = "2024" [dependencies] +async-trait = "*" json = "*" axum = "*" futures-util = "*" @@ -40,3 +41,4 @@ futures = "*" aes-gcm = "*" hkdf = "*" hmac = "*" +reactive-rs = "*" diff --git a/src/communities/community.rs b/src/communities/community.rs index 26b2337..8374bb4 100644 --- a/src/communities/community.rs +++ b/src/communities/community.rs @@ -89,6 +89,20 @@ impl Community { .await .insert(other.get_user_id().await.unwrap(), vec); } + pub async fn remove_connection(self: &Arc, other: Arc) { + let mut vec = self + .connections + .read() + .await + .get(&other.get_user_id().await.unwrap()) + .cloned() + .unwrap_or_default(); + vec.retain(|conn| !Arc::ptr_eq(conn, &other)); + self.connections + .write() + .await + .insert(other.get_user_id().await.unwrap(), vec); + } pub async fn get_connections(&self) -> HashMap>> { self.connections.read().await.clone() } @@ -127,13 +141,14 @@ impl Community { cv: &CommunicationValue, ) -> CommunicationValue { if path.is_empty() { - let mut target_interactables = &self.interactables.read().await.clone(); + let target_interactables = &self.interactables.read().await.clone(); for interactable in target_interactables.iter() { if interactable.get_name() == name { if interactable.get_codec() == "category" { return CommunicationValue::new(CommunicationType::error); } else { - return interactable.run_function(cv.clone()); + // cannot move a value of type dyn Interactable the size of dyn Interactable cannot be statically determined (rustc E0161) + return interactable.run_function(cv.clone()).await; } } } @@ -144,10 +159,12 @@ impl Community { if interactable.get_codec() == "category" { let category: &Category = interactable.as_any().downcast_ref::().unwrap(); + // cannot move a value of type dyn Interactable the size of dyn Interactable cannot be statically determined (rustc E0161) return category .get_child(path.to_string(), name.to_string()) .unwrap() - .run_function(cv.clone()); + .run_function(cv.clone()) + .await; } else { return CommunicationValue::new(CommunicationType::error); } diff --git a/src/communities/community_connection.rs b/src/communities/community_connection.rs index 970ab85..a4f7b6c 100644 --- a/src/communities/community_connection.rs +++ b/src/communities/community_connection.rs @@ -8,7 +8,6 @@ use base64::{Engine as _, engine::general_purpose::STANDARD}; use futures::SinkExt; use hkdf::Hkdf; use json::JsonValue; -use json::object::Object; use rand::{Rng, distributions::Alphanumeric}; use sha2::Sha256; use std::sync::Arc; @@ -392,10 +391,16 @@ impl CommunityConnection { let mut session = self.session.lock().await; let _ = session.close(None).await; } - pub async fn handle_close(&self) { + pub async fn handle_close(self: Arc) { if self.is_identified().await { - if let Some(user_id) = self.get_user_id().await { - todo!(); + if let Some(_) = self.get_user_id().await { + self.community + .read() + .await + .as_ref() + .unwrap() + .remove_connection(self.clone()) + .await; } } } diff --git a/src/communities/community_manager.rs b/src/communities/community_manager.rs index ae6acab..bb628ed 100644 --- a/src/communities/community_manager.rs +++ b/src/communities/community_manager.rs @@ -5,7 +5,6 @@ use once_cell::sync::Lazy; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::Mutex; -use uuid::Uuid; pub static COMMUNITY_REGISTRY: Lazy>>>> = Lazy::new(|| Arc::new(Mutex::new(HashMap::new()))); diff --git a/src/communities/community_socket.rs b/src/communities/community_socket.rs index c095bd2..77d44c2 100644 --- a/src/communities/community_socket.rs +++ b/src/communities/community_socket.rs @@ -1,6 +1,4 @@ -use crate::communities::{ - community::Community, community_connection::CommunityConnection, community_manager, -}; +use crate::communities::{community_connection::CommunityConnection, community_manager}; use futures::StreamExt; use std::sync::Arc; use tokio::net::TcpListener; diff --git a/src/communities/interactables/category.rs b/src/communities/interactables/category.rs index e352022..e1c151b 100644 --- a/src/communities/interactables/category.rs +++ b/src/communities/interactables/category.rs @@ -2,6 +2,7 @@ use crate::{ communities::{community::Community, interactables::interactable::Interactable}, data::communication::{CommunicationType, CommunicationValue}, }; +use async_trait::async_trait; use axum::Json; use json::JsonValue; use std::any::Any; @@ -48,6 +49,7 @@ impl Category { } } +#[async_trait] impl Interactable for Category { fn as_any(&self) -> &dyn Any { self @@ -89,7 +91,7 @@ impl Interactable for Category { } v } - fn run_function(&self, cv: CommunicationValue) -> CommunicationValue { + async fn run_function(&self, cv: CommunicationValue) -> CommunicationValue { CommunicationValue::new(CommunicationType::error) } fn to_json(&self) -> JsonValue { diff --git a/src/communities/interactables/interactable.rs b/src/communities/interactables/interactable.rs index 7982a69..72f5252 100644 --- a/src/communities/interactables/interactable.rs +++ b/src/communities/interactables/interactable.rs @@ -1,14 +1,12 @@ -use crate::{ - communities::community::Community, - data::communication::{CommunicationType, CommunicationValue}, -}; -use axum::Json; +use crate::{communities::community::Community, data::communication::CommunicationValue}; +use async_trait::async_trait; use json::JsonValue; +use std::any::Any; use std::sync::Arc; -use std::{any::Any, pin::Pin}; pub type InteractableFactory = fn() -> Box; +#[async_trait] pub trait Interactable: Send + Sync + Any { fn as_any(&self) -> &dyn Any; fn as_any_mut(&mut self) -> &mut dyn Any; @@ -20,7 +18,7 @@ pub trait Interactable: Send + Sync + Any { fn set_path(&mut self, path: String); fn get_community(&self) -> &Arc; fn set_community(&mut self, community: Arc); - fn run_function(&self, cv: CommunicationValue) -> CommunicationValue; + async fn run_function(&self, cv: CommunicationValue) -> CommunicationValue; fn get_data(&self) -> JsonValue; fn to_json(&self) -> JsonValue; fn load(&mut self, community: Arc, path: String, name: String, json: &JsonValue); diff --git a/src/communities/interactables/registry.rs b/src/communities/interactables/registry.rs index d9f6954..67ecf3a 100644 --- a/src/communities/interactables/registry.rs +++ b/src/communities/interactables/registry.rs @@ -3,7 +3,6 @@ use crate::communities::interactables::category::Category; use crate::communities::interactables::interactable::{Interactable, InteractableFactory}; use crate::communities::interactables::text_chat::TextChat; use crate::communities::interactables::voice_chat::VoiceChat; -use crate::gui::log_panel; use crate::util::file_util; use json::JsonValue; use once_cell::sync::Lazy; diff --git a/src/communities/interactables/text_chat.rs b/src/communities/interactables/text_chat.rs index 5bc0ead..aa87418 100644 --- a/src/communities/interactables/text_chat.rs +++ b/src/communities/interactables/text_chat.rs @@ -8,6 +8,7 @@ use crate::{ util::file_util::{get_children, load_file, save_file}, }; use aes_gcm::aead::Payload; +use async_trait::async_trait; use axum::Json; use json::{JsonValue, array, object}; use rustls::ClientConnection; @@ -155,6 +156,7 @@ impl TextChat { messages } } +#[async_trait] impl Interactable for TextChat { fn as_any(&self) -> &dyn Any { self @@ -189,62 +191,59 @@ impl Interactable for TextChat { fn get_data(&self) -> JsonValue { JsonValue::new_object() } - fn run_function(&self, cv: CommunicationValue) -> CommunicationValue { - let ret: Pin + Send>> = Box::pin(async move { - let payload = cv.get_data(DataTypes::payload).unwrap(); - if cv.get_data(DataTypes::function).unwrap().as_str().unwrap() == "get_messages" { - let amount = payload["amount"].as_i64().unwrap(); - let loaded_messages = payload["loaded_messages"].as_i64().unwrap(); - let messages = self.get_messages(loaded_messages, amount).clone(); - let mut payload = JsonValue::new_object(); - payload["messages"] = messages; - return CommunicationValue::new(CommunicationType::function) - .with_id(cv.get_id()) - .add_data_str(DataTypes::name, self.name.clone()) - .add_data_str(DataTypes::path, self.path.clone()) - .add_data_str(DataTypes::result, "message_chunk".to_string()) - .add_data(DataTypes::payload, payload); - } - if cv.get_data(DataTypes::function).unwrap().as_str().unwrap() == "send_message" { - let message = payload["message"].as_str().unwrap(); - let milliseconds_timestamp: u128 = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis(); - self.add_message(milliseconds_timestamp, cv.get_sender().unwrap(), message); + async fn run_function(&self, cv: CommunicationValue) -> CommunicationValue { + let payload = cv.get_data(DataTypes::payload).unwrap(); + if cv.get_data(DataTypes::function).unwrap().as_str().unwrap() == "get_messages" { + let amount = payload["amount"].as_i64().unwrap(); + let loaded_messages = payload["loaded_messages"].as_i64().unwrap(); + let messages = self.get_messages(loaded_messages, amount).clone(); + let mut payload = JsonValue::new_object(); + payload["messages"] = messages; + return CommunicationValue::new(CommunicationType::function) + .with_id(cv.get_id()) + .add_data_str(DataTypes::name, self.name.clone()) + .add_data_str(DataTypes::path, self.path.clone()) + .add_data_str(DataTypes::result, "message_chunk".to_string()) + .add_data(DataTypes::payload, payload); + } + if cv.get_data(DataTypes::function).unwrap().as_str().unwrap() == "send_message" { + let message = payload["message"].as_str().unwrap(); + let milliseconds_timestamp: u128 = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis(); + self.add_message(milliseconds_timestamp, cv.get_sender().unwrap(), message); - let mut distribution_payload = JsonValue::new_object(); - distribution_payload["message"] = JsonValue::String(message.to_string()); - distribution_payload["sender_id"] = - JsonValue::String(cv.get_sender().unwrap().to_string()); - distribution_payload["send_time"] = - JsonValue::String(milliseconds_timestamp.to_string()); - let distribution = CommunicationValue::new(CommunicationType::update) - .with_id(cv.get_id()) - .add_data_str(DataTypes::name, self.name.clone()) - .add_data_str(DataTypes::path, self.path.clone()) - .add_data_str(DataTypes::result, "message_live".to_string()) - .add_data(DataTypes::payload, distribution_payload); + let mut distribution_payload = JsonValue::new_object(); + distribution_payload["message"] = JsonValue::String(message.to_string()); + distribution_payload["sender_id"] = + JsonValue::String(cv.get_sender().unwrap().to_string()); + distribution_payload["send_time"] = + JsonValue::String(milliseconds_timestamp.to_string()); + let distribution = CommunicationValue::new(CommunicationType::update) + .with_id(cv.get_id()) + .add_data_str(DataTypes::name, self.name.clone()) + .add_data_str(DataTypes::path, self.path.clone()) + .add_data_str(DataTypes::result, "message_live".to_string()) + .add_data(DataTypes::payload, distribution_payload); - let connections: HashMap>> = - self.get_community().get_connections().await.clone(); + let connections: HashMap>> = + self.get_community().get_connections().await.clone(); - for con in connections.values() { - for c in con { - let cd: &Arc = c; - cd.send_message(&distribution).await; - } + for con in connections.values() { + for c in con { + let cd: &Arc = c; + cd.send_message(&distribution).await; } - return CommunicationValue::new(CommunicationType::function) - .with_id(cv.get_id()) - .add_data_str(DataTypes::name, self.name.clone()) - .add_data_str(DataTypes::path, self.path.clone()) - .add_data_str(DataTypes::result, "message_received".to_string()) - .add_data(DataTypes::payload, JsonValue::new_object()); } - CommunicationValue::new(CommunicationType::error).with_id(cv.get_id()) - }); - CommunicationValue::new(CommunicationType::error) + return CommunicationValue::new(CommunicationType::function) + .with_id(cv.get_id()) + .add_data_str(DataTypes::name, self.name.clone()) + .add_data_str(DataTypes::path, self.path.clone()) + .add_data_str(DataTypes::result, "message_received".to_string()) + .add_data(DataTypes::payload, JsonValue::new_object()); + } + CommunicationValue::new(CommunicationType::error).with_id(cv.get_id()) } fn to_json(&self) -> JsonValue { let mut v = JsonValue::new_object(); diff --git a/src/communities/interactables/voice_chat.rs b/src/communities/interactables/voice_chat.rs index 5841e24..2becbc4 100644 --- a/src/communities/interactables/voice_chat.rs +++ b/src/communities/interactables/voice_chat.rs @@ -1,28 +1,46 @@ use crate::{ communities::{community::Community, interactables::interactable::Interactable}, - data::communication::{CommunicationType, CommunicationValue}, + data::communication::{CommunicationType, CommunicationValue, DataTypes}, }; +use async_trait::async_trait; use json::JsonValue; -use std::any::Any; use std::sync::Arc; +use std::{any::Any, sync::RwLock}; use uuid::Uuid; pub enum CallUserState { Active, Muted, Deafed, } +impl CallUserState { + pub fn parse(state: &str) -> CallUserState { + match state { + "active" => CallUserState::Active, + "muted" => CallUserState::Muted, + "deafed" => CallUserState::Deafed, + _ => CallUserState::Active, + } + } + pub fn to_string(&self) -> String { + match self { + CallUserState::Active => "active".to_string(), + CallUserState::Muted => "muted".to_string(), + CallUserState::Deafed => "deafed".to_string(), + } + } +} pub struct CallUser { - user_id: Uuid, - user_state: CallUserState, - streaming: bool, + pub user_id: Uuid, + pub user_state: CallUserState, + pub streaming: bool, } pub struct VoiceChat { name: String, path: String, community: Arc, - users: Vec, + users: RwLock>, } impl VoiceChat { pub fn new() -> VoiceChat { @@ -30,10 +48,28 @@ impl VoiceChat { name: String::new(), path: String::new(), community: Arc::new(Community::new()), - users: Vec::new(), + users: RwLock::new(Vec::new()), + } + } + pub fn update_user_state( + self: Arc, + user_id: Uuid, + state: CallUserState, + streaming: bool, + ) { + if let Some(user) = self + .users + .write() + .unwrap() + .iter_mut() + .find(|u| u.user_id == user_id) + { + user.user_state = state; + user.streaming = streaming; } } } +#[async_trait] impl Interactable for VoiceChat { fn as_any(&self) -> &dyn Any { self @@ -66,16 +102,93 @@ impl Interactable for VoiceChat { String::new() + &self.path + "/" + &self.name } fn get_data(&self) -> JsonValue { - JsonValue::new_object() + /* + * "data": { + "active_users": { + "user_id": { + "state": "", + "streaming": boolean + }, + "user_id": { + "state": "", + "streaming": boolean + }, + "user_id": { + "state": "", + "streaming": boolean + } + } + } + }, + */ + let mut data = JsonValue::new_object(); + let mut active_users = JsonValue::new_object(); + for user in self.users.read().unwrap().iter() { + let mut user_data = JsonValue::new_object(); + let _ = user_data.insert("state", JsonValue::String(user.user_state.to_string())); + let _ = user_data.insert("streaming", JsonValue::Boolean(user.streaming)); + let _ = active_users.insert(&user.user_id.to_string(), user_data); + } + let _ = data.insert("active_users", active_users); + data } - fn run_function(&self, cv: CommunicationValue) -> CommunicationValue { - CommunicationValue::new(CommunicationType::error) + async fn run_function(&self, cv: CommunicationValue) -> CommunicationValue { + let payload = cv.get_data(DataTypes::payload).unwrap(); + let function = cv.get_data(DataTypes::function).unwrap().as_str().unwrap(); + + if function == "get_call" { + let sender_id = payload["sender_id"].as_str().unwrap(); + let message_id = payload["message"].as_str().unwrap(); + let send_time = payload["send_time"].as_str().unwrap(); + + let mut response_payload = JsonValue::new_object(); + response_payload["sender_id"] = JsonValue::String(sender_id.to_string()); + response_payload["message"] = JsonValue::String(message_id.to_string()); + response_payload["send_time"] = JsonValue::String(send_time.to_string()); + + return CommunicationValue::new(CommunicationType::function) + .with_id(cv.get_id()) + .add_data_str(DataTypes::name, self.name.clone()) + .add_data_str(DataTypes::path, self.path.clone()) + .add_data_str(DataTypes::result, "getting_call".to_string()) + .add_data(DataTypes::payload, response_payload); + } + + if function == "update_user_state" { + let user_id = payload["user_id"].as_str().unwrap(); + let state = payload["state"].as_str().unwrap(); + let streaming = payload["streaming"].as_bool().unwrap(); + + if let Some(user) = self + .users + .write() + .unwrap() + .iter_mut() + .find(|u| u.user_id == Uuid::parse_str(user_id).unwrap()) + { + user.user_state = CallUserState::parse(state); + user.streaming = streaming; + } + let mut response_payload = JsonValue::new_object(); + response_payload["user_id"] = JsonValue::String(user_id.to_string()); + response_payload["state"] = JsonValue::String(state.to_string()); + response_payload["streaming"] = JsonValue::Boolean(streaming); + + return CommunicationValue::new(CommunicationType::update) + .with_id(cv.get_id()) + .add_data_str(DataTypes::name, self.name.clone()) + .add_data_str(DataTypes::path, self.path.clone()) + .add_data_str(DataTypes::result, "user_changed".to_string()) + .add_data(DataTypes::payload, response_payload); + } + CommunicationValue::new(CommunicationType::error).with_id(cv.get_id()) } + fn to_json(&self) -> JsonValue { - let mut v = JsonValue::new_object(); + let v = JsonValue::new_object(); v } - fn load(&mut self, community: Arc, path: String, name: String, json: &JsonValue) { + fn load(&mut self, community: Arc, path: String, name: String, _: &JsonValue) { self.community = community; self.name = name; self.path = path; diff --git a/src/main.rs b/src/main.rs index 5aff9df..4eaed70 100644 --- a/src/main.rs +++ b/src/main.rs @@ -52,7 +52,7 @@ async fn main() { } // USER MANAGEMENT - user_manager::load_users().await; + let _ = user_manager::load_users().await; let mut sb = "".to_string(); for up in user_manager::get_users() { sb = sb + "," + &up.user_id.to_string().as_str();