Data streams v2

.changeset/data_streams_v2.md

@@ -0,0 +1,10 @@
+---
+livekit: patch
+livekit-api: patch
+livekit-datatrack: patch
+livekit-ffi: patch
+livekit-protocol: patch
+livekit-uniffi: patch
+---
+
+Add data streams v2 - #1192 (@1egoman)

livekit-api/src/signal_client/mod.rs

@@ -57,14 +57,21 @@
 pub const PROTOCOL_VERSION: u32 = 17;
 
 /// Capabilities the Rust SDK advertises to the SFU at connect time.
-const CLIENT_CAPABILITIES: &[proto::client_info::Capability] =
-    &[proto::client_info::Capability::CapPacketTrailer];
-
-pub use livekit_common::{CLIENT_PROTOCOL_DATA_STREAM_RPC, CLIENT_PROTOCOL_DEFAULT};
+///
+/// `CapCompressionDeflateRaw` is always advertised because the SDK's deflate-raw codec
+/// (flate2/miniz_oxide) is pure-Rust and compiled in unconditionally.
+const CLIENT_CAPABILITIES: &[proto::client_info::Capability] = &[
+    proto::client_info::Capability::CapPacketTrailer,
+    proto::client_info::Capability::CapCompressionDeflateRaw,
+];
+
+pub use livekit_common::{
+    CLIENT_PROTOCOL_DATA_STREAM_RPC, CLIENT_PROTOCOL_DATA_STREAM_V2, CLIENT_PROTOCOL_DEFAULT,
+};
 
 /// The client protocol which is sent to other clients and indicates the set of apis that other
 /// clients should assume this client supports.
-const CLIENT_PROTOCOL_VERSION: i32 = CLIENT_PROTOCOL_DATA_STREAM_RPC;
+const CLIENT_PROTOCOL_VERSION: i32 = CLIENT_PROTOCOL_DATA_STREAM_V2;
 
 #[derive(Error, Debug)]
 pub enum SignalError {
@@ -1013,7 +1020,7 @@
     /// in `send`. The stream slot is None so any actual write would be dropped,
     /// which is fine — these tests only assert which side of the queue each
     /// message lands on.
     fn make_stub_inner() -> Arc<SignalInner> {
         Arc::new(SignalInner {
             stream: AsyncRwLock::new(None),
             token: Mutex::new(String::new()),

livekit-common/src/lib.rs

@@ -13,7 +13,8 @@
 // limitations under the License.
 
 //! Foundational types shared across LiveKit crates: participant identities, the
-//! encryption enum, and client-protocol constants.
+//! encryption/capability enums, client-protocol constants, and the remote-participant
+//! registry trait consulted by the data-stream and RPC send paths.
 
 use std::fmt::Display;
 
@@ -31,6 +32,9 @@ pub const CLIENT_PROTOCOL_DEFAULT: i32 = 0;
 /// RPC v2 (see RPC spec).
 pub const CLIENT_PROTOCOL_DATA_STREAM_RPC: i32 = 1;
 
+/// Understands inline single-packet data streams (data streams v2).
+pub const CLIENT_PROTOCOL_DATA_STREAM_V2: i32 = 2;
+
 // -------------------------------------------------------------------------------------------------
 // ParticipantIdentity
 // -------------------------------------------------------------------------------------------------
@@ -109,3 +113,66 @@ impl From<EncryptionType> for i32 {
         }
     }
 }
+
+// -------------------------------------------------------------------------------------------------
+// ClientCapability
+// -------------------------------------------------------------------------------------------------
+
+/// A capability a participant's client advertises, mirroring the `ClientInfo.Capability` protobuf
+/// enum.
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+#[non_exhaustive]
+pub enum ClientCapability {
+    Unused,
+    PacketTrailer,
+    CompressionDeflateRaw,
+}
+
+impl TryFrom<i32> for ClientCapability {
+    type Error = &'static str;
+
+    fn try_from(value: i32) -> Result<Self, Self::Error> {
+        match proto::client_info::Capability::try_from(value) {
+            Ok(proto::client_info::Capability::CapPacketTrailer) => Ok(Self::PacketTrailer),
+            Ok(proto::client_info::Capability::CapCompressionDeflateRaw) => {
+                Ok(Self::CompressionDeflateRaw)
+            }
+            Ok(proto::client_info::Capability::CapUnused) => Ok(Self::Unused),
+            Err(_) => Err("unknown client capability"),
+        }
+    }
+}
+
+impl From<ClientCapability> for i32 {
+    fn from(value: ClientCapability) -> Self {
+        match value {
+            ClientCapability::Unused => proto::client_info::Capability::CapUnused as i32,
+            ClientCapability::PacketTrailer => {
+                proto::client_info::Capability::CapPacketTrailer as i32
+            }
+            ClientCapability::CompressionDeflateRaw => {
+                proto::client_info::Capability::CapCompressionDeflateRaw as i32
+            }
+        }
+    }
+}
+
+// -------------------------------------------------------------------------------------------------
+// RemoteParticipantRegistry
+// -------------------------------------------------------------------------------------------------
+
+/// Read access to remote participants' advertised protocol and capabilities.
+///
+/// Used by downstream modules like the the RPC transport (v1/v2 transport selection) and
+/// the data-stream send path (inline / compression eligibility) to determine what level of support
+/// a participant has for protocol level features.
+pub trait RemoteParticipantRegistry: Send + Sync {
+    /// A remote participant's `client_protocol`, or `CLIENT_PROTOCOL_DEFAULT` (0) if unknown.
+    fn remote_client_protocol(&self, identity: &ParticipantIdentity) -> i32;
+
+    /// A remote participant's advertised capabilities, or empty if unknown.
+    fn remote_capabilities(&self, identity: &ParticipantIdentity) -> Vec<ClientCapability>;
+
+    /// The identities of every remote participant, used to resolve a broadcast send.
+    fn remote_identities(&self) -> Vec<ParticipantIdentity>;
+}

livekit-data-stream/Cargo.toml

@@ -8,8 +8,10 @@ edition.workspace = true
 repository.workspace = true
 
 [features]
-# Exposes test-only hooks used by downstream crates' test suites (e.g.
-# `TextStreamReader::new_for_test`). Forwarded from the `livekit` crate's `__lk-e2e-test` feature.
+# End-to-end testing hooks (exposes is_compressed/is_inline on stream info). Forwarded from
+# the `livekit` crate's `__lk-e2e-test` feature.
+__e2e-test = []
+# Exposes constructors used by downstream crates' test suites (e.g. `TextStreamReader::new_for_test`).
 test-utils = []
 
 [dependencies]
@@ -20,12 +22,18 @@ thiserror = { workspace = true }
 parking_lot = { workspace = true }
 bytes = { workspace = true }
 tokio = { workspace = true, default-features = false, features = ["sync", "fs", "io-util", "rt"] }
-futures-util = { workspace = true, default-features = false, features = ["sink"] }
+futures-util = { workspace = true, default-features = false, features = ["sink", "io"] }
 prost = "0.12"
 chrono = "0.4.38"
+# Still used by the streaming file-compression path (`send_file`/`write_file`) and the test
+# suite's cross-implementation interop check. The in-memory (de)compression paths use
+# `async-compression` instead.
 flate2 = "1"
 bmrng = "0.5.2"
 uuid = { version = "1", features = ["v4"] }
+async-compression = { version = "0.4.42", features = ["deflate", "futures-io"] }
+from_variants.workspace = true
 
 [dev-dependencies]
 tokio = { workspace = true, default-features = false, features = ["macros", "rt", "rt-multi-thread", "time"] }
+rand = { workspace = true }