use crate::{AudioStream, Participant, RemoteTrack, RoomEvent, TrackPublication}; use crate::mock_client::{participant::*, publication::*, track::*};use anyhow::{Context as _, Result, anyhow};use async_trait::async_trait;use collections::{BTreeMap, HashMap, HashSet, btree_map::Entry as BTreeEntry, hash_map::Entry};use gpui::{App, AsyncApp, BackgroundExecutor};use livekit_api::{proto, token};use parking_lot::Mutex;use postage::{mpsc, sink::Sink};use std::sync::{ Arc, Weak, atomic::{AtomicBool, Ordering::SeqCst},}; #[derive(Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]pub struct ParticipantIdentity(pub String); #[derive(Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]pub struct TrackSid(pub(crate) String); impl std::fmt::Display for TrackSid { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { self.0.fmt(f) }} impl TryFrom<String> for TrackSid { type Error = anyhow::Error; fn try_from(value: String) -> Result<Self, Self::Error> { Ok(TrackSid(value)) }} #[derive(Copy, Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]#[non_exhaustive]pub enum ConnectionState { Connected, Disconnected,} static SERVERS: Mutex<BTreeMap<String, Arc<TestServer>>> = Mutex::new(BTreeMap::new()); pub struct TestServer { pub url: String, pub api_key: String, pub secret_key: String, rooms: Mutex<HashMap<String, TestServerRoom>>, executor: BackgroundExecutor,} impl TestServer { pub fn create( url: String, api_key: String, secret_key: String, executor: BackgroundExecutor, ) -> Result<Arc<TestServer>> { let mut servers = SERVERS.lock(); if let BTreeEntry::Vacant(e) = servers.entry(url.clone()) { let server = Arc::new(TestServer { url, api_key, secret_key, rooms: Default::default(), executor, }); e.insert(server.clone()); Ok(server) } else { Err(anyhow!("a server with url {:?} already exists", url)) } } fn get(url: &str) -> Result<Arc<TestServer>> { Ok(SERVERS .lock() .get(url) .ok_or_else(|| anyhow!("no server found for url"))? .clone()) } pub fn teardown(&self) -> Result<()> { SERVERS .lock() .remove(&self.url) .ok_or_else(|| anyhow!("server with url {:?} does not exist", self.url))?; Ok(()) } pub fn create_api_client(&self) -> TestApiClient { TestApiClient { url: self.url.clone(), } } pub async fn create_room(&self, room: String) -> Result<()> { self.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); if let Entry::Vacant(e) = server_rooms.entry(room.clone()) { e.insert(Default::default()); Ok(()) } else { Err(anyhow!("room {:?} already exists", room)) } } async fn delete_room(&self, room: String) -> Result<()> { self.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); server_rooms .remove(&room) .ok_or_else(|| anyhow!("room {:?} does not exist", room))?; Ok(()) } async fn join_room(&self, token: String, client_room: Room) -> Result<ParticipantIdentity> { self.simulate_random_delay().await; let claims = livekit_api::token::validate(&token, &self.secret_key)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); let room = (*server_rooms).entry(room_name.to_string()).or_default(); if let Entry::Vacant(e) = room.client_rooms.entry(identity.clone()) { for server_track in &room.video_tracks { let track = RemoteTrack::Video(RemoteVideoTrack { server_track: server_track.clone(), _room: client_room.downgrade(), }); client_room .0 .lock() .updates_tx .blocking_send(RoomEvent::TrackSubscribed { track: track.clone(), publication: RemoteTrackPublication { sid: server_track.sid.clone(), room: client_room.downgrade(), track, }, participant: RemoteParticipant { room: client_room.downgrade(), identity: server_track.publisher_id.clone(), }, }) .unwrap(); } for server_track in &room.audio_tracks { let track = RemoteTrack::Audio(RemoteAudioTrack { server_track: server_track.clone(), room: client_room.downgrade(), }); client_room .0 .lock() .updates_tx .blocking_send(RoomEvent::TrackSubscribed { track: track.clone(), publication: RemoteTrackPublication { sid: server_track.sid.clone(), room: client_room.downgrade(), track, }, participant: RemoteParticipant { room: client_room.downgrade(), identity: server_track.publisher_id.clone(), }, }) .unwrap(); } e.insert(client_room); Ok(identity) } else { Err(anyhow!( "{:?} attempted to join room {:?} twice", identity, room_name )) } } async fn leave_room(&self, token: String) -> Result<()> { self.simulate_random_delay().await; let claims = livekit_api::token::validate(&token, &self.secret_key)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&*room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; room.client_rooms.remove(&identity).ok_or_else(|| { anyhow!( "{:?} attempted to leave room {:?} before joining it", identity, room_name ) })?; Ok(()) } fn remote_participants( &self, token: String, ) -> Result<HashMap<ParticipantIdentity, RemoteParticipant>> { let claims = livekit_api::token::validate(&token, &self.secret_key)?; let local_identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap().to_string(); if let Some(server_room) = self.rooms.lock().get(&room_name) { let room = server_room .client_rooms .get(&local_identity) .unwrap() .downgrade(); Ok(server_room .client_rooms .iter() .filter(|(identity, _)| *identity != &local_identity) .map(|(identity, _)| { ( identity.clone(), RemoteParticipant { room: room.clone(), identity: identity.clone(), }, ) }) .collect()) } else { Ok(Default::default()) } } async fn remove_participant( &self, room_name: String, identity: ParticipantIdentity, ) -> Result<()> { self.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; room.client_rooms.remove(&identity).ok_or_else(|| { anyhow!( "participant {:?} did not join room {:?}", identity, room_name ) })?; Ok(()) } async fn update_participant( &self, room_name: String, identity: String, permission: proto::ParticipantPermission, ) -> Result<()> { self.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; room.participant_permissions .insert(ParticipantIdentity(identity), permission); Ok(()) } pub async fn disconnect_client(&self, client_identity: String) { let client_identity = ParticipantIdentity(client_identity); self.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); for room in server_rooms.values_mut() { if let Some(room) = room.client_rooms.remove(&client_identity) { let mut room = room.0.lock(); room.connection_state = ConnectionState::Disconnected; room.updates_tx .blocking_send(RoomEvent::Disconnected { reason: "SIGNAL_CLOSED", }) .ok(); } } } pub(crate) async fn publish_video_track( &self, token: String, _local_track: LocalVideoTrack, ) -> Result<TrackSid> { self.simulate_random_delay().await; let claims = livekit_api::token::validate(&token, &self.secret_key)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&*room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; let can_publish = room .participant_permissions .get(&identity) .map(|permission| permission.can_publish) .or(claims.video.can_publish) .unwrap_or(true); if !can_publish { return Err(anyhow!("user is not allowed to publish")); } let sid: TrackSid = format!("TR_{}", nanoid::nanoid!(17)).try_into().unwrap(); let server_track = Arc::new(TestServerVideoTrack { sid: sid.clone(), publisher_id: identity.clone(), }); room.video_tracks.push(server_track.clone()); for (room_identity, client_room) in &room.client_rooms { if *room_identity != identity { let track = RemoteTrack::Video(RemoteVideoTrack { server_track: server_track.clone(), _room: client_room.downgrade(), }); let publication = RemoteTrackPublication { sid: sid.clone(), room: client_room.downgrade(), track: track.clone(), }; let participant = RemoteParticipant { identity: identity.clone(), room: client_room.downgrade(), }; client_room .0 .lock() .updates_tx .blocking_send(RoomEvent::TrackSubscribed { track, publication, participant, }) .unwrap(); } } Ok(sid) } pub(crate) async fn publish_audio_track( &self, token: String, _local_track: &LocalAudioTrack, ) -> Result<TrackSid> { self.simulate_random_delay().await; let claims = livekit_api::token::validate(&token, &self.secret_key)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&*room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; let can_publish = room .participant_permissions .get(&identity) .map(|permission| permission.can_publish) .or(claims.video.can_publish) .unwrap_or(true); if !can_publish { return Err(anyhow!("user is not allowed to publish")); } let sid: TrackSid = format!("TR_{}", nanoid::nanoid!(17)).try_into().unwrap(); let server_track = Arc::new(TestServerAudioTrack { sid: sid.clone(), publisher_id: identity.clone(), muted: AtomicBool::new(false), }); room.audio_tracks.push(server_track.clone()); for (room_identity, client_room) in &room.client_rooms { if *room_identity != identity { let track = RemoteTrack::Audio(RemoteAudioTrack { server_track: server_track.clone(), room: client_room.downgrade(), }); let publication = RemoteTrackPublication { sid: sid.clone(), room: client_room.downgrade(), track: track.clone(), }; let participant = RemoteParticipant { identity: identity.clone(), room: client_room.downgrade(), }; client_room .0 .lock() .updates_tx .blocking_send(RoomEvent::TrackSubscribed { track, publication, participant, }) .ok(); } } Ok(sid) } pub(crate) async fn unpublish_track(&self, _token: String, _track: &TrackSid) -> Result<()> { Ok(()) } pub(crate) fn set_track_muted( &self, token: &str, track_sid: &TrackSid, muted: bool, ) -> Result<()> { let claims = livekit_api::token::validate(&token, &self.secret_key)?; let room_name = claims.video.room.unwrap(); let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&*room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; if let Some(track) = room .audio_tracks .iter_mut() .find(|track| track.sid == *track_sid) { track.muted.store(muted, SeqCst); for (id, client_room) in room.client_rooms.iter() { if *id != identity { let participant = Participant::Remote(RemoteParticipant { identity: identity.clone(), room: client_room.downgrade(), }); let track = RemoteTrack::Audio(RemoteAudioTrack { server_track: track.clone(), room: client_room.downgrade(), }); let publication = TrackPublication::Remote(RemoteTrackPublication { sid: track_sid.clone(), room: client_room.downgrade(), track, }); let event = if muted { RoomEvent::TrackMuted { participant, publication, } } else { RoomEvent::TrackUnmuted { participant, publication, } }; client_room .0 .lock() .updates_tx .blocking_send(event) .unwrap(); } } } Ok(()) } pub(crate) fn is_track_muted(&self, token: &str, track_sid: &TrackSid) -> Option<bool> { let claims = livekit_api::token::validate(&token, &self.secret_key).ok()?; let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); let room = server_rooms.get_mut(&*room_name)?; room.audio_tracks.iter().find_map(|track| { if track.sid == *track_sid {Lines 1–500 of 771. Load more
use crate::{AudioStream, Participant, RemoteTrack, RoomEvent, TrackPublication}; use crate::mock_client::{participant::*, publication::*, track::*}; use anyhow::{Context as _, Result, anyhow}; use async_trait::async_trait; use collections::{BTreeMap, HashMap, HashSet, btree_map::Entry as BTreeEntry, hash_map::Entry}; use gpui::{App, AsyncApp, BackgroundExecutor}; use livekit_api::{proto, token}; use parking_lot::Mutex; use postage::{mpsc, sink::Sink}; use std::sync::{ Arc, Weak, atomic::{AtomicBool, Ordering::SeqCst}, }; #[derive(Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] pub struct ParticipantIdentity(pub String); #[derive(Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] pub struct TrackSid(pub(crate) String); impl std::fmt::Display for TrackSid { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { self.0.fmt(f) } } impl TryFrom<String> for TrackSid { type Error = anyhow::Error; fn try_from(value: String) -> Result<Self, Self::Error> { Ok(TrackSid(value)) } } #[derive(Copy, Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] #[non_exhaustive] pub enum ConnectionState { Connected, Disconnected, } static SERVERS: Mutex<BTreeMap<String, Arc<TestServer>>> = Mutex::new(BTreeMap::new()); pub struct TestServer { pub url: String, pub api_key: String, pub secret_key: String, rooms: Mutex<HashMap<String, TestServerRoom>>, executor: BackgroundExecutor, } impl TestServer { pub fn create( url: String, api_key: String, secret_key: String, executor: BackgroundExecutor, ) -> Result<Arc<TestServer>> { let mut servers = SERVERS.lock(); if let BTreeEntry::Vacant(e) = servers.entry(url.clone()) { let server = Arc::new(TestServer { url, api_key, secret_key, rooms: Default::default(), executor, }); e.insert(server.clone()); Ok(server) } else { Err(anyhow!("a server with url {:?} already exists", url)) } } fn get(url: &str) -> Result<Arc<TestServer>> { Ok(SERVERS .lock() .get(url) .ok_or_else(|| anyhow!("no server found for url"))? .clone()) } pub fn teardown(&self) -> Result<()> { SERVERS .lock() .remove(&self.url) .ok_or_else(|| anyhow!("server with url {:?} does not exist", self.url))?; Ok(()) } pub fn create_api_client(&self) -> TestApiClient { TestApiClient { url: self.url.clone(), } } pub async fn create_room(&self, room: String) -> Result<()> { self.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); if let Entry::Vacant(e) = server_rooms.entry(room.clone()) { e.insert(Default::default()); Ok(()) } else { Err(anyhow!("room {:?} already exists", room)) } } async fn delete_room(&self, room: String) -> Result<()> { self.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); server_rooms .remove(&room) .ok_or_else(|| anyhow!("room {:?} does not exist", room))?; Ok(()) } async fn join_room(&self, token: String, client_room: Room) -> Result<ParticipantIdentity> { self.simulate_random_delay().await; let claims = livekit_api::token::validate(&token, &self.secret_key)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); let room = (*server_rooms).entry(room_name.to_string()).or_default(); if let Entry::Vacant(e) = room.client_rooms.entry(identity.clone()) { for server_track in &room.video_tracks { let track = RemoteTrack::Video(RemoteVideoTrack { server_track: server_track.clone(), _room: client_room.downgrade(), }); client_room .0 .lock() .updates_tx .blocking_send(RoomEvent::TrackSubscribed { track: track.clone(), publication: RemoteTrackPublication { sid: server_track.sid.clone(), room: client_room.downgrade(), track, }, participant: RemoteParticipant { room: client_room.downgrade(), identity: server_track.publisher_id.clone(), }, }) .unwrap(); } for server_track in &room.audio_tracks { let track = RemoteTrack::Audio(RemoteAudioTrack { server_track: server_track.clone(), room: client_room.downgrade(), }); client_room .0 .lock() .updates_tx .blocking_send(RoomEvent::TrackSubscribed { track: track.clone(), publication: RemoteTrackPublication { sid: server_track.sid.clone(), room: client_room.downgrade(), track, }, participant: RemoteParticipant { room: client_room.downgrade(), identity: server_track.publisher_id.clone(), }, }) .unwrap(); } e.insert(client_room); Ok(identity) } else { Err(anyhow!( "{:?} attempted to join room {:?} twice", identity, room_name )) } } async fn leave_room(&self, token: String) -> Result<()> { self.simulate_random_delay().await; let claims = livekit_api::token::validate(&token, &self.secret_key)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&*room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; room.client_rooms.remove(&identity).ok_or_else(|| { anyhow!( "{:?} attempted to leave room {:?} before joining it", identity, room_name ) })?; Ok(()) } fn remote_participants( &self, token: String, ) -> Result<HashMap<ParticipantIdentity, RemoteParticipant>> { let claims = livekit_api::token::validate(&token, &self.secret_key)?; let local_identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap().to_string(); if let Some(server_room) = self.rooms.lock().get(&room_name) { let room = server_room .client_rooms .get(&local_identity) .unwrap() .downgrade(); Ok(server_room .client_rooms .iter() .filter(|(identity, _)| *identity != &local_identity) .map(|(identity, _)| { ( identity.clone(), RemoteParticipant { room: room.clone(), identity: identity.clone(), }, ) }) .collect()) } else { Ok(Default::default()) } } async fn remove_participant( &self, room_name: String, identity: ParticipantIdentity, ) -> Result<()> { self.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; room.client_rooms.remove(&identity).ok_or_else(|| { anyhow!( "participant {:?} did not join room {:?}", identity, room_name ) })?; Ok(()) } async fn update_participant( &self, room_name: String, identity: String, permission: proto::ParticipantPermission, ) -> Result<()> { self.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; room.participant_permissions .insert(ParticipantIdentity(identity), permission); Ok(()) } pub async fn disconnect_client(&self, client_identity: String) { let client_identity = ParticipantIdentity(client_identity); self.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); for room in server_rooms.values_mut() { if let Some(room) = room.client_rooms.remove(&client_identity) { let mut room = room.0.lock(); room.connection_state = ConnectionState::Disconnected; room.updates_tx .blocking_send(RoomEvent::Disconnected { reason: "SIGNAL_CLOSED", }) .ok(); } } } pub(crate) async fn publish_video_track( &self, token: String, _local_track: LocalVideoTrack, ) -> Result<TrackSid> { self.simulate_random_delay().await; let claims = livekit_api::token::validate(&token, &self.secret_key)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&*room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; let can_publish = room .participant_permissions .get(&identity) .map(|permission| permission.can_publish) .or(claims.video.can_publish) .unwrap_or(true); if !can_publish { return Err(anyhow!("user is not allowed to publish")); } let sid: TrackSid = format!("TR_{}", nanoid::nanoid!(17)).try_into().unwrap(); let server_track = Arc::new(TestServerVideoTrack { sid: sid.clone(), publisher_id: identity.clone(), }); room.video_tracks.push(server_track.clone()); for (room_identity, client_room) in &room.client_rooms { if *room_identity != identity { let track = RemoteTrack::Video(RemoteVideoTrack { server_track: server_track.clone(), _room: client_room.downgrade(), }); let publication = RemoteTrackPublication { sid: sid.clone(), room: client_room.downgrade(), track: track.clone(), }; let participant = RemoteParticipant { identity: identity.clone(), room: client_room.downgrade(), }; client_room .0 .lock() .updates_tx .blocking_send(RoomEvent::TrackSubscribed { track, publication, participant, }) .unwrap(); } } Ok(sid) } pub(crate) async fn publish_audio_track( &self, token: String, _local_track: &LocalAudioTrack, ) -> Result<TrackSid> { self.simulate_random_delay().await; let claims = livekit_api::token::validate(&token, &self.secret_key)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&*room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; let can_publish = room .participant_permissions .get(&identity) .map(|permission| permission.can_publish) .or(claims.video.can_publish) .unwrap_or(true); if !can_publish { return Err(anyhow!("user is not allowed to publish")); } let sid: TrackSid = format!("TR_{}", nanoid::nanoid!(17)).try_into().unwrap(); let server_track = Arc::new(TestServerAudioTrack { sid: sid.clone(), publisher_id: identity.clone(), muted: AtomicBool::new(false), }); room.audio_tracks.push(server_track.clone()); for (room_identity, client_room) in &room.client_rooms { if *room_identity != identity { let track = RemoteTrack::Audio(RemoteAudioTrack { server_track: server_track.clone(), room: client_room.downgrade(), }); let publication = RemoteTrackPublication { sid: sid.clone(), room: client_room.downgrade(), track: track.clone(), }; let participant = RemoteParticipant { identity: identity.clone(), room: client_room.downgrade(), }; client_room .0 .lock() .updates_tx .blocking_send(RoomEvent::TrackSubscribed { track, publication, participant, }) .ok(); } } Ok(sid) } pub(crate) async fn unpublish_track(&self, _token: String, _track: &TrackSid) -> Result<()> { Ok(()) } pub(crate) fn set_track_muted( &self, token: &str, track_sid: &TrackSid, muted: bool, ) -> Result<()> { let claims = livekit_api::token::validate(&token, &self.secret_key)?; let room_name = claims.video.room.unwrap(); let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&*room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; if let Some(track) = room .audio_tracks .iter_mut() .find(|track| track.sid == *track_sid) { track.muted.store(muted, SeqCst); for (id, client_room) in room.client_rooms.iter() { if *id != identity { let participant = Participant::Remote(RemoteParticipant { identity: identity.clone(), room: client_room.downgrade(), }); let track = RemoteTrack::Audio(RemoteAudioTrack { server_track: track.clone(), room: client_room.downgrade(), }); let publication = TrackPublication::Remote(RemoteTrackPublication { sid: track_sid.clone(), room: client_room.downgrade(), track, }); let event = if muted { RoomEvent::TrackMuted { participant, publication, } } else { RoomEvent::TrackUnmuted { participant, publication, } }; client_room .0 .lock() .updates_tx .blocking_send(event) .unwrap(); } } } Ok(()) } pub(crate) fn is_track_muted(&self, token: &str, track_sid: &TrackSid) -> Option<bool> { let claims = livekit_api::token::validate(&token, &self.secret_key).ok()?; let room_name = claims.video.room.unwrap(); let mut server_rooms = self.rooms.lock(); let room = server_rooms.get_mut(&*room_name)?; room.audio_tracks.iter().find_map(|track| { if track.sid == *track_sid {