add adaptive streaming;

Author: zesun96

livekit-ffi/generate_proto.sh

@@ -24,6 +24,7 @@ protoc \
     $PROTOCOL/handle.proto \
     $PROTOCOL/room.proto \
     $PROTOCOL/track.proto \
+    $PROTOCOL/track_publication.proto \
     $PROTOCOL/participant.proto \
     $PROTOCOL/video_frame.proto \
     $PROTOCOL/audio_frame.proto \

livekit-ffi/protocol/ffi.proto

@@ -20,6 +20,7 @@ option csharp_namespace = "LiveKit.Proto";
 // import "handle.proto";
 import "e2ee.proto";
 import "track.proto";
+import "track_publication.proto";
 import "room.proto";
 import "video_frame.proto";
 import "audio_frame.proto";
@@ -106,6 +107,10 @@ message FfiRequest {
     RegisterRpcMethodRequest register_rpc_method = 39;
     UnregisterRpcMethodRequest unregister_rpc_method = 40;
     RpcMethodInvocationResponseRequest rpc_method_invocation_response = 41;
+
+    // Track Publication
+    EnableRemoteTrackPublicationRequest enable_remote_track_publication = 42;
+    UpdateRemoteTrackPublicationDimensionRequest update_remote_track_publication_dimension = 43;
   }
 }
 
@@ -160,6 +165,9 @@ message FfiResponse {
     RegisterRpcMethodResponse register_rpc_method = 38;
     UnregisterRpcMethodResponse unregister_rpc_method = 39;
     RpcMethodInvocationResponseResponse rpc_method_invocation_response = 40;
+    // Track Publication
+    EnableRemoteTrackPublicationResponse enable_remote_track_publication = 41;
+    UpdateRemoteTrackPublicationDimensionResponse update_remote_track_publication_dimension = 42;
   }
 }
 

livekit-ffi/protocol/track_publication.proto

@@ -0,0 +1,38 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto2";
+
+package livekit.proto;
+option csharp_namespace = "LiveKit.Proto";
+
+// Enable/Disable a remote track publication
+message EnableRemoteTrackPublicationRequest {
+  required uint64 track_publication_handle = 1;
+  required bool enabled = 2;
+}
+
+message EnableRemoteTrackPublicationResponse {}
+
+// update a remote track publication dimension
+message UpdateRemoteTrackPublicationDimensionRequest {
+  required uint64 track_publication_handle = 1;
+  required uint32 width = 2;
+  required uint32 height = 3;
+}
+
+message UpdateRemoteTrackPublicationDimensionResponse {}
+
+
+

livekit-ffi/src/server/requests.rs

@@ -152,6 +152,36 @@ fn on_set_subscribed(
     Ok(proto::SetSubscribedResponse {})
 }
 
+fn on_enable_remote_track_publication(
+    server: &'static FfiServer,
+    request: proto::EnableRemoteTrackPublicationRequest,
+) -> FfiResult<proto::EnableRemoteTrackPublicationResponse> {
+    let ffi_publication =
+        server.retrieve_handle::<FfiPublication>(request.track_publication_handle)?;
+
+    let TrackPublication::Remote(publication) = &ffi_publication.publication else {
+        return Err(FfiError::InvalidRequest("publication is not a RemotePublication".into()));
+    };
+    
+    publication.set_enabled(request.enabled);
+    Ok(proto::EnableRemoteTrackPublicationResponse {})
+}
+
+fn on_update_remote_track_publication_dimension(
+    server: &'static FfiServer,
+    request: proto::UpdateRemoteTrackPublicationDimensionRequest,
+) -> FfiResult<proto::UpdateRemoteTrackPublicationDimensionResponse> {
+    let ffi_publication =
+        server.retrieve_handle::<FfiPublication>(request.track_publication_handle)?;
+
+    let TrackPublication::Remote(publication) = &ffi_publication.publication else {
+        return Err(FfiError::InvalidRequest("publication is not a RemotePublication".into()));
+    };
+    let dimension = TrackDimension(request.width, request.height);
+    publication.update_video_dimensions(dimension);
+    Ok(proto::UpdateRemoteTrackPublicationDimensionResponse {})
+}
+
 fn on_set_local_metadata(
     server: &'static FfiServer,
     set_local_metadata: proto::SetLocalMetadataRequest,
@@ -995,6 +1025,12 @@ pub fn handle_request(
                 on_rpc_method_invocation_response(server, request)?,
             )
         }
