Handle reconnects to the livekit server in which local tracks are unpublished (#3992)

Release notes:

* Fixed a bug where network interruptions could cause audio and screen
sharing to stop working without indicating that they were stopped, and
there was no way to restart the audio stream.

crates/call/src/room.rs

@@ -15,10 +15,7 @@ use gpui::{
     AppContext, AsyncAppContext, Context, EventEmitter, Model, ModelContext, Task, WeakModel,
 };
 use language::LanguageRegistry;
-use live_kit_client::{
-    LocalAudioTrack, LocalTrackPublication, LocalVideoTrack, RemoteAudioTrackUpdate,
-    RemoteVideoTrackUpdate,
-};
+use live_kit_client::{LocalAudioTrack, LocalTrackPublication, LocalVideoTrack, RoomUpdate};
 use postage::{sink::Sink, stream::Stream, watch};
 use project::Project;
 use settings::Settings as _;
@@ -131,11 +128,11 @@ impl Room {
                 }
             });
 
-            let _maintain_video_tracks = cx.spawn({
+            let _handle_updates = cx.spawn({
                 let room = room.clone();
                 move |this, mut cx| async move {
-                    let mut track_video_changes = room.remote_video_track_updates();
-                    while let Some(track_change) = track_video_changes.next().await {
+                    let mut updates = room.updates();
+                    while let Some(update) = updates.next().await {
                         let this = if let Some(this) = this.upgrade() {
                             this
                         } else {
@@ -143,26 +140,7 @@ impl Room {
                         };
 
                         this.update(&mut cx, |this, cx| {
-                            this.remote_video_track_updated(track_change, cx).log_err()
-                        })
-                        .ok();
-                    }
-                }
-            });
-
-            let _maintain_audio_tracks = cx.spawn({
-                let room = room.clone();
-                |this, mut cx| async move {
-                    let mut track_audio_changes = room.remote_audio_track_updates();
-                    while let Some(track_change) = track_audio_changes.next().await {
-                        let this = if let Some(this) = this.upgrade() {
-                            this
-                        } else {
-                            break;
-                        };
-
-                        this.update(&mut cx, |this, cx| {
-                            this.remote_audio_track_updated(track_change, cx).log_err()
+                            this.live_kit_room_updated(update, cx).log_err()
                         })
                         .ok();
                     }
@@ -195,7 +173,7 @@ impl Room {
                 deafened: false,
                 speaking: false,
                 _maintain_room,
-                _maintain_tracks: [_maintain_video_tracks, _maintain_audio_tracks],
+                _handle_updates,
             })
         } else {
             None
@@ -877,8 +855,8 @@ impl Room {
                                     .remote_audio_track_publications(&user.id.to_string());
 
                                 for track in video_tracks {
-                                    this.remote_video_track_updated(
-                                        RemoteVideoTrackUpdate::Subscribed(track),
+                                    this.live_kit_room_updated(
+                                        RoomUpdate::SubscribedToRemoteVideoTrack(track),
                                         cx,
                                     )
                                     .log_err();
@@ -887,8 +865,8 @@ impl Room {
                                 for (track, publication) in
                                     audio_tracks.iter().zip(publications.iter())
                                 {
-                                    this.remote_audio_track_updated(
-                                        RemoteAudioTrackUpdate::Subscribed(
+                                    this.live_kit_room_updated(
+                                        RoomUpdate::SubscribedToRemoteAudioTrack(
                                             track.clone(),
                                             publication.clone(),
                                         ),
@@ -979,13 +957,13 @@ impl Room {
         }
     }
 
-    fn remote_video_track_updated(
+    fn live_kit_room_updated(
         &mut self,
-        change: RemoteVideoTrackUpdate,
+        update: RoomUpdate,
         cx: &mut ModelContext<Self>,
     ) -> Result<()> {
-        match change {
-            RemoteVideoTrackUpdate::Subscribed(track) => {
+        match update {
+            RoomUpdate::SubscribedToRemoteVideoTrack(track) => {
                 let user_id = track.publisher_id().parse()?;
                 let track_id = track.sid().to_string();
                 let participant = self
@@ -997,7 +975,8 @@ impl Room {
                     participant_id: participant.peer_id,
                 });
             }
-            RemoteVideoTrackUpdate::Unsubscribed {
+
+            RoomUpdate::UnsubscribedFromRemoteVideoTrack {
                 publisher_id,
                 track_id,
             } => {
@@ -1011,19 +990,8 @@ impl Room {
                     participant_id: participant.peer_id,
                 });
             }
-        }
 
-        cx.notify();
-        Ok(())
-    }
-
-    fn remote_audio_track_updated(
-        &mut self,
-        change: RemoteAudioTrackUpdate,
-        cx: &mut ModelContext<Self>,
-    ) -> Result<()> {
-        match change {
-            RemoteAudioTrackUpdate::ActiveSpeakersChanged { speakers } => {
+            RoomUpdate::ActiveSpeakersChanged { speakers } => {
                 let mut speaker_ids = speakers
                     .into_iter()
                     .filter_map(|speaker_sid| speaker_sid.parse().ok())
@@ -1045,9 +1013,9 @@ impl Room {
                         }
                     }
                 }
-                cx.notify();
             }
-            RemoteAudioTrackUpdate::MuteChanged { track_id, muted } => {
+
+            RoomUpdate::RemoteAudioTrackMuteChanged { track_id, muted } => {
                 let mut found = false;
                 for participant in &mut self.remote_participants.values_mut() {
                     for track in participant.audio_tracks.values() {
@@ -1061,10 +1029,9 @@ impl Room {
                         break;
                     }
                 }
-
-                cx.notify();
             }
-            RemoteAudioTrackUpdate::Subscribed(track, publication) => {
+
+            RoomUpdate::SubscribedToRemoteAudioTrack(track, publication) => {
                 let user_id = track.publisher_id().parse()?;
                 let track_id = track.sid().to_string();
                 let participant = self
@@ -1078,7 +1045,8 @@ impl Room {
                     participant_id: participant.peer_id,
                 });
             }
-            RemoteAudioTrackUpdate::Unsubscribed {
+
+            RoomUpdate::UnsubscribedFromRemoteAudioTrack {
                 publisher_id,
                 track_id,
             } => {
@@ -1092,6 +1060,28 @@ impl Room {
                     participant_id: participant.peer_id,
                 });
             }
+
+            RoomUpdate::LocalAudioTrackUnpublished { publication } => {
+                log::info!("unpublished audio track {}", publication.sid());
+                if let Some(room) = &mut self.live_kit {
+                    room.microphone_track = LocalTrack::None;
+                }
+            }
+
+            RoomUpdate::LocalVideoTrackUnpublished { publication } => {
+                log::info!("unpublished video track {}", publication.sid());
+                if let Some(room) = &mut self.live_kit {
+                    room.screen_track = LocalTrack::None;
+                }
+            }
+
+            RoomUpdate::LocalAudioTrackPublished { publication } => {
+                log::info!("published audio track {}", publication.sid());
+            }
+
+            RoomUpdate::LocalVideoTrackPublished { publication } => {
+                log::info!("published video track {}", publication.sid());
+            }
         }
 
         cx.notify();
@@ -1597,7 +1587,7 @@ struct LiveKitRoom {
     speaking: bool,
     next_publish_id: usize,
     _maintain_room: Task<()>,
-    _maintain_tracks: [Task<()>; 2],
+    _handle_updates: Task<()>,
 }
 
 impl LiveKitRoom {

crates/live_kit_client/LiveKitBridge/Sources/LiveKitBridge/LiveKitBridge.swift

@@ -12,6 +12,8 @@ class LKRoomDelegate: RoomDelegate {
     var onActiveSpeakersChanged: @convention(c) (UnsafeRawPointer, CFArray) -> Void
     var onDidSubscribeToRemoteVideoTrack: @convention(c) (UnsafeRawPointer, CFString, CFString, UnsafeRawPointer) -> Void
     var onDidUnsubscribeFromRemoteVideoTrack: @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void
+    var onDidPublishOrUnpublishLocalAudioTrack: @convention(c) (UnsafeRawPointer, UnsafeRawPointer, Bool) -> Void
+    var onDidPublishOrUnpublishLocalVideoTrack: @convention(c) (UnsafeRawPointer, UnsafeRawPointer, Bool) -> Void
 
     init(
         data: UnsafeRawPointer,
@@ -21,7 +23,10 @@ class LKRoomDelegate: RoomDelegate {
         onMuteChangedFromRemoteAudioTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, Bool) -> Void,
         onActiveSpeakersChanged: @convention(c) (UnsafeRawPointer, CFArray) -> Void,
         onDidSubscribeToRemoteVideoTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString, UnsafeRawPointer) -> Void,
-        onDidUnsubscribeFromRemoteVideoTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void)
+        onDidUnsubscribeFromRemoteVideoTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void,
+        onDidPublishOrUnpublishLocalAudioTrack: @escaping @convention(c) (UnsafeRawPointer, UnsafeRawPointer, Bool) -> Void,
+        onDidPublishOrUnpublishLocalVideoTrack: @escaping @convention(c) (UnsafeRawPointer, UnsafeRawPointer, Bool) -> Void
+    )
     {
         self.data = data
         self.onDidDisconnect = onDidDisconnect
@@ -31,6 +36,8 @@ class LKRoomDelegate: RoomDelegate {
         self.onDidUnsubscribeFromRemoteVideoTrack = onDidUnsubscribeFromRemoteVideoTrack
         self.onMuteChangedFromRemoteAudioTrack = onMuteChangedFromRemoteAudioTrack
         self.onActiveSpeakersChanged = onActiveSpeakersChanged
+        self.onDidPublishOrUnpublishLocalAudioTrack = onDidPublishOrUnpublishLocalAudioTrack
+        self.onDidPublishOrUnpublishLocalVideoTrack = onDidPublishOrUnpublishLocalVideoTrack
     }
 
     func room(_ room: Room, didUpdate connectionState: ConnectionState, oldValue: ConnectionState) {
@@ -65,6 +72,22 @@ class LKRoomDelegate: RoomDelegate {
             self.onDidUnsubscribeFromRemoteAudioTrack(self.data, participant.identity as CFString, track.sid! as CFString)
         }
     }
+
+    func room(_ room: Room, localParticipant: LocalParticipant, didPublish publication: LocalTrackPublication) {
+        if publication.kind == .video {
+            self.onDidPublishOrUnpublishLocalVideoTrack(self.data, Unmanaged.passUnretained(publication).toOpaque(), true)
+        } else if publication.kind == .audio {
+            self.onDidPublishOrUnpublishLocalAudioTrack(self.data, Unmanaged.passUnretained(publication).toOpaque(), true)
+        }
+    }
+
+    func room(_ room: Room, localParticipant: LocalParticipant, didUnpublish publication: LocalTrackPublication) {
+        if publication.kind == .video {
+            self.onDidPublishOrUnpublishLocalVideoTrack(self.data, Unmanaged.passUnretained(publication).toOpaque(), false)
+        } else if publication.kind == .audio {
+            self.onDidPublishOrUnpublishLocalAudioTrack(self.data, Unmanaged.passUnretained(publication).toOpaque(), false)
+        }
+    }
 }
 
 class LKVideoRenderer: NSObject, VideoRenderer {
@@ -109,7 +132,9 @@ public func LKRoomDelegateCreate(
     onMuteChangedFromRemoteAudioTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, Bool) -> Void,
     onActiveSpeakerChanged: @escaping @convention(c) (UnsafeRawPointer, CFArray) -> Void,
     onDidSubscribeToRemoteVideoTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString, UnsafeRawPointer) -> Void,
-    onDidUnsubscribeFromRemoteVideoTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void
+    onDidUnsubscribeFromRemoteVideoTrack: @escaping @convention(c) (UnsafeRawPointer, CFString, CFString) -> Void,
+    onDidPublishOrUnpublishLocalAudioTrack: @escaping @convention(c) (UnsafeRawPointer, UnsafeRawPointer, Bool) -> Void,
+    onDidPublishOrUnpublishLocalVideoTrack: @escaping @convention(c) (UnsafeRawPointer, UnsafeRawPointer, Bool) -> Void
 ) -> UnsafeMutableRawPointer {
     let delegate = LKRoomDelegate(
         data: data,
@@ -119,7 +144,9 @@ public func LKRoomDelegateCreate(
         onMuteChangedFromRemoteAudioTrack: onMuteChangedFromRemoteAudioTrack,
         onActiveSpeakersChanged: onActiveSpeakerChanged,
         onDidSubscribeToRemoteVideoTrack: onDidSubscribeToRemoteVideoTrack,
-        onDidUnsubscribeFromRemoteVideoTrack: onDidUnsubscribeFromRemoteVideoTrack
+        onDidUnsubscribeFromRemoteVideoTrack: onDidUnsubscribeFromRemoteVideoTrack,
+        onDidPublishOrUnpublishLocalAudioTrack: onDidPublishOrUnpublishLocalAudioTrack,
+        onDidPublishOrUnpublishLocalVideoTrack: onDidPublishOrUnpublishLocalVideoTrack
     )
     return Unmanaged.passRetained(delegate).toOpaque()
 }
@@ -292,6 +319,14 @@ public func LKLocalTrackPublicationSetMute(
     }
 }
 
+@_cdecl("LKLocalTrackPublicationIsMuted")
+public func LKLocalTrackPublicationIsMuted(
+    publication: UnsafeRawPointer
+) -> Bool {
+    let publication = Unmanaged<LocalTrackPublication>.fromOpaque(publication).takeUnretainedValue()
+    return publication.muted
+}
+
 @_cdecl("LKRemoteTrackPublicationSetEnabled")
 public func LKRemoteTrackPublicationSetEnabled(
     publication: UnsafeRawPointer,
@@ -325,3 +360,12 @@ public func LKRemoteTrackPublicationGetSid(
 
     return publication.sid as CFString
 }
+
+@_cdecl("LKLocalTrackPublicationGetSid")
+public func LKLocalTrackPublicationGetSid(
+    publication: UnsafeRawPointer
+) -> CFString {
+    let publication = Unmanaged<LocalTrackPublication>.fromOpaque(publication).takeUnretainedValue()
+
+    return publication.sid as CFString
+}

crates/live_kit_client/examples/test_app.rs

@@ -2,9 +2,7 @@ use std::{sync::Arc, time::Duration};
 
 use futures::StreamExt;
 use gpui::{actions, KeyBinding, Menu, MenuItem};
-use live_kit_client::{
-    LocalAudioTrack, LocalVideoTrack, RemoteAudioTrackUpdate, RemoteVideoTrackUpdate, Room,
-};
+use live_kit_client::{LocalAudioTrack, LocalVideoTrack, Room, RoomUpdate};
 use live_kit_server::token::{self, VideoGrant};
 use log::LevelFilter;
 use simplelog::SimpleLogger;
@@ -60,12 +58,12 @@ fn main() {
             let room_b = Room::new();
             room_b.connect(&live_kit_url, &user2_token).await.unwrap();
 
-            let mut audio_track_updates = room_b.remote_audio_track_updates();
+            let mut room_updates = room_b.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()
+            if let RoomUpdate::SubscribedToRemoteAudioTrack(track, _) =
+                room_updates.next().await.unwrap()
             {
                 let remote_tracks = room_b.remote_audio_tracks("test-participant-1");
                 assert_eq!(remote_tracks.len(), 1);
@@ -78,8 +76,8 @@ fn main() {
             audio_track_publication.set_mute(true).await.unwrap();
 
             println!("waiting for mute changed!");
-            if let RemoteAudioTrackUpdate::MuteChanged { track_id, muted } =
-                audio_track_updates.next().await.unwrap()
+            if let RoomUpdate::RemoteAudioTrackMuteChanged { track_id, muted } =
+                room_updates.next().await.unwrap()
             {
                 let remote_tracks = room_b.remote_audio_tracks("test-participant-1");
                 assert_eq!(remote_tracks[0].sid(), track_id);
@@ -90,8 +88,8 @@ fn main() {
 
             audio_track_publication.set_mute(false).await.unwrap();
 
-            if let RemoteAudioTrackUpdate::MuteChanged { track_id, muted } =
-                audio_track_updates.next().await.unwrap()
+            if let RoomUpdate::RemoteAudioTrackMuteChanged { track_id, muted } =
+                room_updates.next().await.unwrap()
             {
                 let remote_tracks = room_b.remote_audio_tracks("test-participant-1");
                 assert_eq!(remote_tracks[0].sid(), track_id);
@@ -110,13 +108,13 @@ fn main() {
             room_a.unpublish_track(audio_track_publication);
 
             // Clear out any active speakers changed messages
-            let mut next = audio_track_updates.next().await.unwrap();
-            while let RemoteAudioTrackUpdate::ActiveSpeakersChanged { speakers } = next {
+            let mut next = room_updates.next().await.unwrap();
+            while let RoomUpdate::ActiveSpeakersChanged { speakers } = next {
                 println!("Speakers changed: {:?}", speakers);
-                next = audio_track_updates.next().await.unwrap();
+                next = room_updates.next().await.unwrap();
             }
 
-            if let RemoteAudioTrackUpdate::Unsubscribed {
+            if let RoomUpdate::UnsubscribedFromRemoteAudioTrack {
                 publisher_id,
                 track_id,
             } = next
@@ -128,7 +126,6 @@ fn main() {
                 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();
 
@@ -136,8 +133,8 @@ fn main() {
             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()
+            if let RoomUpdate::SubscribedToRemoteVideoTrack(track) =
+                room_updates.next().await.unwrap()
             {
                 let remote_video_tracks = room_b.remote_video_tracks("test-participant-1");
                 assert_eq!(remote_video_tracks.len(), 1);
@@ -152,10 +149,10 @@ fn main() {
                 .pop()
                 .unwrap();
             room_a.unpublish_track(local_video_track_publication);
-            if let RemoteVideoTrackUpdate::Unsubscribed {
+            if let RoomUpdate::UnsubscribedFromRemoteVideoTrack {
                 publisher_id,
                 track_id,
-            } = video_track_updates.next().await.unwrap()
+            } = room_updates.next().await.unwrap()
             {
                 assert_eq!(publisher_id, "test-participant-1");
                 assert_eq!(remote_video_track.sid(), track_id);

crates/live_kit_client/src/live_kit_client.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 #[cfg(not(any(test, feature = "test-support")))]
 pub mod prod;
 
@@ -9,3 +11,25 @@ pub mod test;
 
 #[cfg(any(test, feature = "test-support"))]
 pub use test::*;
+
+pub type Sid = String;
+
+#[derive(Clone, Eq, PartialEq)]
+pub enum ConnectionState {
+    Disconnected,
+    Connected { url: String, token: String },
+}
+
+#[derive(Clone)]
+pub enum RoomUpdate {
+    ActiveSpeakersChanged { speakers: Vec<Sid> },
+    RemoteAudioTrackMuteChanged { track_id: Sid, muted: bool },
+    SubscribedToRemoteVideoTrack(Arc<RemoteVideoTrack>),
+    SubscribedToRemoteAudioTrack(Arc<RemoteAudioTrack>, Arc<RemoteTrackPublication>),
+    UnsubscribedFromRemoteVideoTrack { publisher_id: Sid, track_id: Sid },
+    UnsubscribedFromRemoteAudioTrack { publisher_id: Sid, track_id: Sid },
+    LocalAudioTrackPublished { publication: LocalTrackPublication },
+    LocalAudioTrackUnpublished { publication: LocalTrackPublication },
+    LocalVideoTrackPublished { publication: LocalTrackPublication },
+    LocalVideoTrackUnpublished { publication: LocalTrackPublication },
+}

crates/live_kit_client/src/prod.rs

@@ -1,3 +1,4 @@
+use crate::{ConnectionState, RoomUpdate, Sid};
 use anyhow::{anyhow, Context, Result};
 use core_foundation::{
     array::{CFArray, CFArrayRef},
@@ -76,6 +77,16 @@ extern "C" {
             publisher_id: CFStringRef,
             track_id: CFStringRef,
         ),
+        on_did_publish_or_unpublish_local_audio_track: extern "C" fn(
+            callback_data: *mut c_void,
+            publication: swift::LocalTrackPublication,
+            is_published: bool,
+        ),
+        on_did_publish_or_unpublish_local_video_track: extern "C" fn(
+            callback_data: *mut c_void,
+            publication: swift::LocalTrackPublication,
+            is_published: bool,
+        ),
     ) -> swift::RoomDelegate;
 
     fn LKRoomCreate(delegate: swift::RoomDelegate) -> swift::Room;
@@ -151,26 +162,19 @@ extern "C" {
         callback_data: *mut c_void,
     );
 
+    fn LKLocalTrackPublicationIsMuted(publication: swift::LocalTrackPublication) -> bool;
     fn LKRemoteTrackPublicationIsMuted(publication: swift::RemoteTrackPublication) -> bool;
+    fn LKLocalTrackPublicationGetSid(publication: swift::LocalTrackPublication) -> CFStringRef;
     fn LKRemoteTrackPublicationGetSid(publication: swift::RemoteTrackPublication) -> CFStringRef;
 }
 
-pub type Sid = String;
-
-#[derive(Clone, Eq, PartialEq)]
-pub enum ConnectionState {
-    Disconnected,
-    Connected { url: String, token: String },
-}
-
 pub struct Room {
     native_room: swift::Room,
     connection: Mutex<(
         watch::Sender<ConnectionState>,
         watch::Receiver<ConnectionState>,
     )>,
-    remote_audio_track_subscribers: Mutex<Vec<mpsc::UnboundedSender<RemoteAudioTrackUpdate>>>,
-    remote_video_track_subscribers: Mutex<Vec<mpsc::UnboundedSender<RemoteVideoTrackUpdate>>>,
+    update_subscribers: Mutex<Vec<mpsc::UnboundedSender<RoomUpdate>>>,
     _delegate: RoomDelegate,
 }
 
@@ -181,8 +185,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(),
+                update_subscribers: Default::default(),
                 _delegate: delegate,
             }
         })
@@ -397,15 +400,9 @@ impl Room {
         }
     }
 
-    pub fn remote_audio_track_updates(&self) -> mpsc::UnboundedReceiver<RemoteAudioTrackUpdate> {
+    pub fn updates(&self) -> mpsc::UnboundedReceiver<RoomUpdate> {
         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);
+        self.update_subscribers.lock().push(tx);
         rx
     }
 
@@ -416,8 +413,8 @@ impl Room {
     ) {
         let track = Arc::new(track);
         let publication = Arc::new(publication);
-        self.remote_audio_track_subscribers.lock().retain(|tx| {
-            tx.unbounded_send(RemoteAudioTrackUpdate::Subscribed(
+        self.update_subscribers.lock().retain(|tx| {
+            tx.unbounded_send(RoomUpdate::SubscribedToRemoteAudioTrack(
                 track.clone(),
                 publication.clone(),
             ))
@@ -426,8 +423,8 @@ impl Room {
     }
 
     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 {
+        self.update_subscribers.lock().retain(|tx| {
+            tx.unbounded_send(RoomUpdate::UnsubscribedFromRemoteAudioTrack {
                 publisher_id: publisher_id.clone(),
                 track_id: track_id.clone(),
             })
@@ -436,8 +433,8 @@ impl Room {
     }
 
     fn mute_changed_from_remote_audio_track(&self, track_id: String, muted: bool) {
-        self.remote_audio_track_subscribers.lock().retain(|tx| {
-            tx.unbounded_send(RemoteAudioTrackUpdate::MuteChanged {
+        self.update_subscribers.lock().retain(|tx| {
+            tx.unbounded_send(RoomUpdate::RemoteAudioTrackMuteChanged {
                 track_id: track_id.clone(),
                 muted,
             })
@@ -445,12 +442,9 @@ impl Room {
         });
     }
 
-    // A vec of publisher IDs
     fn active_speakers_changed(&self, speakers: Vec<String>) {
-        self.remote_audio_track_subscribers
-            .lock()
-            .retain(move |tx| {
-                tx.unbounded_send(RemoteAudioTrackUpdate::ActiveSpeakersChanged {
+        self.update_subscribers.lock().retain(move |tx| {
+            tx.unbounded_send(RoomUpdate::ActiveSpeakersChanged {
                 speakers: speakers.clone(),
             })
             .is_ok()
@@ -459,15 +453,15 @@ impl Room {
 
     fn did_subscribe_to_remote_video_track(&self, track: RemoteVideoTrack) {
         let track = Arc::new(track);
-        self.remote_video_track_subscribers.lock().retain(|tx| {
-            tx.unbounded_send(RemoteVideoTrackUpdate::Subscribed(track.clone()))
+        self.update_subscribers.lock().retain(|tx| {
+            tx.unbounded_send(RoomUpdate::SubscribedToRemoteVideoTrack(track.clone()))
                 .is_ok()
         });
     }
 
     fn did_unsubscribe_from_remote_video_track(&self, publisher_id: String, track_id: String) {
-        self.remote_video_track_subscribers.lock().retain(|tx| {
-            tx.unbounded_send(RemoteVideoTrackUpdate::Unsubscribed {
+        self.update_subscribers.lock().retain(|tx| {
+            tx.unbounded_send(RoomUpdate::UnsubscribedFromRemoteVideoTrack {
                 publisher_id: publisher_id.clone(),
                 track_id: track_id.clone(),
             })
@@ -529,6 +523,8 @@ impl RoomDelegate {
                 Self::on_active_speakers_changed,
                 Self::on_did_subscribe_to_remote_video_track,
                 Self::on_did_unsubscribe_from_remote_video_track,
+                Self::on_did_publish_or_unpublish_local_audio_track,
+                Self::on_did_publish_or_unpublish_local_video_track,
             )
         };
         Self {
@@ -642,6 +638,46 @@ impl RoomDelegate {
         }
         let _ = Weak::into_raw(room);
     }
+
+    extern "C" fn on_did_publish_or_unpublish_local_audio_track(
+        room: *mut c_void,
+        publication: swift::LocalTrackPublication,
+        is_published: bool,
+    ) {
+        let room = unsafe { Weak::from_raw(room as *mut Room) };
+        if let Some(room) = room.upgrade() {
+            let publication = LocalTrackPublication::new(publication);
+            let update = if is_published {
+                RoomUpdate::LocalAudioTrackPublished { publication }
+            } else {
+                RoomUpdate::LocalAudioTrackUnpublished { publication }
+            };
+            room.update_subscribers
+                .lock()
+                .retain(|tx| tx.unbounded_send(update.clone()).is_ok());
+        }
+        let _ = Weak::into_raw(room);
+    }
+
+    extern "C" fn on_did_publish_or_unpublish_local_video_track(
+        room: *mut c_void,
+        publication: swift::LocalTrackPublication,
+        is_published: bool,
+    ) {
+        let room = unsafe { Weak::from_raw(room as *mut Room) };
+        if let Some(room) = room.upgrade() {
+            let publication = LocalTrackPublication::new(publication);
+            let update = if is_published {
+                RoomUpdate::LocalVideoTrackPublished { publication }
+            } else {
+                RoomUpdate::LocalVideoTrackUnpublished { publication }
+            };
+            room.update_subscribers
+                .lock()
+                .retain(|tx| tx.unbounded_send(update.clone()).is_ok());
+        }
+        let _ = Weak::into_raw(room);
+    }
 }
 
 impl Drop for RoomDelegate {
@@ -691,6 +727,10 @@ impl LocalTrackPublication {
         Self(native_track_publication)
     }
 
+    pub fn sid(&self) -> String {
+        unsafe { CFString::wrap_under_get_rule(LKLocalTrackPublicationGetSid(self.0)).to_string() }
+    }
+
     pub fn set_mute(&self, muted: bool) -> impl Future<Output = Result<()>> {
         let (tx, rx) = futures::channel::oneshot::channel();
 
@@ -715,6 +755,19 @@ impl LocalTrackPublication {
 
         async move { rx.await.unwrap() }
     }
+
+    pub fn is_muted(&self) -> bool {
+        unsafe { LKLocalTrackPublicationIsMuted(self.0) }
+    }
+}
+
+impl Clone for LocalTrackPublication {
+    fn clone(&self) -> Self {
+        unsafe {
+            CFRetain(self.0 .0);
+        }
+        Self(self.0)
+    }
 }
 
 impl Drop for LocalTrackPublication {
@@ -889,18 +942,6 @@ impl Drop for RemoteVideoTrack {
     }
 }
 
-pub enum RemoteVideoTrackUpdate {
-    Subscribed(Arc<RemoteVideoTrack>),
-    Unsubscribed { publisher_id: Sid, track_id: Sid },
-}
-
-pub enum RemoteAudioTrackUpdate {
-    ActiveSpeakersChanged { speakers: Vec<Sid> },
-    MuteChanged { track_id: Sid, muted: bool },
-    Subscribed(Arc<RemoteAudioTrack>, Arc<RemoteTrackPublication>),
-    Unsubscribed { publisher_id: Sid, track_id: Sid },
-}
-
 pub struct MacOSDisplay(swift::MacOSDisplay);
 
 impl MacOSDisplay {

crates/live_kit_client/src/test.rs

@@ -1,3 +1,4 @@
+use crate::{ConnectionState, RoomUpdate, Sid};
 use anyhow::{anyhow, Context, Result};
 use async_trait::async_trait;
 use collections::{BTreeMap, HashMap};
@@ -7,7 +8,14 @@ use live_kit_server::{proto, token};
 use media::core_video::CVImageBuffer;
 use parking_lot::Mutex;
 use postage::watch;
-use std::{future::Future, mem, sync::Arc};
+use std::{
+    future::Future,
+    mem,
+    sync::{
+        atomic::{AtomicBool, Ordering::SeqCst},
+        Arc,
+    },
+};
 
 static SERVERS: Mutex<BTreeMap<String, Arc<TestServer>>> = Mutex::new(BTreeMap::new());
 
@@ -104,9 +112,8 @@ impl TestServer {
                 client_room
                     .0
                     .lock()
-                    .video_track_updates
-                    .0
-                    .try_broadcast(RemoteVideoTrackUpdate::Subscribed(track.clone()))
+                    .updates_tx
+                    .try_broadcast(RoomUpdate::SubscribedToRemoteVideoTrack(track.clone()))
                     .unwrap();
             }
             room.client_rooms.insert(identity, client_room);
@@ -176,7 +183,11 @@ impl TestServer {
         }
     }
 
-    async fn publish_video_track(&self, token: String, local_track: LocalVideoTrack) -> Result<()> {
+    async fn publish_video_track(
+        &self,
+        token: String,
+        local_track: LocalVideoTrack,
+    ) -> Result<Sid> {
         self.executor.simulate_random_delay().await;
         let claims = live_kit_server::token::validate(&token, &self.secret_key)?;
         let identity = claims.sub.unwrap().to_string();
@@ -198,8 +209,9 @@ impl TestServer {
             return Err(anyhow!("user is not allowed to publish"));
         }
 
+        let sid = nanoid::nanoid!(17);
         let track = Arc::new(RemoteVideoTrack {
-            sid: nanoid::nanoid!(17),
+            sid: sid.clone(),
             publisher_id: identity.clone(),
             frames_rx: local_track.frames_rx.clone(),
         });
@@ -211,21 +223,20 @@ impl TestServer {
                 let _ = client_room
                     .0
                     .lock()
-                    .video_track_updates
-                    .0
-                    .try_broadcast(RemoteVideoTrackUpdate::Subscribed(track.clone()))
+                    .updates_tx
+                    .try_broadcast(RoomUpdate::SubscribedToRemoteVideoTrack(track.clone()))
                     .unwrap();
             }
         }
 
-        Ok(())
+        Ok(sid)
     }
 
     async fn publish_audio_track(
         &self,
         token: String,
         _local_track: &LocalAudioTrack,
-    ) -> Result<()> {
+    ) -> Result<Sid> {
         self.executor.simulate_random_delay().await;
         let claims = live_kit_server::token::validate(&token, &self.secret_key)?;
         let identity = claims.sub.unwrap().to_string();
@@ -247,8 +258,9 @@ impl TestServer {
             return Err(anyhow!("user is not allowed to publish"));
         }
 
+        let sid = nanoid::nanoid!(17);
         let track = Arc::new(RemoteAudioTrack {
-            sid: nanoid::nanoid!(17),
+            sid: sid.clone(),
             publisher_id: identity.clone(),
         });
 
@@ -261,9 +273,8 @@ impl TestServer {
                 let _ = client_room
                     .0
                     .lock()
-                    .audio_track_updates
-                    .0
-                    .try_broadcast(RemoteAudioTrackUpdate::Subscribed(
+                    .updates_tx
+                    .try_broadcast(RoomUpdate::SubscribedToRemoteAudioTrack(
                         track.clone(),
                         publication.clone(),
                     ))
@@ -271,7 +282,7 @@ impl TestServer {
             }
         }
 
-        Ok(())
+        Ok(sid)
     }
 
     fn video_tracks(&self, token: String) -> Result<Vec<Arc<RemoteVideoTrack>>> {
@@ -369,39 +380,26 @@ impl live_kit_server::api::Client for TestApiClient {
     }
 }
 
-pub type Sid = String;
-
 struct RoomState {
     connection: (
         watch::Sender<ConnectionState>,
         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>,
-    ),
-}
-
-#[derive(Clone, Eq, PartialEq)]
-pub enum ConnectionState {
-    Disconnected,
-    Connected { url: String, token: String },
+    updates_tx: async_broadcast::Sender<RoomUpdate>,
+    updates_rx: async_broadcast::Receiver<RoomUpdate>,
 }
 
 pub struct Room(Mutex<RoomState>);
 
 impl Room {
     pub fn new() -> Arc<Self> {
+        let (updates_tx, updates_rx) = async_broadcast::broadcast(128);
         Arc::new(Self(Mutex::new(RoomState {
             connection: watch::channel_with(ConnectionState::Disconnected),
             display_sources: Default::default(),
-            video_track_updates: async_broadcast::broadcast(128),
-            audio_track_updates: async_broadcast::broadcast(128),
+            updates_tx,
+            updates_rx,
         })))
     }
 
@@ -440,10 +438,14 @@ impl Room {
         let this = self.clone();
         let track = track.clone();
         async move {
-            this.test_server()
+            let sid = this
+                .test_server()
                 .publish_video_track(this.token(), track)
                 .await?;
-            Ok(LocalTrackPublication)
+            Ok(LocalTrackPublication {
+                muted: Default::default(),
+                sid,
+            })
         }
     }
     pub fn publish_audio_track(
@@ -453,10 +455,14 @@ impl Room {
         let this = self.clone();
         let track = track.clone();
         async move {
-            this.test_server()
+            let sid = this
+                .test_server()
                 .publish_audio_track(this.token(), &track)
                 .await?;
-            Ok(LocalTrackPublication)
+            Ok(LocalTrackPublication {
+                muted: Default::default(),
+                sid,
+            })
         }
     }
 
@@ -505,12 +511,8 @@ 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()
+    pub fn updates(&self) -> impl Stream<Item = RoomUpdate> {
+        self.0.lock().updates_rx.clone()
     }
 
     pub fn set_display_sources(&self, sources: Vec<MacOSDisplay>) {
@@ -555,11 +557,27 @@ impl Drop for Room {
     }
 }
 
-pub struct LocalTrackPublication;
+#[derive(Clone)]
+pub struct LocalTrackPublication {
+    sid: String,
+    muted: Arc<AtomicBool>,
+}
 
 impl LocalTrackPublication {
-    pub fn set_mute(&self, _mute: bool) -> impl Future<Output = Result<()>> {
-        async { Ok(()) }
+    pub fn set_mute(&self, mute: bool) -> impl Future<Output = Result<()>> {
+        let muted = self.muted.clone();
+        async move {
+            muted.store(mute, SeqCst);
+            Ok(())
+        }
+    }
+
+    pub fn is_muted(&self) -> bool {
+        self.muted.load(SeqCst)
+    }
+
+    pub fn sid(&self) -> String {
+        self.sid.clone()
     }
 }
 
@@ -646,20 +664,6 @@ impl RemoteAudioTrack {
     }
 }
 
-#[derive(Clone)]
-pub enum RemoteVideoTrackUpdate {
-    Subscribed(Arc<RemoteVideoTrack>),
-    Unsubscribed { publisher_id: Sid, track_id: Sid },
-}
-
-#[derive(Clone)]
-pub enum RemoteAudioTrackUpdate {
-    ActiveSpeakersChanged { speakers: Vec<Sid> },
-    MuteChanged { track_id: Sid, muted: bool },
-    Subscribed(Arc<RemoteAudioTrack>, Arc<RemoteTrackPublication>),
-    Unsubscribed { publisher_id: Sid, track_id: Sid },
-}
-
 #[derive(Clone)]
 pub struct MacOSDisplay {
     frames: (

script/zed-local

@@ -4,20 +4,28 @@ const { spawn, execFileSync } = require("child_process");
 
 const RESOLUTION_REGEX = /(\d+) x (\d+)/;
 const DIGIT_FLAG_REGEX = /^--?(\d+)$/;
-const RELEASE_MODE = "--release";
-
-const args = process.argv.slice(2);
 
 // Parse the number of Zed instances to spawn.
 let instanceCount = 1;
-const digitMatch = args[0]?.match(DIGIT_FLAG_REGEX);
-if (digitMatch) {
+let isReleaseMode = false;
+let isTop = false;
+
+const args = process.argv.slice(2);
+for (const arg of args) {
+  const digitMatch = arg.match(DIGIT_FLAG_REGEX);
+  if (digitMatch) {
     instanceCount = parseInt(digitMatch[1]);
-  args.shift();
-}
-const isReleaseMode = args.some((arg) => arg === RELEASE_MODE);
-if (instanceCount > 4) {
-  throw new Error("Cannot spawn more than 4 instances");
+    continue;
+  }
+
+  if (arg == "--release") {
+    isReleaseMode = true;
+    continue;
+  }
+
+  if (arg == "--top") {
+    isTop = true;
+  }
 }
 
 // Parse the resolution of the main screen
@@ -34,7 +42,11 @@ if (!mainDisplayResolution) {
   throw new Error("Could not parse screen resolution");
 }
 const screenWidth = parseInt(mainDisplayResolution[1]);
-const screenHeight = parseInt(mainDisplayResolution[2]);
+let screenHeight = parseInt(mainDisplayResolution[2]);
+
+if (isTop) {
+  screenHeight = Math.floor(screenHeight / 2);
+}
 
 // Determine the window size for each instance
 let instanceWidth = screenWidth;