This commit is contained in:
Alex Emmet 2025-11-14 11:11:30 +01:00
commit 7baa5a52fe
4 changed files with 27 additions and 25 deletions

View file

@ -133,7 +133,7 @@ impl CallConnection {
let mut user_info = JsonValue::new_object(); let mut user_info = JsonValue::new_object();
let _ = user_info.insert("state", JsonValue::from("muted")); let _ = user_info.insert("state", JsonValue::from("muted"));
let _ = user_info.insert("streaming", JsonValue::from(false)); let _ = user_info.insert("streaming", JsonValue::from(false));
let _ = users.insert(&caller.user_id.to_string(), user_info); let _ = users.insert(&caller.user_id.read().await.to_string(), user_info);
} }
} }
response = response.add_data(DataTypes::about, users); response = response.add_data(DataTypes::about, users);
@ -249,8 +249,7 @@ impl CallConnection {
.to_json() .to_json()
.to_string(), .to_string(),
); );
group.lock().await.get_member(uid).await; group.lock().await.disconnect_member(uid).await;
group.lock().await.remove_member(uid);
} }
call_manager::remove_inactive().await; call_manager::remove_inactive().await;
} }

View file

@ -21,19 +21,21 @@ impl CallGroup {
self.callers self.callers
.insert(user_id, Arc::new(Caller::new(user_id, tx))); .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) { pub fn remove_member(&mut self, user_id: Uuid) {
self.callers.remove(&user_id); self.callers.remove(&user_id);
} }
pub fn get_member(&self, user_id: &Uuid) -> Option<&Arc<Caller>> {
self.callers.get(user_id)
}
pub async fn disconnect_member(&mut self, user_id: Uuid) {
if let Some(caller) = self.callers.get(&user_id) {
caller.disconnect().await;
}
}
pub fn is_empty(&self) -> bool { pub async fn is_empty(&self) -> bool {
for caller in self.callers.values() { for caller in self.callers.values() {
if !caller.is_connected() { if !caller.is_connected().await {
return false; return false;
} }
} }

View file

@ -46,6 +46,7 @@ pub async fn remove_inactive() {
.lock() .lock()
.await .await
.is_empty() .is_empty()
.await
{ {
rem.push(cg.clone()); rem.push(cg.clone());
} }

View file

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