+        proto::ffi_request::Message::EnableRemoteTrackPublication(request) => {
+            proto::ffi_response::Message::EnableRemoteTrackPublication(on_enable_remote_track_publication(server, request)?)
+        }
+        proto::ffi_request::Message::UpdateRemoteTrackPublicationDimension(request) => {
+            proto::ffi_response::Message::UpdateRemoteTrackPublicationDimension(on_update_remote_track_publication_dimension(server, request)?)
+        }
     });
 
     Ok(res)

livekit/src/room/participant/remote_participant.rs

@@ -384,7 +384,32 @@ impl RemoteParticipant {
                         .await
                 });
             }
-        })
+        });
+
+        publication.on_video_dimensions_changed({
+            let rtc_engine = self.inner.rtc_engine.clone();
+            move |publication, dimension| {
+                let rtc_engine = rtc_engine.clone();
+                livekit_runtime::spawn(async move {
+                    let tsid: String = publication.sid().into();
+                    let TrackDimension(width, height) = dimension;
+                    let enabled = publication.is_enabled();
+                    let update_track_settings = proto::UpdateTrackSettings {
+                        track_sids: vec![tsid.clone()],
+                        disabled: !enabled,
+                        width,
+                        height,
+                        ..Default::default()
+                    };
+
+                    rtc_engine
+                        .send_request(proto::signal_request::Message::TrackSetting(
+                            update_track_settings,
+                        ))
+                        .await
+                });
+            }
+        });
     }
 
     pub(crate) fn remove_publication(&self, sid: &TrackSid) -> Option<TrackPublication> {

livekit/src/room/publication/remote.rs

@@ -28,6 +28,7 @@ type PermissionStatusChangedHandler =
     Box<dyn Fn(RemoteTrackPublication, PermissionStatus, PermissionStatus) + Send>; // old_status, new_status
 type SubscriptionUpdateNeededHandler = Box<dyn Fn(RemoteTrackPublication, bool) + Send>;
 type EnabledStatusChangedHandler = Box<dyn Fn(RemoteTrackPublication, bool) + Send>;
+type VideoDimensionsChangedHandler = Box<dyn Fn(RemoteTrackPublication, TrackDimension) + Send>;
 
 #[derive(Default)]
 struct RemoteEvents {
@@ -37,6 +38,7 @@ struct RemoteEvents {
     permission_status_changed: Mutex<Option<PermissionStatusChangedHandler>>,
     subscription_update_needed: Mutex<Option<SubscriptionUpdateNeededHandler>>,
     enabled_status_changed: Mutex<Option<EnabledStatusChangedHandler>>,
+    video_dimensions_changed: Mutex<Option<VideoDimensionsChangedHandler>>,
 }
 
 #[derive(Debug)]
@@ -215,6 +217,13 @@ impl RemoteTrackPublication {
         *self.remote.events.enabled_status_changed.lock() = Some(Box::new(f));
     }
 
+    pub(crate) fn on_video_dimensions_changed(
+        &self,
+        f: impl Fn(RemoteTrackPublication, TrackDimension) + Send + 'static,
+    ) {
+        *self.remote.events.video_dimensions_changed.lock() = Some(Box::new(f));
+    }
+
     pub fn set_subscribed(&self, subscribed: bool) {
         let old_subscription_state = self.subscription_status();
         let old_permission_state = self.permission_status();
@@ -262,6 +271,17 @@ impl RemoteTrackPublication {
         }
     }
 
+    pub fn update_video_dimensions(&self, dimension: TrackDimension) {
+        if self.is_subscribed() && dimension != self.dimension() {
+            // Request to send an update to the SFU
+            if let Some(video_dimensions_changed) =
+                self.remote.events.video_dimensions_changed.lock().as_ref()
+            {
+                video_dimensions_changed(self.clone(), dimension)
+            }
+        }
+    }
+
     pub fn subscription_status(&self) -> SubscriptionStatus {
         if !self.remote.info.read().subscribed {
             return SubscriptionStatus::Unsubscribed;