livekit, livekit-ffi: Update to v2 APIs (#342)

* room: update DataPacket

* Remove SID from API and mv {,remote}_participants

* Rename tracks to track_publications

* livekit-ffi: destination_{sids -> identities}

* async room.sid(&self)

* bump: livekit 0.4.0, livekit-ffi 0.6.0

* ffi: PublishData.{kind -> reliable}

* livekit/room: use inner sid for impl Debug

* Add sid FfiRequest, streamline async sid()

* livekit-protocol: add Promise<T>

* livekit-protocol: allow multi-recv from Promise<T>

* livekit-ffi: remove FfiClient GetRoomSid

* livekit-ffi: don't expose RoomSidChanged

Author: nbsp

Cargo.lock

@@ -8,7 +8,7 @@ description = "Livekit safe bindings to libwebrtc"
 repository = "https://github.com/livekit/rust-sdks"
 
 [dependencies]
-livekit-protocol = { path = "../livekit-protocol", version = "0.3.2" }
+livekit-protocol = { path = "../livekit-protocol", version = "0.3.3" }
 log = "0.4"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

libwebrtc/Cargo.toml

@@ -65,7 +65,7 @@ rustls-tls-webpki-roots = [
 __rustls-tls = ["tokio-tungstenite?/__rustls-tls", "reqwest?/__rustls"]
 
 [dependencies]
-livekit-protocol = { path = "../livekit-protocol", version = "0.3.2" }
+livekit-protocol = { path = "../livekit-protocol", version = "0.3.3" }
 thiserror = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 sha2 = "0.10"

livekit-api/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "livekit-ffi"
-version = "0.5.0"
+version = "0.6.0"
 edition = "2021"
 license = "Apache-2.0"
 description = "FFI interface for bindings in other languages"
@@ -18,8 +18,8 @@ __rustls-tls = ["livekit/__rustls-tls"]
 tracing = ["tokio/tracing", "console-subscriber"]
 
 [dependencies]
-livekit = { path = "../livekit", version = "0.3.2" }
-livekit-protocol = { path = "../livekit-protocol", version = "0.3.2" }
+livekit = { path = "../livekit", version = "0.4.0" }
+livekit-protocol = { path = "../livekit-protocol", version = "0.3.3" }
 tokio = { version = "1", features = ["full", "parking_lot"] }
 futures-util = { version = "0.3", default-features = false, features = ["sink"] }
 parking_lot = { version = "0.12", features = ["deadlock_detection"] }

livekit-ffi/Cargo.toml

@@ -88,9 +88,10 @@ message PublishDataRequest {
   uint64 local_participant_handle = 1;
   uint64 data_ptr = 2;
   uint64 data_len = 3;
-  DataPacketKind kind = 4;
-  repeated string destination_sids = 5; // destination
+  bool reliable = 4;
+  repeated string destination_sids = 5 [deprecated=true];
   optional string topic = 6;
+  repeated string destination_identities = 7;
 }
 message PublishDataResponse {
   uint64 async_id = 1;
@@ -160,7 +161,6 @@ message GetSessionStatsCallback {
   repeated RtcStats subscriber_stats = 4;
 }
 
-
 //
 // Options
 //
@@ -273,9 +273,10 @@ message RoomEvent {
     TrackUnmuted track_unmuted = 12;
     ActiveSpeakersChanged active_speakers_changed = 13;
     RoomMetadataChanged room_metadata_changed = 14;
-    ParticipantMetadataChanged participant_metadata_changed = 15;
-    ParticipantNameChanged participant_name_changed = 16;
-    ConnectionQualityChanged connection_quality_changed = 17;
+    RoomSidChanged room_sid_changed = 15;
+    ParticipantMetadataChanged participant_metadata_changed = 16;
+    ParticipantNameChanged participant_name_changed = 17;
+    ConnectionQualityChanged connection_quality_changed = 18;
     ConnectionStateChanged connection_state_changed = 19;
     // Connected connected = 20;
     Disconnected disconnected = 21;
@@ -364,6 +365,10 @@ message RoomMetadataChanged {
   string metadata = 1;
 }
 
+message RoomSidChanged {
+  string sid = 1;
+}
+
 message ParticipantMetadataChanged { 
   string participant_sid = 1;
   string metadata = 2;

livekit-ffi/protocol/room.proto

@@ -215,6 +215,10 @@ impl From<proto::AudioEncoding> for AudioEncoding {
 impl From<&FfiRoom> for proto::RoomInfo {
     fn from(value: &FfiRoom) -> Self {
         let room = &value.inner.room;
-        Self { sid: room.sid().into(), name: room.name(), metadata: room.metadata() }
+        Self {
+            sid: room.maybe_sid().unwrap_or_default().into(),
+            name: room.name(),
+            metadata: room.metadata(),
+        }
     }
 }

livekit-ffi/src/conversion/room.rs

@@ -1,5 +1,4 @@
 // @generated
-// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FrameCryptor {
@@ -2019,13 +2018,15 @@ pub struct PublishDataRequest {
     pub data_ptr: u64,
     #[prost(uint64, tag="3")]
     pub data_len: u64,
-    #[prost(enumeration="DataPacketKind", tag="4")]
-    pub kind: i32,
-    /// destination
+    #[prost(bool, tag="4")]
+    pub reliable: bool,
+    #[deprecated]
     #[prost(string, repeated, tag="5")]
     pub destination_sids: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
     #[prost(string, optional, tag="6")]
     pub topic: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(string, repeated, tag="7")]
+    pub destination_identities: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -2044,20 +2045,6 @@ pub struct PublishDataCallback {
 /// Publish transcription messages to room
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
-pub struct TranscriptionSegment {
-    #[prost(string, tag="1")]
-    pub id: ::prost::alloc::string::String,
-    #[prost(string, tag="2")]
-    pub text: ::prost::alloc::string::String,
-    #[prost(uint64, tag="3")]
-    pub start_time: u64,
-    #[prost(uint64, tag="4")]
-    pub end_time: u64,
-    #[prost(bool, tag="5")]
-    pub r#final: bool,
-}
-#[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct PublishTranscriptionRequest {
     #[prost(uint64, tag="1")]
     pub local_participant_handle: u64,
@@ -2240,6 +2227,20 @@ pub struct RoomOptions {
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TranscriptionSegment {
+    #[prost(string, tag="1")]
+    pub id: ::prost::alloc::string::String,
+    #[prost(string, tag="2")]
+    pub text: ::prost::alloc::string::String,
+    #[prost(uint64, tag="3")]
+    pub start_time: u64,
+    #[prost(uint64, tag="4")]
+    pub end_time: u64,
+    #[prost(bool, tag="5")]
+    pub r#final: bool,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct BufferInfo {
     #[prost(uint64, tag="1")]
     pub data_ptr: u64,
@@ -2259,7 +2260,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, 19, 21, 22, 23, 24, 25, 26")]
+    #[prost(oneof="room_event::Message", tags="2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 21, 22, 23, 24, 25, 26")]
     pub message: ::core::option::Option<room_event::Message>,
 }
 /// Nested message and enum types in `RoomEvent`.
@@ -2294,10 +2295,12 @@ pub mod room_event {
         #[prost(message, tag="14")]
         RoomMetadataChanged(super::RoomMetadataChanged),
         #[prost(message, tag="15")]
-        ParticipantMetadataChanged(super::ParticipantMetadataChanged),
+        RoomSidChanged(super::RoomSidChanged),
         #[prost(message, tag="16")]
-        ParticipantNameChanged(super::ParticipantNameChanged),
+        ParticipantMetadataChanged(super::ParticipantMetadataChanged),
         #[prost(message, tag="17")]
+        ParticipantNameChanged(super::ParticipantNameChanged),
+        #[prost(message, tag="18")]
         ConnectionQualityChanged(super::ConnectionQualityChanged),
         #[prost(message, tag="19")]
         ConnectionStateChanged(super::ConnectionStateChanged),
@@ -2445,6 +2448,12 @@ pub struct RoomMetadataChanged {
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
+pub struct RoomSidChanged {
+    #[prost(string, tag="1")]
+    pub sid: ::prost::alloc::string::String,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ParticipantMetadataChanged {
     #[prost(string, tag="1")]
     pub participant_sid: ::prost::alloc::string::String,

livekit-ffi/src/livekit.proto.rs

@@ -166,6 +166,21 @@ impl FfiRoom {
                         },
                     ));
 
+                    // Update Room SID on promise resolve
+                    let room_handle = inner.handle_id.clone();
+                    server.async_runtime.spawn(async move {
+                        let _ = server.send_event(proto::ffi_event::Message::RoomEvent(
+                            proto::RoomEvent {
+                                room_handle,
+                                message: Some(proto::room_event::Message::RoomSidChanged(
+                                    proto::RoomSidChanged {
+                                        sid: ffi_room.inner.room.sid().await.into(),
+                                    },
+                                )),
+                            },
+                        ));
+                    });
+
                     // Forward events
                     let event_handle = server.watch_panic({
                         let close_rx = close_rx.resubscribe();
@@ -237,20 +252,20 @@ impl RoomInner {
         }
         .to_vec();
 
-        let kind = publish.kind();
-        let destination_sids = publish.destination_sids;
+        let reliable = publish.reliable;
+        let topic = publish.topic;
+        let destination_identities = publish.destination_identities;
         let async_id = server.next_id();
 
         if let Err(err) = self.data_tx.send(FfiDataPacket {
             payload: DataPacket {
                 payload: data.to_vec(), // Avoid copy?
-                kind: kind.into(),
-                topic: publish.topic,
-                destination_sids: destination_sids
+                reliable,
+                topic,
+                destination_identities: destination_identities
                     .into_iter()
                     .map(|str| str.try_into().unwrap())
                     .collect(),
-                destination_identities: Vec::new(), // TODO
             },
             async_id,
         }) {

livekit-ffi/src/server/room.rs

@@ -1,14 +1,14 @@
 [package]
 name = "livekit-protocol"
-version = "0.3.2"
+version = "0.3.3"
 edition = "2021"
 license = "Apache-2.0"
 description = "Livekit protocol and utilities for the Rust SDK"
 repository = "https://github.com/livekit/rust-sdks"
 
 [dependencies]
 livekit-runtime = { path = "../livekit-runtime", version = "0.3.0" }
-tokio = { version = "1", default-features = false, features = [ "sync", "macros" ] }
+tokio = { version = "1", default-features = false, features = [ "sync", "macros", "rt" ] }
 futures-util = { version = "0.3", features = ["sink"] }
 parking_lot = "0.12"
 prost = "0.12"

livekit-protocol/Cargo.toml

@@ -18,6 +18,7 @@
 pub mod debouncer;
 pub mod enum_dispatch;
 pub mod observer;
+pub mod promise;
 
 include!("livekit.rs");
 

livekit-protocol/src/lib.rs

@@ -0,0 +1,50 @@
+// Copyright 2024 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.
+
+use tokio::sync::{oneshot, Mutex};
+
+pub struct Promise<T> {
+    tx: Mutex<Option<oneshot::Sender<T>>>,
+    rx: Mutex<Option<oneshot::Receiver<T>>>,
+    result: Mutex<Option<T>>,
+}
+
+impl<T: Clone> Promise<T> {
+    pub fn new() -> Self {
+        let (tx, rx) = oneshot::channel();
+        Self { tx: Mutex::new(Some(tx)), rx: Mutex::new(Some(rx)), result: Default::default() }
+    }
+
+    pub fn resolve(&self, result: T) -> Result<(), &'static str> {
+        let mut tx = self.tx.try_lock().unwrap();
+        if tx.is_some() {
+            let _ = tx.take().unwrap().send(result);
+            Ok(())
+        } else {
+            Err("promise already used")
+        }
+    }
+
+    pub async fn result(&self) -> T {
+        let mut rx = self.rx.lock().await;
+        if rx.is_some() {
+            self.result.lock().await.replace(rx.take().unwrap().await.unwrap());
+        }
+        self.result.lock().await.clone().unwrap()
+    }
+
+    pub fn try_result(&self) -> Option<T> {
+        self.result.try_lock().unwrap().clone()
+    }
+}

livekit-protocol/src/promise.rs

@@ -1,6 +1,6 @@
 [package]
 name = "livekit"
-version = "0.3.2"
+version = "0.4.0"
 edition = "2021"
 license = "Apache-2.0"
 description = "Rust Client SDK for LiveKit"
@@ -30,7 +30,7 @@ __lk-internal = []
 livekit-runtime = { path = "../livekit-runtime", version = "0.3.0", default-features = false }
 livekit-api = { path = "../livekit-api", version = "0.3.2", default-features = false }
 libwebrtc = { path = "../libwebrtc", version = "0.3.2" }
-livekit-protocol = { path = "../livekit-protocol", version = "0.3.2" }
+livekit-protocol = { path = "../livekit-protocol", version = "0.3.3" }
 prost = "0.12"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1.0"

livekit/Cargo.toml

@@ -15,7 +15,7 @@ pub struct ParticipantIdentity(pub String);
 #[derive(Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
 pub struct TrackSid(String);
 
-#[derive(Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
+#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
 pub struct RoomSid(String);
 
 impl From<String> for ParticipantIdentity {