Add support for TrackSubscribed

exported as LocalTrackSubscribed in the client and FFI.

livekit-ffi/protocol/room.proto

@@ -318,21 +318,22 @@ message RoomEvent {
     ParticipantDisconnected participant_disconnected = 3;
     LocalTrackPublished local_track_published = 4;
     LocalTrackUnpublished local_track_unpublished = 5;
-    TrackPublished track_published = 6;
-    TrackUnpublished track_unpublished = 7;
-    TrackSubscribed track_subscribed = 8;
-    TrackUnsubscribed track_unsubscribed = 9;
-    TrackSubscriptionFailed track_subscription_failed = 10;
-    TrackMuted track_muted = 11;
-    TrackUnmuted track_unmuted = 12;
-    ActiveSpeakersChanged active_speakers_changed = 13;
-    RoomMetadataChanged room_metadata_changed = 14;
-    RoomSidChanged room_sid_changed = 15;
-    ParticipantMetadataChanged participant_metadata_changed = 16;
-    ParticipantNameChanged participant_name_changed = 17;
-    ParticipantAttributesChanged participant_attributes_changed = 18;
-    ConnectionQualityChanged connection_quality_changed = 19;
-    ConnectionStateChanged connection_state_changed = 20;
+    LocalTrackSubscribed local_track_subscribed = 6;
+    TrackPublished track_published = 7;
+    TrackUnpublished track_unpublished = 8;
+    TrackSubscribed track_subscribed = 9;
+    TrackUnsubscribed track_unsubscribed = 10;
+    TrackSubscriptionFailed track_subscription_failed = 11;
+    TrackMuted track_muted = 12;
+    TrackUnmuted track_unmuted = 13;
+    ActiveSpeakersChanged active_speakers_changed = 14;
+    RoomMetadataChanged room_metadata_changed = 15;
+    RoomSidChanged room_sid_changed = 16;
+    ParticipantMetadataChanged participant_metadata_changed = 17;
+    ParticipantNameChanged participant_name_changed = 18;
+    ParticipantAttributesChanged participant_attributes_changed = 19;
+    ConnectionQualityChanged connection_quality_changed = 20;
+    ConnectionStateChanged connection_state_changed = 21;
     // Connected connected = 21;
     Disconnected disconnected = 22;
     Reconnecting reconnecting = 23;
@@ -371,6 +372,10 @@ message LocalTrackUnpublished {
   string publication_sid = 1;
 }
 
+message LocalTrackSubscribed {
+  string track_sid = 2;
+}
+
 message TrackPublished {
   string participant_identity = 1;
   OwnedTrackPublication publication = 2;

livekit-ffi/src/livekit.proto.rs

@@ -2316,7 +2316,7 @@ pub struct OwnedBuffer {
 pub struct RoomEvent {
     #[prost(uint64, tag="1")]
     pub room_handle: u64,
-    #[prost(oneof="room_event::Message", tags="2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 22, 23, 24, 25, 26, 27, 28")]
+    #[prost(oneof="room_event::Message", tags="2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28")]
     pub message: ::core::option::Option<room_event::Message>,
 }
 /// Nested message and enum types in `RoomEvent`.
@@ -2333,34 +2333,36 @@ pub mod room_event {
         #[prost(message, tag="5")]
         LocalTrackUnpublished(super::LocalTrackUnpublished),
         #[prost(message, tag="6")]
-        TrackPublished(super::TrackPublished),
+        LocalTrackSubscribed(super::LocalTrackSubscribed),
         #[prost(message, tag="7")]
-        TrackUnpublished(super::TrackUnpublished),
+        TrackPublished(super::TrackPublished),
         #[prost(message, tag="8")]
-        TrackSubscribed(super::TrackSubscribed),
+        TrackUnpublished(super::TrackUnpublished),
         #[prost(message, tag="9")]
-        TrackUnsubscribed(super::TrackUnsubscribed),
+        TrackSubscribed(super::TrackSubscribed),
         #[prost(message, tag="10")]
-        TrackSubscriptionFailed(super::TrackSubscriptionFailed),
+        TrackUnsubscribed(super::TrackUnsubscribed),
         #[prost(message, tag="11")]
-        TrackMuted(super::TrackMuted),
+        TrackSubscriptionFailed(super::TrackSubscriptionFailed),
         #[prost(message, tag="12")]
-        TrackUnmuted(super::TrackUnmuted),
+        TrackMuted(super::TrackMuted),
         #[prost(message, tag="13")]
-        ActiveSpeakersChanged(super::ActiveSpeakersChanged),
+        TrackUnmuted(super::TrackUnmuted),
         #[prost(message, tag="14")]
-        RoomMetadataChanged(super::RoomMetadataChanged),
+        ActiveSpeakersChanged(super::ActiveSpeakersChanged),
         #[prost(message, tag="15")]
-        RoomSidChanged(super::RoomSidChanged),
+        RoomMetadataChanged(super::RoomMetadataChanged),
         #[prost(message, tag="16")]
-        ParticipantMetadataChanged(super::ParticipantMetadataChanged),
+        RoomSidChanged(super::RoomSidChanged),
         #[prost(message, tag="17")]
-        ParticipantNameChanged(super::ParticipantNameChanged),
+        ParticipantMetadataChanged(super::ParticipantMetadataChanged),
         #[prost(message, tag="18")]
-        ParticipantAttributesChanged(super::ParticipantAttributesChanged),
+        ParticipantNameChanged(super::ParticipantNameChanged),
         #[prost(message, tag="19")]
-        ConnectionQualityChanged(super::ConnectionQualityChanged),
+        ParticipantAttributesChanged(super::ParticipantAttributesChanged),
         #[prost(message, tag="20")]
+        ConnectionQualityChanged(super::ConnectionQualityChanged),
+        #[prost(message, tag="21")]
         ConnectionStateChanged(super::ConnectionStateChanged),
         /// Connected connected = 21;
         #[prost(message, tag="22")]
@@ -2426,6 +2428,12 @@ pub struct LocalTrackUnpublished {
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
+pub struct LocalTrackSubscribed {
+    #[prost(string, tag="2")]
+    pub track_sid: ::prost::alloc::string::String,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct TrackPublished {
     #[prost(string, tag="1")]
     pub participant_identity: ::prost::alloc::string::String,

livekit-ffi/src/server/room.rs

@@ -773,6 +773,11 @@ async fn forward_event(
 
             inner.pending_unpublished_tracks.lock().insert(publication.sid());
         }
+        RoomEvent::LocalTrackSubscribed { track } => {
+            let _ = send_event(proto::room_event::Message::LocalTrackSubscribed(
+                proto::LocalTrackSubscribed { track_sid: track.sid().to_string() },
+            ));
+        }
         RoomEvent::TrackPublished { publication, participant } => {
             let handle_id = server.next_id();
             let ffi_publication = FfiPublication {

livekit/src/room/mod.rs

@@ -85,6 +85,9 @@ pub enum RoomEvent {
         publication: LocalTrackPublication,
         participant: LocalParticipant,
     },
+    LocalTrackSubscribed {
+        track: LocalTrack,
+    },
     TrackSubscribed {
         track: RemoteTrack,
         publication: RemoteTrackPublication,
@@ -611,6 +614,7 @@ impl RoomSession {
             EngineEvent::ConnectionQuality { updates } => {
                 self.handle_connection_quality_update(updates)
             }
+            EngineEvent::TrackSubscribed { track_sid } => self.handle_track_subscribed(track_sid),
         }
 
         Ok(())
@@ -789,6 +793,17 @@ impl RoomSession {
         }
     }
 
+    /// Handle the first time a participant subscribes to a track
+    /// Pass this event forward
+    fn handle_track_subscribed(&self, track_sid: String) {
+        let publications = self.local_participant.track_publications().clone();
+        let publication = publications.get(&track_sid.to_owned().try_into().unwrap());
+        if let Some(publication) = publication {
+            self.dispatcher
+                .dispatch(&RoomEvent::LocalTrackSubscribed { track: publication.track().unwrap() });
+        }
+    }
+
     async fn send_sync_state(self: &Arc<Self>) {
         let auto_subscribe = self.options.auto_subscribe;
         let session = self.rtc_engine.session();

livekit/src/rtc_engine/mod.rs

@@ -136,6 +136,9 @@ pub enum EngineEvent {
     Disconnected {
         reason: DisconnectReason,
     },
+    TrackSubscribed {
+        track_sid: String,
+    },
 }
 
 /// Represents a running RtcSession with the ability to close the session
@@ -443,6 +446,9 @@ impl EngineInner {
             SessionEvent::RoomUpdate { room } => {
                 let _ = self.engine_tx.send(EngineEvent::RoomUpdate { room });
             }
+            SessionEvent::TrackSubscribed { track_sid } => {
+                let _ = self.engine_tx.send(EngineEvent::TrackSubscribed { track_sid });
+            }
         }
         Ok(())
     }

livekit/src/rtc_engine/rtc_session.rs

@@ -106,6 +106,9 @@ pub enum SessionEvent {
     RoomUpdate {
         room: proto::Room,
     },
+    TrackSubscribed {
+        track_sid: String,
+    },
     Close {
         source: String,
         reason: DisconnectReason,
@@ -499,6 +502,11 @@ impl SessionInner {
                 let _ =
                     self.emitter.send(SessionEvent::RoomUpdate { room: room_update.room.unwrap() });
             }
+            proto::signal_response::Message::TrackSubscribed(track_subscribed) => {
+                let _ = self
+                    .emitter
+                    .send(SessionEvent::TrackSubscribed { track_sid: track_subscribed.track_sid });
+            }
             _ => {}
         }