Community interactables

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"

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 = "*"

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<Self>, other: Arc<CommunityConnection>) {
+        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<Uuid, Vec<Arc<CommunityConnection>>> {
         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::<Category>().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);
                     }

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<Self>) {
         if self.is_identified().await {
-            if let Some(user_id) = self.get_user_id().await {
+            if let Some(_) = self.get_user_id().await {
-                todo!();
+                self.community
+                    .read()
+                    .await
+                    .as_ref()
+                    .unwrap()
+                    .remove_connection(self.clone())
+                    .await;
             }
         }
     }

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<Arc<Mutex<HashMap<String, Arc<Community>>>>> =
     Lazy::new(|| Arc::new(Mutex::new(HashMap::new())));

src/communities/community_socket.rs

@@ -1,6 +1,4 @@
-use crate::communities::{
+use crate::communities::{community_connection::CommunityConnection, community_manager};
-    community::Community, community_connection::CommunityConnection, community_manager,
-};
 use futures::StreamExt;
 use std::sync::Arc;
 use tokio::net::TcpListener;

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 {

src/communities/interactables/interactable.rs

@@ -1,14 +1,12 @@
-use crate::{
+use crate::{communities::community::Community, data::communication::CommunicationValue};
-    communities::community::Community,
+use async_trait::async_trait;
-    data::communication::{CommunicationType, CommunicationValue},
-};
-use axum::Json;
 use json::JsonValue;
+use std::any::Any;
 use std::sync::Arc;
-use std::{any::Any, pin::Pin};
 
 pub type InteractableFactory = fn() -> Box<dyn Interactable>;
 
+#[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<Community>;
     fn set_community(&mut self, community: Arc<Community>);
-    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<Community>, path: String, name: String, json: &JsonValue);

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;

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,8 +191,7 @@ impl Interactable for TextChat {
     fn get_data(&self) -> JsonValue {
         JsonValue::new_object()
     }
-    fn run_function(&self, cv: CommunicationValue) -> CommunicationValue {
+    async fn run_function(&self, cv: CommunicationValue) -> CommunicationValue {
-        let ret: Pin<Box<dyn Future<Output = CommunicationValue> + 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();
@@ -243,8 +244,6 @@ impl Interactable for TextChat {
                 .add_data(DataTypes::payload, JsonValue::new_object());
         }
         CommunicationValue::new(CommunicationType::error).with_id(cv.get_id())
-        });
-        CommunicationValue::new(CommunicationType::error)
     }
     fn to_json(&self) -> JsonValue {
         let mut v = JsonValue::new_object();

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,
+    pub user_id: Uuid,
-    user_state: CallUserState,
+    pub user_state: CallUserState,
-    streaming: bool,
+    pub streaming: bool,
 }
 
 pub struct VoiceChat {
     name: String,
     path: String,
     community: Arc<Community>,
-    users: Vec<CallUser>,
+    users: RwLock<Vec<CallUser>>,
 }
 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<Self>,
+        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": "<call_status>",
+                          "streaming": boolean
+                        },
+                        "user_id": {
+                          "state": "<call_status>",
+                          "streaming": boolean
+                         },
+                         "user_id": {
+                          "state": "<call_status>",
+                          "streaming": boolean
                          }
-    fn run_function(&self, cv: CommunicationValue) -> CommunicationValue {
-        CommunicationValue::new(CommunicationType::error)
                       }
+                    }
+                  },
+        */
+        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
+    }
+    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<Community>, path: String, name: String, json: &JsonValue) {
+    fn load(&mut self, community: Arc<Community>, path: String, name: String, _: &JsonValue) {
         self.community = community;
         self.name = name;
         self.path = path;

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();