Detect when a track is unpublished due to reconnecting to livekit

Co-authored-by: Julia <julia@zed.dev>

crates/live_kit_client/src/live_kit_client.rs

@@ -28,4 +28,8 @@ pub enum RoomUpdate {
     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

@@ -77,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;
@@ -152,7 +162,9 @@ 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;
 }
 
@@ -511,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 {
@@ -624,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 {
@@ -673,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();
 
@@ -697,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 {

crates/live_kit_client/src/test.rs

@@ -8,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());
 
@@ -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(),
         });
@@ -217,14 +229,14 @@ impl TestServer {
             }
         }
 
-        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();
@@ -246,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(),
         });
 
@@ -269,7 +282,7 @@ impl TestServer {
             }
         }
 
-        Ok(())
+        Ok(sid)
     }
 
     fn video_tracks(&self, token: String) -> Result<Vec<Arc<RemoteVideoTrack>>> {
@@ -425,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(
@@ -438,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,
+            })
         }
     }
 
@@ -536,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()
     }
 }