oak-team/oak-gpui
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Voice support (#2602)

Browse Source 
This PR adds APIs for voice chat to our live kit integration

Release Notes:

- N/A
This commit is contained in:
Mikayla Maki
2023-06-15 19:40:01 -07:00
committed by GitHub
parent 8592208d3a d7dbc3b5d3
commit 8235d78afe
6 changed files with 446 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -18,4 +18,5 @@ DerivedData/
.swiftpm/config/registries.json
.swiftpm/xcode/package.xcworkspace/contents.xcworkspacedata
.netrc
.swiftpm
**/*.db
crates/live_kit_client/LiveKitBridge/Sources/LiveKitBridge/LiveKitBridge.swift
+70
View File
@@ -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<Room>.fromOpaque(room).takeUnretainedValue()
let track = Unmanaged<LocalAudioTrack>.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<Room>.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<Room>.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<Room>.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<MacOSDisplay>.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<Track>.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<RemoteAudioTrack>.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
crates/live_kit_client/examples/test_app.rs
+64 -14
View File
@@ -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,16 +59,14 @@ 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();
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();
if let RemoteVideoTrackUpdate::Subscribed(track) = track_changes.next().await.unwrap() {
let remote_tracks = room_b.remote_video_tracks("test-participant-1");
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");
@@ -66,18 +74,60 @@ fn main() {
panic!("unexpected message");
}
let remote_track = room_b
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 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) =
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_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");
crates/live_kit_client/src/live_kit_client.rs
+1 -1
View File
@@ -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::*;
crates/live_kit_client/src/prod.rs
+190
View File
@@ -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<ConnectionState>,
watch::Receiver<ConnectionState>,
)>,
remote_audio_track_subscribers: Mutex<Vec<mpsc::UnboundedSender<RemoteAudioTrackUpdate>>>,
remote_video_track_subscribers: Mutex<Vec<mpsc::UnboundedSender<RemoteVideoTrackUpdate>>>,
_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<Self>,
track: &LocalAudioTrack,
) -> impl Future<Output = Result<LocalTrackPublication>> {
let (tx, rx) = oneshot::channel::<Result<LocalTrackPublication>>();
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<Result<LocalTrackPublication>>) };
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<Arc<RemoteAudioTrack>> {
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<RemoteAudioTrackUpdate> {
let (tx, rx) = mpsc::unbounded();
self.remote_audio_track_subscribers.lock().push(tx);
rx
}
pub fn remote_video_track_updates(&self) -> mpsc::UnboundedReceiver<RemoteVideoTrackUpdate> {
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<RemoteAudioTrack>),
Unsubscribed { publisher_id: Sid, track_id: Sid },
}
pub struct MacOSDisplay(*const c_void);
impl MacOSDisplay {
crates/live_kit_client/src/test.rs
+120 -6
View File
@@ -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<Vec<Arc<RemoteVideoTrack>>> {
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<Vec<Arc<RemoteAudioTrack>>> {
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<Sid, Arc<Room>>,
tracks: Vec<Arc<RemoteVideoTrack>>,
video_tracks: Vec<Arc<RemoteVideoTrack>>,
audio_tracks: Vec<Arc<RemoteAudioTrack>>,
}
impl TestServerRoom {}
@@ -266,6 +315,10 @@ struct RoomState {
watch::Receiver<ConnectionState>,
),
display_sources: Vec<MacOSDisplay>,
audio_track_updates: (
async_broadcast::Sender<RemoteAudioTrackUpdate>,
async_broadcast::Receiver<RemoteAudioTrackUpdate>,
),
video_track_updates: (
async_broadcast::Sender<RemoteVideoTrackUpdate>,
async_broadcast::Receiver<RemoteVideoTrackUpdate>,
@@ -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<Self>,
track: &LocalAudioTrack,
) -> impl Future<Output = Result<LocalTrackPublication>> {
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<Arc<RemoteAudioTrack>> {
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<Arc<RemoteVideoTrack>> {
if !self.is_connected() {
@@ -343,6 +423,10 @@ impl Room {
.collect()
}
pub fn remote_audio_track_updates(&self) -> impl Stream<Item = RemoteAudioTrackUpdate> {
self.0.lock().audio_track_updates.1.clone()
}
pub fn remote_video_track_updates(&self) -> impl Stream<Item = RemoteVideoTrackUpdate> {
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<RemoteVideoTrack>),
Unsubscribed { publisher_id: Sid, track_id: Sid },
}
#[derive(Clone)]
pub enum RemoteAudioTrackUpdate {
Subscribed(Arc<RemoteAudioTrack>),
Unsubscribed { publisher_id: Sid, track_id: Sid },
}
#[derive(Clone)]
pub struct MacOSDisplay {
frames: (