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)

Cargo.toml

@@ -4,6 +4,8 @@ members = [
     "livekit",
     "livekit-api",
     "livekit-protocol",
+    "livekit-common",
+    "livekit-data-stream",
     "livekit-ffi",
     "livekit-uniffi",
     "livekit-datatrack",
@@ -51,6 +53,8 @@ livekit = { version = "0.7.50", path = "livekit" }
 livekit-api = { version = "0.5.4", path = "livekit-api" }
 livekit-ffi = { version = "0.12.68", path = "livekit-ffi" }
 livekit-datatrack = { version = "0.1.9", path = "livekit-datatrack" }
+livekit-common = { version = "0.1.0", path = "livekit-common" }
+livekit-data-stream = { version = "0.1.0", path = "livekit-data-stream" }
 livekit-protocol = { version = "0.7.10", path = "livekit-protocol" }
 livekit-runtime = { version = "0.4.0", path = "livekit-runtime" }
 soxr-sys = { version = "0.1.3", path = "soxr-sys" }

livekit-api/Cargo.toml

@@ -101,6 +101,7 @@ __rustls-tls = ["tokio-tungstenite?/__rustls-tls", "reqwest?/__rustls"]
 
 [dependencies]
 livekit-protocol = { workspace = true }
+livekit-common = { workspace = true }
 thiserror = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 sha2 = "0.10"

livekit-api/src/signal_client/mod.rs

@@ -57,18 +57,21 @@ const VALIDATE_TIMEOUT: Duration = Duration::from_secs(3);
 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];
-
-/// Default value for `ClientInfo.client_protocol` when a participant has not
-/// advertised one (treat as v1-only / no data-stream RPC support).
-pub const CLIENT_PROTOCOL_DEFAULT: i32 = 0;
-/// `ClientInfo.client_protocol` value indicating support for RPC v2 over data streams.
-pub const CLIENT_PROTOCOL_DATA_STREAM_RPC: i32 = 1;
+///
+/// `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 {

livekit-common/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "livekit-common"
+description = "Common foundational types shared across LiveKit crates"
+version = "0.1.0"
+readme = "README.md"
+license.workspace = true
+edition.workspace = true
+repository.workspace = true
+
+[dependencies]
+livekit-protocol = { workspace = true }

livekit-common/README.md

@@ -0,0 +1,6 @@
+# LiveKit Common
+
+An internal crate which holds shared data structures that many downstream modules all use, like
+`ParticipantIdentity` or `ClientCapability`.
+
+To build applications with LiveKit, please use the public APIs provided by the [livekit](../livekit) crate.

livekit-common/src/lib.rs

@@ -0,0 +1,176 @@
+// Copyright 2026 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.
+
+//! Foundational types shared across LiveKit crates: participant identities, the
+//! 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;
+
+use livekit_protocol as proto;
+
+// -------------------------------------------------------------------------------------------------
+// Client protocol
+// -------------------------------------------------------------------------------------------------
+
+/// Legacy client. No v2 data-stream features.
+pub const CLIENT_PROTOCOL_DEFAULT: i32 = 0;
+
+/// RPC v2 (see RPC spec). No v2 data-stream features.
+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
+// -------------------------------------------------------------------------------------------------
+
+#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
+pub struct ParticipantIdentity(pub String);
+
+impl From<String> for ParticipantIdentity {
+    fn from(value: String) -> Self {
+        Self(value)
+    }
+}
+
+impl From<&str> for ParticipantIdentity {
+    fn from(value: &str) -> Self {
+        Self(value.to_string())
+    }
+}
+
+impl From<ParticipantIdentity> for String {
+    fn from(value: ParticipantIdentity) -> Self {
+        value.0
+    }
+}
+
+impl Display for ParticipantIdentity {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+impl ParticipantIdentity {
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+// -------------------------------------------------------------------------------------------------
+// EncryptionType
+// -------------------------------------------------------------------------------------------------
+
+#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)]
+pub enum EncryptionType {
+    #[default]
+    None,
+    Gcm,
+    Custom,
+}
+
+impl From<proto::encryption::Type> for EncryptionType {
+    fn from(value: proto::encryption::Type) -> Self {
+        match value {
+            proto::encryption::Type::None => Self::None,
+            proto::encryption::Type::Gcm => Self::Gcm,
+            proto::encryption::Type::Custom => Self::Custom,
+        }
+    }
+}
+
+impl From<EncryptionType> for proto::encryption::Type {
+    fn from(value: EncryptionType) -> Self {
+        match value {
+            EncryptionType::None => Self::None,
+            EncryptionType::Gcm => Self::Gcm,
+            EncryptionType::Custom => Self::Custom,
+        }
+    }
+}
+
+impl From<EncryptionType> for i32 {
+    fn from(value: EncryptionType) -> Self {
+        match value {
+            EncryptionType::None => 0,
+            EncryptionType::Gcm => 1,
+            EncryptionType::Custom => 2,
+        }
+    }
+}
+
+// -------------------------------------------------------------------------------------------------
+// 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

@@ -0,0 +1,34 @@
+[package]
+name = "livekit-data-stream"
+description = "Data stream core logic for LiveKit"
+version = "0.1.0"
+readme = "README.md"
+license.workspace = true
+edition.workspace = true
+repository.workspace = true
+
+[features]
+# 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]
+livekit-common = { workspace = true }
+livekit-protocol = { workspace = true }
+log = { workspace = true }
+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"] }
+prost = "0.12"
+chrono = "0.4.38"
+flate2 = "1"
+bmrng = "0.5.2"
+uuid = { version = "1", features = ["v4"] }
+
+[dev-dependencies]
+tokio = { workspace = true, default-features = false, features = ["macros", "rt", "rt-multi-thread", "time"] }
+rand = { workspace = true }

livekit-data-stream/README.md

@@ -0,0 +1,7 @@
+# LiveKit Data Stream
+
+**Important**:
+This is an internal crate that powers the data streams feature in LiveKit client SDKs (including [Rust](https://crates.io/crates/livekit) and others) and is not usable directly.
+
+To use data streams in your application, please use the public APIs provided by the
+[livekit](../livekit) crate and other client sdks.