diff --git a/.gitignore b/.gitignore index 5a4d2ff25e19e7406f79c1b627510f4c94d58212..30d0fcbf1c8e5786d4c6145c5ae6d7043d6d5590 100644 --- a/.gitignore +++ b/.gitignore @@ -18,4 +18,5 @@ DerivedData/ .swiftpm/config/registries.json .swiftpm/xcode/package.xcworkspace/contents.xcworkspacedata .netrc +.swiftpm **/*.db diff --git a/crates/live_kit_client/LiveKitBridge/Sources/LiveKitBridge/LiveKitBridge.swift b/crates/live_kit_client/LiveKitBridge/Sources/LiveKitBridge/LiveKitBridge.swift index a0326b24a1c480d55bc307b178f6562cba1b3d07..40c1319e8f9111dac6139e723699ba76fd8d6111 100644 --- a/crates/live_kit_client/LiveKitBridge/Sources/LiveKitBridge/LiveKitBridge.swift +++ b/crates/live_kit_client/LiveKitBridge/Sources/LiveKitBridge/LiveKitBridge.swift @@ -6,17 +6,23 @@ import ScreenCaptureKit class LKRoomDelegate: RoomDelegate { var data: UnsafeRawPointer var onDidDisconnect: @convention(c) (UnsafeRawPointer) -> Void + var onDidSubscribeToRemoteAudioTrack: @convention(c) (UnsafeRawPointer, CFString, CFString, UnsafeRawPointer) -> Void + var onDidUnsubscribeFromRemoteAudioTrack: @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void var onDidSubscribeToRemoteVideoTrack: @convention(c) (UnsafeRawPointer, CFString, CFString, UnsafeRawPointer) -> Void var onDidUnsubscribeFromRemoteVideoTrack: @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void init( data: UnsafeRawPointer, onDidDisconnect: @escaping @convention(c) (UnsafeRawPointer) -> Void, + onDidSubscribeToRemoteAudioTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString, UnsafeRawPointer) -> Void, + onDidUnsubscribeFromRemoteAudioTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void, onDidSubscribeToRemoteVideoTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString, UnsafeRawPointer) -> Void, onDidUnsubscribeFromRemoteVideoTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void) { self.data = data self.onDidDisconnect = onDidDisconnect + self.onDidSubscribeToRemoteAudioTrack = onDidSubscribeToRemoteAudioTrack + self.onDidUnsubscribeFromRemoteAudioTrack = onDidUnsubscribeFromRemoteAudioTrack self.onDidSubscribeToRemoteVideoTrack = onDidSubscribeToRemoteVideoTrack self.onDidUnsubscribeFromRemoteVideoTrack = onDidUnsubscribeFromRemoteVideoTrack } @@ -30,12 +36,16 @@ class LKRoomDelegate: RoomDelegate { func room(_ room: Room, participant: RemoteParticipant, didSubscribe publication: RemoteTrackPublication, track: Track) { if track.kind == .video { self.onDidSubscribeToRemoteVideoTrack(self.data, participant.identity as CFString, track.sid! as CFString, Unmanaged.passUnretained(track).toOpaque()) + } else if track.kind == .audio { + self.onDidSubscribeToRemoteAudioTrack(self.data, participant.identity as CFString, track.sid! as CFString, Unmanaged.passUnretained(track).toOpaque()) } } func room(_ room: Room, participant: RemoteParticipant, didUnsubscribe publication: RemoteTrackPublication, track: Track) { if track.kind == .video { self.onDidUnsubscribeFromRemoteVideoTrack(self.data, participant.identity as CFString, track.sid! as CFString) + } else if track.kind == .audio { + self.onDidUnsubscribeFromRemoteAudioTrack(self.data, participant.identity as CFString, track.sid! as CFString) } } } @@ -77,12 +87,16 @@ class LKVideoRenderer: NSObject, VideoRenderer { public func LKRoomDelegateCreate( data: UnsafeRawPointer, onDidDisconnect: @escaping @convention(c) (UnsafeRawPointer) -> Void, + onDidSubscribeToRemoteAudioTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString, UnsafeRawPointer) -> Void, + onDidUnsubscribeFromRemoteAudioTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void, onDidSubscribeToRemoteVideoTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString, UnsafeRawPointer) -> Void, onDidUnsubscribeFromRemoteVideoTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void ) -> UnsafeMutableRawPointer { let delegate = LKRoomDelegate( data: data, onDidDisconnect: onDidDisconnect, + onDidSubscribeToRemoteAudioTrack: onDidSubscribeToRemoteAudioTrack, + onDidUnsubscribeFromRemoteAudioTrack: onDidUnsubscribeFromRemoteAudioTrack, onDidSubscribeToRemoteVideoTrack: onDidSubscribeToRemoteVideoTrack, onDidUnsubscribeFromRemoteVideoTrack: onDidUnsubscribeFromRemoteVideoTrack ) @@ -123,6 +137,18 @@ public func LKRoomPublishVideoTrack(room: UnsafeRawPointer, track: UnsafeRawPoin } } +@_cdecl("LKRoomPublishAudioTrack") +public func LKRoomPublishAudioTrack(room: UnsafeRawPointer, track: UnsafeRawPointer, callback: @escaping @convention(c) (UnsafeRawPointer, UnsafeMutableRawPointer?, CFString?) -> Void, callback_data: UnsafeRawPointer) { + let room = Unmanaged.fromOpaque(room).takeUnretainedValue() + let track = Unmanaged.fromOpaque(track).takeUnretainedValue() + room.localParticipant?.publishAudioTrack(track: track).then { publication in + callback(callback_data, Unmanaged.passRetained(publication).toOpaque(), nil) + }.catch { error in + callback(callback_data, nil, error.localizedDescription as CFString) + } +} + + @_cdecl("LKRoomUnpublishTrack") public func LKRoomUnpublishTrack(room: UnsafeRawPointer, publication: UnsafeRawPointer) { let room = Unmanaged.fromOpaque(room).takeUnretainedValue() @@ -130,6 +156,20 @@ public func LKRoomUnpublishTrack(room: UnsafeRawPointer, publication: UnsafeRawP let _ = room.localParticipant?.unpublish(publication: publication) } +@_cdecl("LKRoomAudioTracksForRemoteParticipant") +public func LKRoomAudioTracksForRemoteParticipant(room: UnsafeRawPointer, participantId: CFString) -> CFArray? { + let room = Unmanaged.fromOpaque(room).takeUnretainedValue() + + for (_, participant) in room.remoteParticipants { + if participant.identity == participantId as String { + return participant.audioTracks.compactMap { $0.track as? RemoteAudioTrack } as CFArray? + } + } + + return nil; +} + + @_cdecl("LKRoomVideoTracksForRemoteParticipant") public func LKRoomVideoTracksForRemoteParticipant(room: UnsafeRawPointer, participantId: CFString) -> CFArray? { let room = Unmanaged.fromOpaque(room).takeUnretainedValue() @@ -143,6 +183,17 @@ public func LKRoomVideoTracksForRemoteParticipant(room: UnsafeRawPointer, partic return nil; } +@_cdecl("LKLocalAudioTrackCreateTrack") +public func LKLocalAudioTrackCreateTrack() -> UnsafeMutableRawPointer { + let track = LocalAudioTrack.createTrack(options: AudioCaptureOptions( + echoCancellation: true, + noiseSuppression: true + )) + + return Unmanaged.passRetained(track).toOpaque() +} + + @_cdecl("LKCreateScreenShareTrackForDisplay") public func LKCreateScreenShareTrackForDisplay(display: UnsafeMutableRawPointer) -> UnsafeMutableRawPointer { let display = Unmanaged.fromOpaque(display).takeUnretainedValue() @@ -150,6 +201,19 @@ public func LKCreateScreenShareTrackForDisplay(display: UnsafeMutableRawPointer) return Unmanaged.passRetained(track).toOpaque() } +@_cdecl("LKRemoteAudioTrackStart") +public func LKRemoteAudioTrackStart(track: UnsafeRawPointer, onStart: @escaping @convention(c) (UnsafeRawPointer, Bool) -> Void, callbackData: UnsafeRawPointer) { + let track = Unmanaged.fromOpaque(track).takeUnretainedValue() as! RemoteAudioTrack + + track.start().then { success in + onStart(callbackData, success) + } + .catch { _ in + onStart(callbackData, false) + } +} + + @_cdecl("LKVideoRendererCreate") public func LKVideoRendererCreate(data: UnsafeRawPointer, onFrame: @escaping @convention(c) (UnsafeRawPointer, CVPixelBuffer) -> Bool, onDrop: @escaping @convention(c) (UnsafeRawPointer) -> Void) -> UnsafeMutableRawPointer { Unmanaged.passRetained(LKVideoRenderer(data: data, onFrame: onFrame, onDrop: onDrop)).toOpaque() @@ -169,6 +233,12 @@ public func LKRemoteVideoTrackGetSid(track: UnsafeRawPointer) -> CFString { return track.sid! as CFString } +@_cdecl("LKRemoteAudioTrackGetSid") +public func LKRemoteAudioTrackGetSid(track: UnsafeRawPointer) -> CFString { + let track = Unmanaged.fromOpaque(track).takeUnretainedValue() + return track.sid! as CFString +} + @_cdecl("LKDisplaySources") public func LKDisplaySources(data: UnsafeRawPointer, callback: @escaping @convention(c) (UnsafeRawPointer, CFArray?, CFString?) -> Void) { MacOSScreenCapturer.sources(for: .display, includeCurrentApplication: false, preferredMethod: .legacy).then { displaySources in diff --git a/crates/live_kit_client/examples/test_app.rs b/crates/live_kit_client/examples/test_app.rs index 96480e92bc53095e68d80c5d86e6b12c5e7de15b..faf1b54798f3722fe335fdf0e27748b94e9aba9e 100644 --- a/crates/live_kit_client/examples/test_app.rs +++ b/crates/live_kit_client/examples/test_app.rs @@ -1,6 +1,10 @@ +use std::time::Duration; + use futures::StreamExt; use gpui::{actions, keymap_matcher::Binding, Menu, MenuItem}; -use live_kit_client::{LocalVideoTrack, RemoteVideoTrackUpdate, Room}; +use live_kit_client::{ + LocalAudioTrack, LocalVideoTrack, RemoteAudioTrackUpdate, RemoteVideoTrackUpdate, Room, +}; use live_kit_server::token::{self, VideoGrant}; use log::LevelFilter; use simplelog::SimpleLogger; @@ -11,6 +15,12 @@ fn main() { SimpleLogger::init(LevelFilter::Info, Default::default()).expect("could not initialize logger"); gpui::App::new(()).unwrap().run(|cx| { + #[cfg(any(test, feature = "test-support"))] + println!("USING TEST LIVEKIT"); + + #[cfg(not(any(test, feature = "test-support")))] + println!("USING REAL LIVEKIT"); + cx.platform().activate(true); cx.add_global_action(quit); @@ -49,35 +59,75 @@ fn main() { let room_b = Room::new(); room_b.connect(&live_kit_url, &user2_token).await.unwrap(); - let mut track_changes = room_b.remote_video_track_updates(); + let mut audio_track_updates = room_b.remote_audio_track_updates(); + let audio_track = LocalAudioTrack::create(); + let audio_track_publication = room_a.publish_audio_track(&audio_track).await.unwrap(); + + if let RemoteAudioTrackUpdate::Subscribed(track) = + audio_track_updates.next().await.unwrap() + { + let remote_tracks = room_b.remote_audio_tracks("test-participant-1"); + assert_eq!(remote_tracks.len(), 1); + assert_eq!(remote_tracks[0].publisher_id(), "test-participant-1"); + assert_eq!(track.publisher_id(), "test-participant-1"); + } else { + panic!("unexpected message"); + } + + println!("Pausing for 5 seconds to test audio, make some noise!"); + let timer = cx.background().timer(Duration::from_secs(5)); + timer.await; + + let remote_audio_track = room_b + .remote_audio_tracks("test-participant-1") + .pop() + .unwrap(); + room_a.unpublish_track(audio_track_publication); + if let RemoteAudioTrackUpdate::Unsubscribed { + publisher_id, + track_id, + } = audio_track_updates.next().await.unwrap() + { + assert_eq!(publisher_id, "test-participant-1"); + assert_eq!(remote_audio_track.sid(), track_id); + assert_eq!(room_b.remote_audio_tracks("test-participant-1").len(), 0); + } else { + panic!("unexpected message"); + } + let mut video_track_updates = room_b.remote_video_track_updates(); let displays = room_a.display_sources().await.unwrap(); let display = displays.into_iter().next().unwrap(); - let track_a = LocalVideoTrack::screen_share_for_display(&display); - let track_a_publication = room_a.publish_video_track(&track_a).await.unwrap(); + let local_video_track = LocalVideoTrack::screen_share_for_display(&display); + let local_video_track_publication = room_a + .publish_video_track(&local_video_track) + .await + .unwrap(); - if let RemoteVideoTrackUpdate::Subscribed(track) = track_changes.next().await.unwrap() { - let remote_tracks = room_b.remote_video_tracks("test-participant-1"); - assert_eq!(remote_tracks.len(), 1); - assert_eq!(remote_tracks[0].publisher_id(), "test-participant-1"); + if let RemoteVideoTrackUpdate::Subscribed(track) = + video_track_updates.next().await.unwrap() + { + let remote_video_tracks = room_b.remote_video_tracks("test-participant-1"); + assert_eq!(remote_video_tracks.len(), 1); + assert_eq!(remote_video_tracks[0].publisher_id(), "test-participant-1"); assert_eq!(track.publisher_id(), "test-participant-1"); } else { panic!("unexpected message"); } - let remote_track = room_b + let remote_video_track = room_b .remote_video_tracks("test-participant-1") .pop() .unwrap(); - room_a.unpublish_track(track_a_publication); + room_a.unpublish_track(local_video_track_publication); if let RemoteVideoTrackUpdate::Unsubscribed { publisher_id, track_id, - } = track_changes.next().await.unwrap() + } = video_track_updates.next().await.unwrap() { assert_eq!(publisher_id, "test-participant-1"); - assert_eq!(remote_track.sid(), track_id); + assert_eq!(remote_video_track.sid(), track_id); assert_eq!(room_b.remote_video_tracks("test-participant-1").len(), 0); } else { panic!("unexpected message"); diff --git a/crates/live_kit_client/src/live_kit_client.rs b/crates/live_kit_client/src/live_kit_client.rs index 2ded57082801428e35ff5ce87d1b8740db0fffb7..0467018c009a9b36f00908d6628311d580fdec91 100644 --- a/crates/live_kit_client/src/live_kit_client.rs +++ b/crates/live_kit_client/src/live_kit_client.rs @@ -4,7 +4,7 @@ pub mod prod; pub use prod::*; #[cfg(any(test, feature = "test-support"))] -mod test; +pub mod test; #[cfg(any(test, feature = "test-support"))] pub use test::*; diff --git a/crates/live_kit_client/src/prod.rs b/crates/live_kit_client/src/prod.rs index f45667e3c3e4b4d00f264b51ff4782e8471e1ebd..6ae493d306136f66f3a6212ccf23ec93b2f50f93 100644 --- a/crates/live_kit_client/src/prod.rs +++ b/crates/live_kit_client/src/prod.rs @@ -21,6 +21,17 @@ extern "C" { fn LKRoomDelegateCreate( callback_data: *mut c_void, on_did_disconnect: extern "C" fn(callback_data: *mut c_void), + on_did_subscribe_to_remote_audio_track: extern "C" fn( + callback_data: *mut c_void, + publisher_id: CFStringRef, + track_id: CFStringRef, + remote_track: *const c_void, + ), + on_did_unsubscribe_from_remote_audio_track: extern "C" fn( + callback_data: *mut c_void, + publisher_id: CFStringRef, + track_id: CFStringRef, + ), on_did_subscribe_to_remote_video_track: extern "C" fn( callback_data: *mut c_void, publisher_id: CFStringRef, @@ -49,7 +60,18 @@ extern "C" { callback: extern "C" fn(*mut c_void, *mut c_void, CFStringRef), callback_data: *mut c_void, ); + fn LKRoomPublishAudioTrack( + room: *const c_void, + track: *const c_void, + callback: extern "C" fn(*mut c_void, *mut c_void, CFStringRef), + callback_data: *mut c_void, + ); fn LKRoomUnpublishTrack(room: *const c_void, publication: *const c_void); + fn LKRoomAudioTracksForRemoteParticipant( + room: *const c_void, + participant_id: CFStringRef, + ) -> CFArrayRef; + fn LKRoomVideoTracksForRemoteParticipant( room: *const c_void, participant_id: CFStringRef, @@ -61,6 +83,13 @@ extern "C" { on_drop: extern "C" fn(callback_data: *mut c_void), ) -> *const c_void; + fn LKRemoteAudioTrackGetSid(track: *const c_void) -> CFStringRef; + // fn LKRemoteAudioTrackStart( + // track: *const c_void, + // callback: extern "C" fn(*mut c_void, bool), + // callback_data: *mut c_void + // ); + fn LKVideoTrackAddRenderer(track: *const c_void, renderer: *const c_void); fn LKRemoteVideoTrackGetSid(track: *const c_void) -> CFStringRef; @@ -73,6 +102,7 @@ extern "C" { ), ); fn LKCreateScreenShareTrackForDisplay(display: *const c_void) -> *const c_void; + fn LKLocalAudioTrackCreateTrack() -> *const c_void; } pub type Sid = String; @@ -89,6 +119,7 @@ pub struct Room { watch::Sender, watch::Receiver, )>, + remote_audio_track_subscribers: Mutex>>, remote_video_track_subscribers: Mutex>>, _delegate: RoomDelegate, } @@ -100,6 +131,7 @@ impl Room { Self { native_room: unsafe { LKRoomCreate(delegate.native_delegate) }, connection: Mutex::new(watch::channel_with(ConnectionState::Disconnected)), + remote_audio_track_subscribers: Default::default(), remote_video_track_subscribers: Default::default(), _delegate: delegate, } @@ -191,6 +223,32 @@ impl Room { async { rx.await.unwrap().context("error publishing video track") } } + pub fn publish_audio_track( + self: &Arc, + track: &LocalAudioTrack, + ) -> impl Future> { + let (tx, rx) = oneshot::channel::>(); + extern "C" fn callback(tx: *mut c_void, publication: *mut c_void, error: CFStringRef) { + let tx = + unsafe { Box::from_raw(tx as *mut oneshot::Sender>) }; + if error.is_null() { + let _ = tx.send(Ok(LocalTrackPublication(publication))); + } else { + let error = unsafe { CFString::wrap_under_get_rule(error).to_string() }; + let _ = tx.send(Err(anyhow!(error))); + } + } + unsafe { + LKRoomPublishAudioTrack( + self.native_room, + track.0, + callback, + Box::into_raw(Box::new(tx)) as *mut c_void, + ); + } + async { rx.await.unwrap().context("error publishing video track") } + } + pub fn unpublish_track(&self, publication: LocalTrackPublication) { unsafe { LKRoomUnpublishTrack(self.native_room, publication.0); @@ -226,12 +284,65 @@ impl Room { } } + pub fn remote_audio_tracks(&self, participant_id: &str) -> Vec> { + unsafe { + let tracks = LKRoomAudioTracksForRemoteParticipant( + self.native_room, + CFString::new(participant_id).as_concrete_TypeRef(), + ); + + if tracks.is_null() { + Vec::new() + } else { + let tracks = CFArray::wrap_under_get_rule(tracks); + tracks + .into_iter() + .map(|native_track| { + let native_track = *native_track; + let id = + CFString::wrap_under_get_rule(LKRemoteAudioTrackGetSid(native_track)) + .to_string(); + Arc::new(RemoteAudioTrack::new( + native_track, + id, + participant_id.into(), + )) + }) + .collect() + } + } + } + + pub fn remote_audio_track_updates(&self) -> mpsc::UnboundedReceiver { + let (tx, rx) = mpsc::unbounded(); + self.remote_audio_track_subscribers.lock().push(tx); + rx + } + pub fn remote_video_track_updates(&self) -> mpsc::UnboundedReceiver { let (tx, rx) = mpsc::unbounded(); self.remote_video_track_subscribers.lock().push(tx); rx } + fn did_subscribe_to_remote_audio_track(&self, track: RemoteAudioTrack) { + let track = Arc::new(track); + self.remote_audio_track_subscribers.lock().retain(|tx| { + tx.unbounded_send(RemoteAudioTrackUpdate::Subscribed(track.clone())) + .is_ok() + }); + } + + fn did_unsubscribe_from_remote_audio_track(&self, publisher_id: String, track_id: String) { + self.remote_audio_track_subscribers.lock().retain(|tx| { + tx.unbounded_send(RemoteAudioTrackUpdate::Unsubscribed { + publisher_id: publisher_id.clone(), + track_id: track_id.clone(), + }) + .is_ok() + }); + } + fn did_subscribe_to_remote_video_track(&self, track: RemoteVideoTrack) { let track = Arc::new(track); self.remote_video_track_subscribers.lock().retain(|tx| { @@ -294,6 +405,8 @@ impl RoomDelegate { LKRoomDelegateCreate( weak_room as *mut c_void, Self::on_did_disconnect, + Self::on_did_subscribe_to_remote_audio_track, + Self::on_did_unsubscribe_from_remote_audio_track, Self::on_did_subscribe_to_remote_video_track, Self::on_did_unsubscribe_from_remote_video_track, ) @@ -312,6 +425,36 @@ impl RoomDelegate { let _ = Weak::into_raw(room); } + extern "C" fn on_did_subscribe_to_remote_audio_track( + room: *mut c_void, + publisher_id: CFStringRef, + track_id: CFStringRef, + track: *const c_void, + ) { + let room = unsafe { Weak::from_raw(room as *mut Room) }; + let publisher_id = unsafe { CFString::wrap_under_get_rule(publisher_id).to_string() }; + let track_id = unsafe { CFString::wrap_under_get_rule(track_id).to_string() }; + let track = RemoteAudioTrack::new(track, track_id, publisher_id); + if let Some(room) = room.upgrade() { + room.did_subscribe_to_remote_audio_track(track); + } + let _ = Weak::into_raw(room); + } + + extern "C" fn on_did_unsubscribe_from_remote_audio_track( + room: *mut c_void, + publisher_id: CFStringRef, + track_id: CFStringRef, + ) { + let room = unsafe { Weak::from_raw(room as *mut Room) }; + let publisher_id = unsafe { CFString::wrap_under_get_rule(publisher_id).to_string() }; + let track_id = unsafe { CFString::wrap_under_get_rule(track_id).to_string() }; + if let Some(room) = room.upgrade() { + room.did_unsubscribe_from_remote_audio_track(publisher_id, track_id); + } + let _ = Weak::into_raw(room); + } + extern "C" fn on_did_subscribe_to_remote_video_track( room: *mut c_void, publisher_id: CFStringRef, @@ -352,6 +495,20 @@ impl Drop for RoomDelegate { } } +pub struct LocalAudioTrack(*const c_void); + +impl LocalAudioTrack { + pub fn create() -> Self { + Self(unsafe { LKLocalAudioTrackCreateTrack() }) + } +} + +impl Drop for LocalAudioTrack { + fn drop(&mut self) { + unsafe { CFRelease(self.0) } + } +} + pub struct LocalVideoTrack(*const c_void); impl LocalVideoTrack { @@ -374,6 +531,34 @@ impl Drop for LocalTrackPublication { } } +#[derive(Debug)] +pub struct RemoteAudioTrack { + _native_track: *const c_void, + sid: Sid, + publisher_id: String, +} + +impl RemoteAudioTrack { + fn new(native_track: *const c_void, sid: Sid, publisher_id: String) -> Self { + unsafe { + CFRetain(native_track); + } + Self { + _native_track: native_track, + sid, + publisher_id, + } + } + + pub fn sid(&self) -> &str { + &self.sid + } + + pub fn publisher_id(&self) -> &str { + &self.publisher_id + } +} + #[derive(Debug)] pub struct RemoteVideoTrack { native_track: *const c_void, @@ -453,6 +638,11 @@ pub enum RemoteVideoTrackUpdate { Unsubscribed { publisher_id: Sid, track_id: Sid }, } +pub enum RemoteAudioTrackUpdate { + Subscribed(Arc), + Unsubscribed { publisher_id: Sid, track_id: Sid }, +} + pub struct MacOSDisplay(*const c_void); impl MacOSDisplay { diff --git a/crates/live_kit_client/src/test.rs b/crates/live_kit_client/src/test.rs index 8d1e4fa16abca12af30878779b7cf7d403f74250..2f89617363db7a9e9f51251ebf70b5e8b0eea482 100644 --- a/crates/live_kit_client/src/test.rs +++ b/crates/live_kit_client/src/test.rs @@ -67,7 +67,7 @@ impl TestServer { } } - async fn create_room(&self, room: String) -> Result<()> { + pub async fn create_room(&self, room: String) -> Result<()> { self.background.simulate_random_delay().await; let mut server_rooms = self.rooms.lock(); if server_rooms.contains_key(&room) { @@ -104,7 +104,7 @@ impl TestServer { room_name )) } else { - for track in &room.tracks { + for track in &room.video_tracks { client_room .0 .lock() @@ -182,7 +182,7 @@ impl TestServer { frames_rx: local_track.frames_rx.clone(), }); - room.tracks.push(track.clone()); + room.video_tracks.push(track.clone()); for (id, client_room) in &room.client_rooms { if *id != identity { @@ -199,6 +199,43 @@ impl TestServer { Ok(()) } + async fn publish_audio_track( + &self, + token: String, + _local_track: &LocalAudioTrack, + ) -> Result<()> { + self.background.simulate_random_delay().await; + let claims = live_kit_server::token::validate(&token, &self.secret_key)?; + let identity = 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 track = Arc::new(RemoteAudioTrack { + sid: nanoid::nanoid!(17), + publisher_id: identity.clone(), + }); + + room.audio_tracks.push(track.clone()); + + for (id, client_room) in &room.client_rooms { + if *id != identity { + let _ = client_room + .0 + .lock() + .audio_track_updates + .0 + .try_broadcast(RemoteAudioTrackUpdate::Subscribed(track.clone())) + .unwrap(); + } + } + + Ok(()) + } + fn video_tracks(&self, token: String) -> Result>> { let claims = live_kit_server::token::validate(&token, &self.secret_key)?; let room_name = claims.video.room.unwrap(); @@ -207,14 +244,26 @@ impl TestServer { let room = server_rooms .get_mut(&*room_name) .ok_or_else(|| anyhow!("room {} does not exist", room_name))?; - Ok(room.tracks.clone()) + Ok(room.video_tracks.clone()) + } + + fn audio_tracks(&self, token: String) -> Result>> { + let claims = live_kit_server::token::validate(&token, &self.secret_key)?; + 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))?; + Ok(room.audio_tracks.clone()) } } #[derive(Default)] struct TestServerRoom { client_rooms: HashMap>, - tracks: Vec>, + video_tracks: Vec>, + audio_tracks: Vec>, } impl TestServerRoom {} @@ -266,6 +315,10 @@ struct RoomState { watch::Receiver, ), display_sources: Vec, + audio_track_updates: ( + async_broadcast::Sender, + async_broadcast::Receiver, + ), video_track_updates: ( async_broadcast::Sender, async_broadcast::Receiver, @@ -286,6 +339,7 @@ impl Room { connection: watch::channel_with(ConnectionState::Disconnected), display_sources: Default::default(), video_track_updates: async_broadcast::broadcast(128), + audio_track_updates: async_broadcast::broadcast(128), }))) } @@ -327,8 +381,34 @@ impl Room { Ok(LocalTrackPublication) } } + pub fn publish_audio_track( + self: &Arc, + track: &LocalAudioTrack, + ) -> impl Future> { + let this = self.clone(); + let track = track.clone(); + async move { + this.test_server() + .publish_audio_track(this.token(), &track) + .await?; + Ok(LocalTrackPublication) + } + } - pub fn unpublish_track(&self, _: LocalTrackPublication) {} + pub fn unpublish_track(&self, _publication: LocalTrackPublication) {} + + pub fn remote_audio_tracks(&self, publisher_id: &str) -> Vec> { + if !self.is_connected() { + return Vec::new(); + } + + self.test_server() + .audio_tracks(self.token()) + .unwrap() + .into_iter() + .filter(|track| track.publisher_id() == publisher_id) + .collect() + } pub fn remote_video_tracks(&self, publisher_id: &str) -> Vec> { if !self.is_connected() { @@ -343,6 +423,10 @@ impl Room { .collect() } + pub fn remote_audio_track_updates(&self) -> impl Stream { + self.0.lock().audio_track_updates.1.clone() + } + pub fn remote_video_track_updates(&self) -> impl Stream { self.0.lock().video_track_updates.1.clone() } @@ -404,6 +488,15 @@ impl LocalVideoTrack { } } +#[derive(Clone)] +pub struct LocalAudioTrack; + +impl LocalAudioTrack { + pub fn create() -> Self { + Self + } +} + pub struct RemoteVideoTrack { sid: Sid, publisher_id: Sid, @@ -424,12 +517,33 @@ impl RemoteVideoTrack { } } +pub struct RemoteAudioTrack { + sid: Sid, + publisher_id: Sid, +} + +impl RemoteAudioTrack { + pub fn sid(&self) -> &str { + &self.sid + } + + pub fn publisher_id(&self) -> &str { + &self.publisher_id + } +} + #[derive(Clone)] pub enum RemoteVideoTrackUpdate { Subscribed(Arc), Unsubscribed { publisher_id: Sid, track_id: Sid }, } +#[derive(Clone)] +pub enum RemoteAudioTrackUpdate { + Subscribed(Arc), + Unsubscribed { publisher_id: Sid, track_id: Sid }, +} + #[derive(Clone)] pub struct MacOSDisplay { frames: (