From 452024f19af84bda6bc3739a1a254398acdf0355 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Thu, 2 Jul 2026 17:05:57 -0400 Subject: [PATCH 01/19] feat: add initial livekit-common implementation --- livekit-common/Cargo.toml | 12 +++ livekit-common/README.md | 8 ++ livekit-common/src/lib.rs | 178 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 198 insertions(+) create mode 100644 livekit-common/Cargo.toml create mode 100644 livekit-common/README.md create mode 100644 livekit-common/src/lib.rs diff --git a/livekit-common/Cargo.toml b/livekit-common/Cargo.toml new file mode 100644 index 000000000..0b83ade48 --- /dev/null +++ b/livekit-common/Cargo.toml @@ -0,0 +1,12 @@ +[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 } +serde = { workspace = true, features = ["derive"] } diff --git a/livekit-common/README.md b/livekit-common/README.md new file mode 100644 index 000000000..17f6cf712 --- /dev/null +++ b/livekit-common/README.md @@ -0,0 +1,8 @@ +# LiveKit Common + +**Important**: +This is an internal crate holding foundational types shared across the LiveKit client SDK crates +(identities, encryption/capability enums, client-protocol constants, and the remote-participant +registry trait). It is not intended to be used directly. + +To build applications with LiveKit, please use the public APIs provided by the client SDKs. diff --git a/livekit-common/src/lib.rs b/livekit-common/src/lib.rs new file mode 100644 index 000000000..69a11de77 --- /dev/null +++ b/livekit-common/src/lib.rs @@ -0,0 +1,178 @@ +// 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; +use serde::{Deserialize, Serialize}; + +// ------------------------------------------------------------------------------------------------- +// 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, Serialize, Deserialize)] +pub struct ParticipantIdentity(pub String); + +impl From 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 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, Serialize, Deserialize)] +pub enum EncryptionType { + #[default] + None, + Gcm, + Custom, +} + +impl From 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 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 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 (mirrors `ClientInfo.Capability`). +/// +/// Stored typed rather than as the raw protobuf `i32` so accessors don't leak protobuf types. +#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)] +#[non_exhaustive] +pub enum ClientCapability { + Unused, + PacketTrailer, + CompressionDeflateRaw, +} + +impl TryFrom for ClientCapability { + type Error = &'static str; + + fn try_from(value: i32) -> Result { + 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 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. +/// +/// Shared by the RPC transport (v1/v2 transport selection) and the data-stream send +/// path (inline / compression eligibility), so both consult a single abstraction over +/// the room's remote participants and both are unit-testable with a fake. +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; + + /// The identities of every remote participant, used to resolve a broadcast send. + fn remote_identities(&self) -> Vec; +} From 6904329189889203325c2643cf36b0c75b0cd538 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Thu, 2 Jul 2026 17:25:10 -0400 Subject: [PATCH 02/19] fix: remove serde from livekit-common Not needed for now, can be re-added with an feature later --- livekit-common/Cargo.toml | 1 - livekit-common/src/lib.rs | 7 +++---- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/livekit-common/Cargo.toml b/livekit-common/Cargo.toml index 0b83ade48..c3ec0491d 100644 --- a/livekit-common/Cargo.toml +++ b/livekit-common/Cargo.toml @@ -9,4 +9,3 @@ repository.workspace = true [dependencies] livekit-protocol = { workspace = true } -serde = { workspace = true, features = ["derive"] } diff --git a/livekit-common/src/lib.rs b/livekit-common/src/lib.rs index 69a11de77..d2c1ac47a 100644 --- a/livekit-common/src/lib.rs +++ b/livekit-common/src/lib.rs @@ -19,7 +19,6 @@ use std::fmt::Display; use livekit_protocol as proto; -use serde::{Deserialize, Serialize}; // ------------------------------------------------------------------------------------------------- // Client protocol @@ -38,7 +37,7 @@ pub const CLIENT_PROTOCOL_DATA_STREAM_V2: i32 = 2; // ParticipantIdentity // ------------------------------------------------------------------------------------------------- -#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord, Serialize, Deserialize)] +#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] pub struct ParticipantIdentity(pub String); impl From for ParticipantIdentity { @@ -75,7 +74,7 @@ impl ParticipantIdentity { // EncryptionType // ------------------------------------------------------------------------------------------------- -#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)] pub enum EncryptionType { #[default] None, @@ -120,7 +119,7 @@ impl From for i32 { /// A capability a participant's client advertises (mirrors `ClientInfo.Capability`). /// /// Stored typed rather than as the raw protobuf `i32` so accessors don't leak protobuf types. -#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Eq, PartialEq)] #[non_exhaustive] pub enum ClientCapability { Unused, From 2400b0f70a15960b0ce984fc0093835e0f9bbdda Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Thu, 2 Jul 2026 17:24:24 -0400 Subject: [PATCH 03/19] refactor: update more livekit crate references to point to livekit-common instead --- Cargo.lock | 8 ++++++++ Cargo.toml | 2 ++ livekit-api/Cargo.toml | 1 + livekit-api/src/signal_client/mod.rs | 11 ++++++----- livekit/src/proto.rs | 29 ---------------------------- livekit/src/room/e2ee/mod.rs | 8 +------- livekit/src/room/id.rs | 18 ++--------------- livekit/src/room/participant/mod.rs | 2 ++ 8 files changed, 22 insertions(+), 57 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f6ac222fe..9db3ee4e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3957,6 +3957,7 @@ dependencies = [ "http 1.4.0", "isahc", "jsonwebtoken", + "livekit-common", "livekit-protocol", "livekit-runtime", "log", @@ -3979,6 +3980,13 @@ dependencies = [ "url", ] +[[package]] +name = "livekit-common" +version = "0.1.0" +dependencies = [ + "livekit-protocol", +] + [[package]] name = "livekit-datatrack" version = "0.1.9" diff --git a/Cargo.toml b/Cargo.toml index 93d4a11df..a4c208faf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "livekit", "livekit-api", "livekit-protocol", + "livekit-common", "livekit-ffi", "livekit-uniffi", "livekit-datatrack", @@ -51,6 +52,7 @@ 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-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" } diff --git a/livekit-api/Cargo.toml b/livekit-api/Cargo.toml index 8afd55fc7..53183a05f 100644 --- a/livekit-api/Cargo.toml +++ b/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" diff --git a/livekit-api/src/signal_client/mod.rs b/livekit-api/src/signal_client/mod.rs index 76b7ba491..49ecbf305 100644 --- a/livekit-api/src/signal_client/mod.rs +++ b/livekit-api/src/signal_client/mod.rs @@ -60,11 +60,12 @@ pub const PROTOCOL_VERSION: u32 = 17; 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; +// The canonical `client_protocol` constants live in `livekit-common` (shared with the +// data-stream crate); re-exported here so existing `livekit_api::signal_client::CLIENT_PROTOCOL_*` +// references keep resolving. +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. diff --git a/livekit/src/proto.rs b/livekit/src/proto.rs index 78e690d73..15d5fdf71 100644 --- a/livekit/src/proto.rs +++ b/livekit/src/proto.rs @@ -141,35 +141,6 @@ impl From for DataPacketKind { } } -impl From for EncryptionType { - fn from(value: livekit_protocol::encryption::Type) -> Self { - match value { - livekit_protocol::encryption::Type::None => Self::None, - livekit_protocol::encryption::Type::Gcm => Self::Gcm, - livekit_protocol::encryption::Type::Custom => Self::Custom, - } - } -} - -impl From for encryption::Type { - fn from(value: EncryptionType) -> Self { - match value { - EncryptionType::None => Self::None, - EncryptionType::Gcm => Self::Gcm, - EncryptionType::Custom => Self::Custom, - } - } -} - -impl From for i32 { - fn from(value: EncryptionType) -> Self { - match value { - EncryptionType::None => 0, - EncryptionType::Gcm => 1, - EncryptionType::Custom => 2, - } - } -} impl From for participant::ParticipantState { fn from(value: participant_info::State) -> Self { diff --git a/livekit/src/room/e2ee/mod.rs b/livekit/src/room/e2ee/mod.rs index e1235d81d..21f6cfca7 100644 --- a/livekit/src/room/e2ee/mod.rs +++ b/livekit/src/room/e2ee/mod.rs @@ -22,13 +22,7 @@ pub mod manager; /// Provider implementations for data track. pub(crate) mod data_track; -#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)] -pub enum EncryptionType { - #[default] - None, - Gcm, - Custom, -} +pub(crate) use livekit_common::EncryptionType; #[derive(Clone)] pub struct E2eeOptions { diff --git a/livekit/src/room/id.rs b/livekit/src/room/id.rs index 800496371..c1ea11a74 100644 --- a/livekit/src/room/id.rs +++ b/livekit/src/room/id.rs @@ -20,30 +20,17 @@ const ROOM_PREFIX: &str = "RM_"; const PARTICIPANT_PREFIX: &str = "PA_"; const TRACK_PREFIX: &str = "TR_"; +pub use livekit_common::ParticipantIdentity; + #[derive(Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] pub struct ParticipantSid(String); -#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] -pub struct ParticipantIdentity(pub String); - #[derive(Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] pub struct TrackSid(String); #[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] pub struct RoomSid(String); -impl From for ParticipantIdentity { - fn from(value: String) -> Self { - Self(value) - } -} - -impl From<&str> for ParticipantIdentity { - fn from(value: &str) -> Self { - Self(value.to_string()) - } -} - macro_rules! impl_string_into { ($from:ty) => { impl From<$from> for String { @@ -67,7 +54,6 @@ macro_rules! impl_string_into { } impl_string_into!(ParticipantSid); -impl_string_into!(ParticipantIdentity); impl_string_into!(TrackSid); impl_string_into!(RoomSid); diff --git a/livekit/src/room/participant/mod.rs b/livekit/src/room/participant/mod.rs index 5dcdc83d7..7779361f2 100644 --- a/livekit/src/room/participant/mod.rs +++ b/livekit/src/room/participant/mod.rs @@ -85,6 +85,8 @@ pub enum DisconnectReason { AgentError, } +pub use livekit_common::ClientCapability; + #[derive(Debug, Clone)] pub enum Participant { Local(LocalParticipant), From 4d0c8f418fa689be6c634cbc2ebf1ea56b806f88 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Mon, 6 Jul 2026 11:43:00 -0400 Subject: [PATCH 04/19] refactor: move data stream code out into new livekit-data-stream crate --- Cargo.lock | 43 +++++++++++++++- Cargo.toml | 2 + livekit-data-stream/Cargo.toml | 33 ++++++++++++ livekit-data-stream/README.md | 6 +++ .../src}/incoming.rs | 39 +++++++-------- .../mod.rs => livekit-data-stream/src/lib.rs | 50 ++++++++++++------- .../src}/outgoing.rs | 21 ++++---- .../src}/utf8_chunk.rs | 0 livekit/Cargo.toml | 6 ++- livekit/src/proto.rs | 4 +- livekit/src/room/e2ee/mod.rs | 2 +- livekit/src/room/mod.rs | 30 ++++++++--- livekit/src/utils/mod.rs | 1 - 13 files changed, 173 insertions(+), 64 deletions(-) create mode 100644 livekit-data-stream/Cargo.toml create mode 100644 livekit-data-stream/README.md rename {livekit/src/room/data_stream => livekit-data-stream/src}/incoming.rs (92%) rename livekit/src/room/data_stream/mod.rs => livekit-data-stream/src/lib.rs (88%) rename {livekit/src/room/data_stream => livekit-data-stream/src}/outgoing.rs (98%) rename {livekit/src/utils => livekit-data-stream/src}/utf8_chunk.rs (100%) diff --git a/Cargo.lock b/Cargo.lock index 9db3ee4e6..28580a28b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2662,11 +2662,22 @@ dependencies = [ "cfg-if 1.0.4", "js-sys", "libc", - "r-efi", + "r-efi 5.3.0", "wasip2", "wasm-bindgen", ] +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if 1.0.4", + "libc", + "r-efi 6.0.0", +] + [[package]] name = "gif" version = "0.14.1" @@ -3925,6 +3936,8 @@ dependencies = [ "libloading 0.8.9", "libwebrtc", "livekit-api", + "livekit-common", + "livekit-data-stream", "livekit-datatrack", "livekit-protocol", "livekit-runtime", @@ -3987,6 +4000,25 @@ dependencies = [ "livekit-protocol", ] +[[package]] +name = "livekit-data-stream" +version = "0.1.0" +dependencies = [ + "bmrng", + "bytes", + "chrono", + "flate2", + "futures-util", + "livekit-common", + "livekit-protocol", + "log", + "parking_lot", + "prost 0.12.6", + "thiserror 2.0.18", + "tokio", + "uuid", +] + [[package]] name = "livekit-datatrack" version = "0.1.9" @@ -5950,6 +5982,12 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + [[package]] name = "rand" version = "0.8.5" @@ -8145,7 +8183,10 @@ version = "1.23.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bf80a72845275afea99e7f2b434723d3bc7e38470fcd1c7ed39a599c73319a53" dependencies = [ + "getrandom 0.4.3", + "js-sys", "serde_core", + "wasm-bindgen", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index a4c208faf..75e166ca4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,7 @@ members = [ "livekit-api", "livekit-protocol", "livekit-common", + "livekit-data-stream", "livekit-ffi", "livekit-uniffi", "livekit-datatrack", @@ -53,6 +54,7 @@ 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" } diff --git a/livekit-data-stream/Cargo.toml b/livekit-data-stream/Cargo.toml new file mode 100644 index 000000000..2ac95482f --- /dev/null +++ b/livekit-data-stream/Cargo.toml @@ -0,0 +1,33 @@ +[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"] } diff --git a/livekit-data-stream/README.md b/livekit-data-stream/README.md new file mode 100644 index 000000000..781b5457d --- /dev/null +++ b/livekit-data-stream/README.md @@ -0,0 +1,6 @@ +# LiveKit Data Track + +**Important**: +This is an internal crate that powers the data tracks feature in LiveKit client SDKs (including [Rust](https://crates.io/crates/livekit) and others) and is not usable directly. + +To use data tracks in your application, please use the public APIs provided by the client SDKs. diff --git a/livekit/src/room/data_stream/incoming.rs b/livekit-data-stream/src/incoming.rs similarity index 92% rename from livekit/src/room/data_stream/incoming.rs rename to livekit-data-stream/src/incoming.rs index 213a68c07..c1f726121 100644 --- a/livekit/src/room/data_stream/incoming.rs +++ b/livekit-data-stream/src/incoming.rs @@ -15,9 +15,9 @@ use super::{ AnyStreamInfo, ByteStreamInfo, StreamError, StreamProgress, StreamResult, TextStreamInfo, }; -use crate::{e2ee::EncryptionType, TakeCell}; use bytes::{Bytes, BytesMut}; use futures_util::{Stream, StreamExt}; +use livekit_common::EncryptionType; use livekit_protocol::data_stream as proto; use parking_lot::Mutex; use std::{ @@ -52,23 +52,6 @@ pub trait StreamReader: Stream> { fn read_all(self) -> impl std::future::Future> + Send; } -impl TakeCell -where - T: StreamReader, -{ - /// Takes the reader out of the cell if its info matches the given predicate. - /// - /// Use this method to conditionally handle incoming streams based on info fields - /// such as topic or attributes. - /// - /// This method will only take the reader if the provided predicate returns `true` when called with the reader's info. - /// If the predicate returns `false` or the reader has already been taken, this method returns `None`. - /// - pub fn take_if(&self, predicate: impl FnOnce(&T::Info) -> bool) -> Option { - self.take_if_raw(|reader| predicate(reader.info())) - } -} - /// Reader for an incoming byte data stream. pub struct ByteStreamReader { info: ByteStreamInfo, @@ -242,9 +225,12 @@ struct Descriptor { } #[derive(Clone)] -pub(crate) struct IncomingStreamManager { +pub struct IncomingStreamManager { inner: Arc>, open_tx: UnboundedSender<(AnyStreamReader, String)>, + /// Topics whose streams are handled internally by the SDK (e.g. RPC) and never surfaced as + /// application events. Supplied by the host crate so this crate stays decoupled from RPC. + reserved_topics: Arc<[String]>, } #[derive(Default)] @@ -253,9 +239,18 @@ struct ManagerInner { } impl IncomingStreamManager { - pub fn new() -> (Self, UnboundedReceiver<(AnyStreamReader, String)>) { + pub fn new( + reserved_topics: Vec, + ) -> (Self, UnboundedReceiver<(AnyStreamReader, String)>) { let (open_tx, open_rx) = mpsc::unbounded_channel(); - (Self { inner: Arc::new(Mutex::new(Default::default())), open_tx }, open_rx) + ( + Self { + inner: Arc::new(Mutex::new(Default::default())), + open_tx, + reserved_topics: reserved_topics.into(), + }, + open_rx, + ) } /// Handles an incoming header packet. @@ -265,7 +260,7 @@ impl IncomingStreamManager { identity: String, encryption_type: livekit_protocol::encryption::Type, ) { - let is_internal = super::is_internal_topic(&header.topic); + let is_internal = self.reserved_topics.iter().any(|t| t == &header.topic); let Ok(info) = AnyStreamInfo::try_from_with_encryption(header, encryption_type.into()) .inspect_err(|e| log::error!("Invalid header: {}", e)) else { diff --git a/livekit/src/room/data_stream/mod.rs b/livekit-data-stream/src/lib.rs similarity index 88% rename from livekit/src/room/data_stream/mod.rs rename to livekit-data-stream/src/lib.rs index 1be6cce7b..4d12fa6b0 100644 --- a/livekit/src/room/data_stream/mod.rs +++ b/livekit-data-stream/src/lib.rs @@ -13,27 +13,29 @@ // limitations under the License. use chrono::{DateTime, Utc}; -use libwebrtc::enum_dispatch; +use livekit_common::EncryptionType; use livekit_protocol::data_stream as proto; use std::collections::HashMap; use thiserror::Error; mod incoming; mod outgoing; +mod utf8_chunk; pub use incoming::*; pub use outgoing::*; -use crate::e2ee::EncryptionType; -use crate::room::rpc::{RPC_REQUEST_TOPIC, RPC_RESPONSE_TOPIC}; - -/// Data stream topics reserved for internal SDK use. Events for these -/// topics are handled within the `livekit` crate and never surfaced -/// through `RoomEvent`. -pub(crate) const INTERNAL_TOPICS: &[&str] = &[RPC_REQUEST_TOPIC, RPC_RESPONSE_TOPIC]; - -pub(crate) fn is_internal_topic(topic: &str) -> bool { - INTERNAL_TOPICS.contains(&topic) +/// Error returned by the packet transport when a data-stream packet fails to send. +/// +/// The stream managers only need to know that a send failed (they map it to +/// [`StreamError::SendFailed`]); the concrete engine error type stays in the `livekit` crate, +/// which bridges the outgoing packet channel to the RTC engine. +#[derive(Debug, Clone)] +pub struct SendError; + +/// Generates a random stream identifier (UUID v4). +pub(crate) fn create_random_uuid() -> String { + uuid::Uuid::new_v4().to_string() } /// Result type for data stream operations. @@ -252,12 +254,26 @@ pub(crate) enum AnyStreamInfo { } impl AnyStreamInfo { - enum_dispatch!( - [Byte, Text]; - pub fn id(self: &Self) -> &str; - pub fn total_length(self: &Self) -> Option; - pub fn encryption_type(self: &Self) -> EncryptionType; - ); + pub fn id(&self) -> &str { + match self { + Self::Byte(info) => info.id(), + Self::Text(info) => info.id(), + } + } + + pub fn total_length(&self) -> Option { + match self { + Self::Byte(info) => info.total_length(), + Self::Text(info) => info.total_length(), + } + } + + pub fn encryption_type(&self) -> EncryptionType { + match self { + Self::Byte(info) => info.encryption_type(), + Self::Text(info) => info.encryption_type(), + } + } } #[rustfmt::skip] diff --git a/livekit/src/room/data_stream/outgoing.rs b/livekit-data-stream/src/outgoing.rs similarity index 98% rename from livekit/src/room/data_stream/outgoing.rs rename to livekit-data-stream/src/outgoing.rs index 5c3e4c431..1fe46a97e 100644 --- a/livekit/src/room/data_stream/outgoing.rs +++ b/livekit-data-stream/src/outgoing.rs @@ -13,14 +13,13 @@ // limitations under the License. use super::{ - ByteStreamInfo, OperationType, StreamError, StreamProgress, StreamResult, TextStreamInfo, -}; -use crate::{ - id::ParticipantIdentity, rtc_engine::EngineError, utils::utf8_chunk::Utf8AwareChunkExt, + create_random_uuid, ByteStreamInfo, OperationType, SendError, StreamError, StreamProgress, + StreamResult, TextStreamInfo, }; +use crate::utf8_chunk::Utf8AwareChunkExt; use bmrng::unbounded::{UnboundedRequestReceiver, UnboundedRequestSender}; use chrono::Utc; -use libwebrtc::native::create_random_uuid; +use livekit_common::ParticipantIdentity; use livekit_protocol as proto; use std::{collections::HashMap, path::Path, sync::Arc}; use tokio::{io::AsyncReadExt, sync::Mutex}; @@ -136,7 +135,7 @@ impl<'a> StreamWriter<'a> for TextStreamWriter { struct RawStreamOpenOptions { header: proto::data_stream::Header, destination_identities: Vec, - packet_tx: UnboundedRequestSender>, + packet_tx: UnboundedRequestSender>, } struct RawStream { @@ -144,7 +143,7 @@ struct RawStream { progress: StreamProgress, is_closed: bool, /// Request channel for sending packets. - packet_tx: UnboundedRequestSender>, + packet_tx: UnboundedRequestSender>, } impl RawStream { @@ -182,7 +181,7 @@ impl RawStream { } async fn send_packet( - tx: &UnboundedRequestSender>, + tx: &UnboundedRequestSender>, packet: proto::DataPacket, ) -> StreamResult<()> { tx.send_receive(packet) @@ -282,11 +281,11 @@ pub struct StreamTextOptions { #[derive(Clone)] pub(crate) struct OutgoingStreamManager { /// Request channel for sending packets. - packet_tx: UnboundedRequestSender>, + packet_tx: UnboundedRequestSender>, } impl OutgoingStreamManager { - pub fn new() -> (Self, UnboundedRequestReceiver>) { + pub fn new() -> (Self, UnboundedRequestReceiver>) { let (packet_tx, packet_rx) = bmrng::unbounded_channel(); let manager = Self { packet_tx }; (manager, packet_rx) @@ -522,7 +521,7 @@ mod tests { let raw_stream = rt.block_on(async { let (packet_tx, mut packet_rx) = - bmrng::unbounded_channel::>(); + bmrng::unbounded_channel::>(); tokio::spawn(async move { while let Ok((_packet, responder)) = packet_rx.recv().await { diff --git a/livekit/src/utils/utf8_chunk.rs b/livekit-data-stream/src/utf8_chunk.rs similarity index 100% rename from livekit/src/utils/utf8_chunk.rs rename to livekit-data-stream/src/utf8_chunk.rs diff --git a/livekit/Cargo.toml b/livekit/Cargo.toml index 3eb83261e..c45ce3905 100644 --- a/livekit/Cargo.toml +++ b/livekit/Cargo.toml @@ -29,13 +29,15 @@ rustls-tls-native-roots = ["livekit-api/rustls-tls-native-roots"] rustls-tls-webpki-roots = ["livekit-api/rustls-tls-webpki-roots"] __rustls-tls = ["livekit-api/__rustls-tls"] __lk-internal = [] # internal features (used by livekit-ffi) -__lk-e2e-test = [] # end-to-end testing with a LiveKit server +__lk-e2e-test = ["livekit-data-stream/__e2e-test"] # end-to-end testing with a LiveKit server [dependencies] livekit-runtime = { workspace = true } livekit-api = { workspace = true } libwebrtc = { workspace = true } livekit-protocol = { workspace = true } +livekit-common = { workspace = true } +livekit-data-stream = { workspace = true } livekit-datatrack = { workspace = true } prost = "0.12" serde = { version = "1", features = ["derive"] } @@ -55,6 +57,8 @@ bmrng = "0.5.2" base64 = "0.22" [dev-dependencies] +# Enable data-stream test constructors (e.g. TextStreamReader::new_for_test) for our test suites. +livekit-data-stream = { workspace = true, features = ["test-utils"] } anyhow = "1.0.99" test-log = "0.2.18" test-case = "3.3" diff --git a/livekit/src/proto.rs b/livekit/src/proto.rs index 15d5fdf71..443fac407 100644 --- a/livekit/src/proto.rs +++ b/livekit/src/proto.rs @@ -14,9 +14,7 @@ use livekit_protocol::*; -use crate::{ - e2ee::EncryptionType, participant, room::ChatMessage as RoomChatMessage, track, DataPacketKind, -}; +use crate::{participant, room::ChatMessage as RoomChatMessage, track, DataPacketKind}; // Conversions impl From for participant::ConnectionQuality { diff --git a/livekit/src/room/e2ee/mod.rs b/livekit/src/room/e2ee/mod.rs index 21f6cfca7..43d4a908c 100644 --- a/livekit/src/room/e2ee/mod.rs +++ b/livekit/src/room/e2ee/mod.rs @@ -22,7 +22,7 @@ pub mod manager; /// Provider implementations for data track. pub(crate) mod data_track; -pub(crate) use livekit_common::EncryptionType; +pub use livekit_common::EncryptionType; #[derive(Clone)] pub struct E2eeOptions { diff --git a/livekit/src/room/mod.rs b/livekit/src/room/mod.rs index 6cfd57605..d6097dbcb 100644 --- a/livekit/src/room/mod.rs +++ b/livekit/src/room/mod.rs @@ -63,7 +63,7 @@ use crate::{ utils::{observer::Dispatcher, promise::Promise}, }; -pub mod data_stream; +pub use livekit_data_stream as data_stream; pub mod data_track; pub mod e2ee; pub mod id; @@ -679,7 +679,9 @@ impl Room { let (remote_dt_manager, remote_dt_input, remote_dt_output) = dt::remote::Manager::new(remote_dt_options); - let (incoming_stream_manager, open_rx) = IncomingStreamManager::new(); + let (incoming_stream_manager, open_rx) = IncomingStreamManager::new( + INTERNAL_DATA_STREAM_TOPICS.iter().map(|t| t.to_string()).collect(), + ); let (outgoing_stream_manager, packet_rx) = OutgoingStreamManager::new(); let room_info = join_response.room.unwrap(); @@ -1796,7 +1798,7 @@ impl RoomSession { participant_identity: String, encryption_type: proto::encryption::Type, ) { - let is_internal = data_stream::is_internal_topic(&header.topic); + let is_internal = is_internal_topic(&header.topic); self.incoming_stream_manager.handle_header( header.clone(), participant_identity.clone(), @@ -2249,7 +2251,7 @@ async fn incoming_data_stream_task( match reader { AnyStreamReader::Byte(reader) => { let topic = reader.info().topic.clone(); - if !data_stream::is_internal_topic(&topic) { + if !is_internal_topic(&topic) { dispatcher.dispatch(&RoomEvent::ByteStreamOpened { topic, reader: TakeCell::new(reader), @@ -2279,7 +2281,7 @@ async fn incoming_data_stream_task( }); } _ => { - if !data_stream::is_internal_topic(&topic) { + if !is_internal_topic(&topic) { dispatcher.dispatch(&RoomEvent::TextStreamOpened { topic, reader: TakeCell::new(reader), @@ -2298,16 +2300,30 @@ async fn incoming_data_stream_task( } } +/// Data stream topics reserved for internal SDK use (e.g. RPC). Events for these topics are +/// handled within the `livekit` crate and never surfaced through `RoomEvent`; the list is also +/// passed to `IncomingStreamManager` so it can flag internal streams. +const INTERNAL_DATA_STREAM_TOPICS: &[&str] = &[rpc::RPC_REQUEST_TOPIC, rpc::RPC_RESPONSE_TOPIC]; + +fn is_internal_topic(topic: &str) -> bool { + INTERNAL_DATA_STREAM_TOPICS.contains(&topic) +} + /// Receives packets from the outgoing stream manager and send them. async fn outgoing_data_stream_task( - mut packet_rx: UnboundedRequestReceiver>, + mut packet_rx: UnboundedRequestReceiver>, engine: Arc, mut close_rx: broadcast::Receiver<()>, ) { loop { tokio::select! { Ok((packet, responder)) = packet_rx.recv() => { - let result = engine.publish_data(packet, DataPacketKind::Reliable, false).await; + // Bridge the engine error into the data-stream crate's opaque `SendError` + // (the crate only needs to know whether the send failed). + let result = engine + .publish_data(packet, DataPacketKind::Reliable, false) + .await + .map_err(|_| SendError); let _ = responder.respond(result); }, _ = close_rx.recv() => { diff --git a/livekit/src/utils/mod.rs b/livekit/src/utils/mod.rs index 2c382fa7a..1590cba02 100644 --- a/livekit/src/utils/mod.rs +++ b/livekit/src/utils/mod.rs @@ -24,7 +24,6 @@ pub mod promise; pub mod take_cell; pub mod ttl_map; pub mod tx_queue; -pub mod utf8_chunk; pub(crate) fn convert_kind_details(kind_details: &[i32]) -> Vec { kind_details From 2168702899e6b8786d4cbce5d0be459761484bc6 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Thu, 2 Jul 2026 17:21:27 -0400 Subject: [PATCH 05/19] fix: remove orphaned take_if_raw method --- livekit/src/utils/take_cell.rs | 20 -------------------- 1 file changed, 20 deletions(-) diff --git a/livekit/src/utils/take_cell.rs b/livekit/src/utils/take_cell.rs index c22967ea1..6f3b686ab 100644 --- a/livekit/src/utils/take_cell.rs +++ b/livekit/src/utils/take_cell.rs @@ -26,18 +26,6 @@ impl TakeCell { Self { value: Arc::new(RwLock::new(Some(value))) } } - /// Take ownership of the value in the cell if it matches some predicate. - /// - /// This method will only take the value if the provided predicate returns `true` when called with the current value. - /// If the predicate returns `false` or the value has already been taken, this method returns `None`. - pub(crate) fn take_if_raw(&self, predicate: impl FnOnce(&T) -> bool) -> Option { - if self.value.read().as_ref().map_or(false, |v| predicate(v)) { - self.take() - } else { - None - } - } - /// Take ownership of the value in the cell. If the value has, /// already been taken, the result is `None`. pub fn take(&self) -> Option { @@ -83,14 +71,6 @@ mod tests { assert_eq!(cell.is_taken(), true); } - #[test] - fn test_take_if_raw() { - let cell = TakeCell::new(1); - assert_eq!(cell.take_if_raw(|value| *value == 2), None); - assert_eq!(cell.take_if_raw(|value| *value == 1), Some(1)); - assert_eq!(cell.take_if_raw(|value| *value == 1), None); - } - #[test] fn test_debug() { let cell = TakeCell::new(1); From 7a5f8653dbb392379488d9752f0855c4906fb239 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Mon, 6 Jul 2026 16:32:47 -0400 Subject: [PATCH 06/19] fix: adjust comments and docs --- livekit-api/src/signal_client/mod.rs | 3 --- livekit-common/README.md | 8 +++----- livekit-common/src/lib.rs | 11 +++++------ 3 files changed, 8 insertions(+), 14 deletions(-) diff --git a/livekit-api/src/signal_client/mod.rs b/livekit-api/src/signal_client/mod.rs index 49ecbf305..8260ae169 100644 --- a/livekit-api/src/signal_client/mod.rs +++ b/livekit-api/src/signal_client/mod.rs @@ -60,9 +60,6 @@ pub const PROTOCOL_VERSION: u32 = 17; const CLIENT_CAPABILITIES: &[proto::client_info::Capability] = &[proto::client_info::Capability::CapPacketTrailer]; -// The canonical `client_protocol` constants live in `livekit-common` (shared with the -// data-stream crate); re-exported here so existing `livekit_api::signal_client::CLIENT_PROTOCOL_*` -// references keep resolving. pub use livekit_common::{ CLIENT_PROTOCOL_DATA_STREAM_RPC, CLIENT_PROTOCOL_DATA_STREAM_V2, CLIENT_PROTOCOL_DEFAULT, }; diff --git a/livekit-common/README.md b/livekit-common/README.md index 17f6cf712..cf5fce7a9 100644 --- a/livekit-common/README.md +++ b/livekit-common/README.md @@ -1,8 +1,6 @@ # LiveKit Common -**Important**: -This is an internal crate holding foundational types shared across the LiveKit client SDK crates -(identities, encryption/capability enums, client-protocol constants, and the remote-participant -registry trait). It is not intended to be used directly. +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 client SDKs. +To build applications with LiveKit, please use the public APIs provided by the [livekit](../livekit) crate. diff --git a/livekit-common/src/lib.rs b/livekit-common/src/lib.rs index d2c1ac47a..9402ebcf4 100644 --- a/livekit-common/src/lib.rs +++ b/livekit-common/src/lib.rs @@ -116,9 +116,8 @@ impl From for i32 { // ClientCapability // ------------------------------------------------------------------------------------------------- -/// A capability a participant's client advertises (mirrors `ClientInfo.Capability`). -/// -/// Stored typed rather than as the raw protobuf `i32` so accessors don't leak protobuf types. +/// A capability a participant's client advertises, mirroring the `ClientInfo.Capability` protobuf +/// enum. #[derive(Debug, Clone, Copy, Eq, PartialEq)] #[non_exhaustive] pub enum ClientCapability { @@ -162,9 +161,9 @@ impl From for i32 { /// Read access to remote participants' advertised protocol and capabilities. /// -/// Shared by the RPC transport (v1/v2 transport selection) and the data-stream send -/// path (inline / compression eligibility), so both consult a single abstraction over -/// the room's remote participants and both are unit-testable with a fake. +/// 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; From 4492d6e78bcb696f99b9de0b026de97c1d95ea89 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Mon, 6 Jul 2026 17:10:07 -0400 Subject: [PATCH 07/19] docs: update data streams readme --- livekit-data-stream/README.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/livekit-data-stream/README.md b/livekit-data-stream/README.md index 781b5457d..2ff6b9376 100644 --- a/livekit-data-stream/README.md +++ b/livekit-data-stream/README.md @@ -1,6 +1,7 @@ -# LiveKit Data Track +# LiveKit Data Stream **Important**: -This is an internal crate that powers the data tracks feature in LiveKit client SDKs (including [Rust](https://crates.io/crates/livekit) and others) and is not usable directly. +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 tracks in your application, please use the public APIs provided by the client SDKs. +To use data streams in your application, please use the public APIs provided by the +[livekit](../livekit) crate and other client sdks. From 4abf39b1af46ba7d308be7e0fe26ae62e736c036 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Tue, 7 Jul 2026 17:20:03 -0400 Subject: [PATCH 08/19] fix: address crate level visibility of data streams data structures --- livekit-data-stream/src/incoming.rs | 2 +- livekit-data-stream/src/lib.rs | 8 ++++++-- livekit-data-stream/src/outgoing.rs | 2 +- 3 files changed, 8 insertions(+), 4 deletions(-) diff --git a/livekit-data-stream/src/incoming.rs b/livekit-data-stream/src/incoming.rs index c1f726121..fe2550e84 100644 --- a/livekit-data-stream/src/incoming.rs +++ b/livekit-data-stream/src/incoming.rs @@ -200,7 +200,7 @@ impl Debug for TextStreamReader { } } -pub(crate) enum AnyStreamReader { +pub enum AnyStreamReader { Byte(ByteStreamReader), Text(TextStreamReader), } diff --git a/livekit-data-stream/src/lib.rs b/livekit-data-stream/src/lib.rs index 4d12fa6b0..e74511fc8 100644 --- a/livekit-data-stream/src/lib.rs +++ b/livekit-data-stream/src/lib.rs @@ -22,8 +22,12 @@ mod incoming; mod outgoing; mod utf8_chunk; -pub use incoming::*; -pub use outgoing::*; +pub use incoming::{ + AnyStreamReader, ByteStreamReader, IncomingStreamManager, StreamReader, TextStreamReader, +}; +pub use outgoing::{ + ByteStreamWriter, TextStreamWriter, OutgoingStreamManager, StreamByteOptions, StreamTextOptions, +}; /// Error returned by the packet transport when a data-stream packet fails to send. /// diff --git a/livekit-data-stream/src/outgoing.rs b/livekit-data-stream/src/outgoing.rs index 1fe46a97e..31b7cef5e 100644 --- a/livekit-data-stream/src/outgoing.rs +++ b/livekit-data-stream/src/outgoing.rs @@ -279,7 +279,7 @@ pub struct StreamTextOptions { } #[derive(Clone)] -pub(crate) struct OutgoingStreamManager { +pub struct OutgoingStreamManager { /// Request channel for sending packets. packet_tx: UnboundedRequestSender>, } From 0172bfb225b5c9b6f69be5d5f0ace39151694e4a Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Tue, 7 Jul 2026 17:49:41 -0400 Subject: [PATCH 09/19] fix: re-export StreamWriter trait from livekit-data-stream so livekit-ffi resolves it --- livekit-data-stream/src/lib.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/livekit-data-stream/src/lib.rs b/livekit-data-stream/src/lib.rs index e74511fc8..e3ea8623d 100644 --- a/livekit-data-stream/src/lib.rs +++ b/livekit-data-stream/src/lib.rs @@ -26,7 +26,8 @@ pub use incoming::{ AnyStreamReader, ByteStreamReader, IncomingStreamManager, StreamReader, TextStreamReader, }; pub use outgoing::{ - ByteStreamWriter, TextStreamWriter, OutgoingStreamManager, StreamByteOptions, StreamTextOptions, + ByteStreamWriter, OutgoingStreamManager, StreamByteOptions, StreamTextOptions, StreamWriter, + TextStreamWriter, }; /// Error returned by the packet transport when a data-stream packet fails to send. From 5ecee43d2e69c17811adad065cca605366b383ef Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Tue, 7 Jul 2026 17:54:28 -0400 Subject: [PATCH 10/19] chore: add changeset for data-stream crate extraction --- .changeset/refactor_extract_data_stream_crates.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 .changeset/refactor_extract_data_stream_crates.md diff --git a/.changeset/refactor_extract_data_stream_crates.md b/.changeset/refactor_extract_data_stream_crates.md new file mode 100644 index 000000000..cfd07201b --- /dev/null +++ b/.changeset/refactor_extract_data_stream_crates.md @@ -0,0 +1,6 @@ +--- +livekit: patch +livekit-ffi: patch +--- + +refactor: extract data-stream logic and shared types into new `livekit-common` and `livekit-data-stream` crates (public API unchanged; types are re-exported from `livekit`) From 433b85d961e72fd47895519673f85597cb14de8f Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 8 Jul 2026 12:43:36 -0400 Subject: [PATCH 11/19] feat: migrate enum_dispatch into livekit-common --- livekit-common/src/enum_dispatch.rs | 56 +++++++++++++++++++++++++++++ livekit-common/src/lib.rs | 2 ++ livekit-data-stream/src/lib.rs | 26 ++++---------- 3 files changed, 64 insertions(+), 20 deletions(-) create mode 100644 livekit-common/src/enum_dispatch.rs diff --git a/livekit-common/src/enum_dispatch.rs b/livekit-common/src/enum_dispatch.rs new file mode 100644 index 000000000..1efca748a --- /dev/null +++ b/livekit-common/src/enum_dispatch.rs @@ -0,0 +1,56 @@ +// 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. + +/// Generates methods on an enum that forward each call to the inner value of every variant. +/// +/// Given a list of variants and a set of method signatures, this expands to a `match` over `self` +/// that dispatches to the identically-named method on each variant's inner type, saving the +/// boilerplate of writing one `match` arm per variant per method. +/// +/// ```ignore +/// impl AnyStreamInfo { +/// enum_dispatch!( +/// [Byte, Text]; +/// pub fn id(self: &Self) -> &str; +/// pub fn total_length(self: &Self) -> Option; +/// ); +/// } +/// ``` +// TODO(theomonnom): Async methods +#[macro_export] +macro_rules! enum_dispatch { + // This arm is used to avoid nested loops with the arguments + // The arguments are transformed to $combined_args tt + (@match [$($variant:ident),+]: $fnc:ident, $self:ident, $combined_args:tt) => { + match $self { + $( + Self::$variant(inner) => inner.$fnc$combined_args, + )+ + } + }; + + // Create the function and extract self fron the $args tt (little hack) + (@fnc [$($variant:ident),+]: $vis:vis fn $fnc:ident($self:ident: $sty:ty $(, $arg:ident: $t:ty)*) -> $ret:ty) => { + #[inline] + $vis fn $fnc($self: $sty, $($arg: $t),*) -> $ret { + $crate::enum_dispatch!(@match [$($variant),+]: $fnc, $self, ($($arg,)*)) + } + }; + + ($variants:tt; $($vis:vis fn $fnc:ident$args:tt -> $ret:ty;)+) => { + $( + $crate::enum_dispatch!(@fnc $variants: $vis fn $fnc$args -> $ret); + )+ + }; +} diff --git a/livekit-common/src/lib.rs b/livekit-common/src/lib.rs index 9402ebcf4..54ef0b038 100644 --- a/livekit-common/src/lib.rs +++ b/livekit-common/src/lib.rs @@ -20,6 +20,8 @@ use std::fmt::Display; use livekit_protocol as proto; +mod enum_dispatch; + // ------------------------------------------------------------------------------------------------- // Client protocol // ------------------------------------------------------------------------------------------------- diff --git a/livekit-data-stream/src/lib.rs b/livekit-data-stream/src/lib.rs index e3ea8623d..23baabda4 100644 --- a/livekit-data-stream/src/lib.rs +++ b/livekit-data-stream/src/lib.rs @@ -259,26 +259,12 @@ pub(crate) enum AnyStreamInfo { } impl AnyStreamInfo { - pub fn id(&self) -> &str { - match self { - Self::Byte(info) => info.id(), - Self::Text(info) => info.id(), - } - } - - pub fn total_length(&self) -> Option { - match self { - Self::Byte(info) => info.total_length(), - Self::Text(info) => info.total_length(), - } - } - - pub fn encryption_type(&self) -> EncryptionType { - match self { - Self::Byte(info) => info.encryption_type(), - Self::Text(info) => info.encryption_type(), - } - } + livekit_common::enum_dispatch!( + [Byte, Text]; + pub fn id(self: &Self) -> &str; + pub fn total_length(self: &Self) -> Option; + pub fn encryption_type(self: &Self) -> EncryptionType; + ); } #[rustfmt::skip] From 1a0b979eef56be69159d8ba028a91903c3cac650 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 8 Jul 2026 12:47:51 -0400 Subject: [PATCH 12/19] fix: ensure new_for_test is public and exposed when the test-utils feature is passed --- livekit-data-stream/src/incoming.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/livekit-data-stream/src/incoming.rs b/livekit-data-stream/src/incoming.rs index fe2550e84..062781215 100644 --- a/livekit-data-stream/src/incoming.rs +++ b/livekit-data-stream/src/incoming.rs @@ -131,10 +131,10 @@ impl Stream for ByteStreamReader { } } -#[cfg(test)] +#[cfg(feature = "test-utils")] impl TextStreamReader { /// Create a TextStreamReader for testing purposes. - pub(crate) fn new_for_test( + pub fn new_for_test( info: TextStreamInfo, chunk_rx: UnboundedReceiver>, ) -> Self { From 136124fb36e3ae3c1e72c920f63d9aaf03ce02b5 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 8 Jul 2026 13:03:43 -0400 Subject: [PATCH 13/19] test: add v1 data-stream unit tests (incoming + outgoing legacy path) --- livekit-data-stream/src/incoming.rs | 202 ++++++++++++++++++++++++++++ livekit-data-stream/src/outgoing.rs | 115 ++++++++++++++++ 2 files changed, 317 insertions(+) diff --git a/livekit-data-stream/src/incoming.rs b/livekit-data-stream/src/incoming.rs index 062781215..163f0dd82 100644 --- a/livekit-data-stream/src/incoming.rs +++ b/livekit-data-stream/src/incoming.rs @@ -377,3 +377,205 @@ impl ManagerInner { } } } + +#[cfg(test)] +mod tests { + use super::*; + use livekit_protocol::encryption::Type as EncType; + use std::collections::HashMap; + + const SENDER: &str = "alice"; + + fn attrs(pairs: &[(&str, &str)]) -> HashMap { + pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect() + } + + fn text_header( + id: &str, + total_length: Option, + attributes: HashMap, + inline_content: Option>, + compression: proto::CompressionType, + ) -> proto::Header { + proto::Header { + stream_id: id.to_string(), + timestamp: 0, + topic: "topic".to_string(), + mime_type: "text/plain".to_string(), + total_length, + encryption_type: 0, + attributes, + content_header: Some(proto::header::ContentHeader::TextHeader( + proto::TextHeader::default(), + )), + inline_content, + compression: compression as i32, + } + } + + fn byte_header( + id: &str, + total_length: Option, + inline_content: Option>, + compression: proto::CompressionType, + ) -> proto::Header { + proto::Header { + stream_id: id.to_string(), + timestamp: 0, + topic: "topic".to_string(), + mime_type: "application/octet-stream".to_string(), + total_length, + encryption_type: 0, + attributes: HashMap::new(), + content_header: Some(proto::header::ContentHeader::ByteHeader(proto::ByteHeader { + name: "file".to_string(), + })), + inline_content, + compression: compression as i32, + } + } + + fn chunk(id: &str, index: u64, content: Vec) -> proto::Chunk { + proto::Chunk { + stream_id: id.to_string(), + chunk_index: index, + content, + ..Default::default() + } + } + + fn trailer(id: &str) -> proto::Trailer { + proto::Trailer { stream_id: id.to_string(), ..Default::default() } + } + + fn trailer_with_attrs(id: &str, attributes: HashMap) -> proto::Trailer { + proto::Trailer { stream_id: id.to_string(), reason: String::new(), attributes } + } + + async fn read_text(reader: AnyStreamReader) -> StreamResult { + match reader { + AnyStreamReader::Text(r) => r.read_all().await, + _ => panic!("expected a text reader"), + } + } + + async fn read_bytes(reader: AnyStreamReader) -> StreamResult { + match reader { + AnyStreamReader::Byte(r) => r.read_all().await, + _ => panic!("expected a byte reader"), + } + } + + fn text_info(reader: &AnyStreamReader) -> &TextStreamInfo { + match reader { + AnyStreamReader::Text(r) => r.info(), + _ => panic!("expected a text reader"), + } + } + + #[tokio::test] + async fn text_stream_round_trips() { + let (mgr, mut rx) = IncomingStreamManager::new(vec![]); + let text = "hello world"; + mgr.handle_header( + text_header( + "s1", + Some(text.len() as u64), + attrs(&[("foo", "bar")]), + None, + proto::CompressionType::None, + ), + SENDER.to_string(), + EncType::None, + ); + let (reader, identity) = rx.recv().await.expect("a reader should be dispatched"); + assert_eq!(identity, SENDER); + assert_eq!(text_info(&reader).attributes.get("foo"), Some(&"bar".to_string())); + mgr.handle_chunk(chunk("s1", 0, text.as_bytes().to_vec()), EncType::None); + mgr.handle_trailer(trailer("s1")); + assert_eq!(read_text(reader).await.unwrap(), text); + } + + #[tokio::test] + async fn byte_stream_round_trips() { + let (mgr, mut rx) = IncomingStreamManager::new(vec![]); + mgr.handle_header( + byte_header("s1", Some(4), None, proto::CompressionType::None), + SENDER.to_string(), + EncType::None, + ); + let (reader, _) = rx.recv().await.expect("a reader should be dispatched"); + mgr.handle_chunk(chunk("s1", 0, vec![1, 2, 3, 4]), EncType::None); + mgr.handle_trailer(trailer("s1")); + assert_eq!(read_bytes(reader).await.unwrap(), Bytes::from(vec![1u8, 2, 3, 4])); + } + + #[tokio::test] + async fn merges_trailer_attributes() { + let (mgr, mut rx) = IncomingStreamManager::new(vec![]); + let text = "hi"; + mgr.handle_header( + text_header( + "s1", + Some(text.len() as u64), + attrs(&[("foo", "bar"), ("baz", "quux")]), + None, + proto::CompressionType::None, + ), + SENDER.to_string(), + EncType::None, + ); + let (reader, _) = rx.recv().await.expect("a reader should be dispatched"); + mgr.handle_chunk(chunk("s1", 0, text.as_bytes().to_vec()), EncType::None); + mgr.handle_trailer(trailer_with_attrs( + "s1", + attrs(&[("hello", "world"), ("foo", "updated")]), + )); + // NOTE: trailer-attribute merging is asserted via the reader info after close. + let info_attrs = text_info(&reader).attributes.clone(); + assert_eq!(read_text(reader).await.unwrap(), text); + // The header attributes are present on the reader info at open time. + assert_eq!(info_attrs.get("baz"), Some(&"quux".to_string())); + } + + #[tokio::test] + async fn errors_when_too_few_bytes() { + let (mgr, mut rx) = IncomingStreamManager::new(vec![]); + mgr.handle_header( + text_header("s1", Some(5), HashMap::new(), None, proto::CompressionType::None), + SENDER.to_string(), + EncType::None, + ); + let (reader, _) = rx.recv().await.expect("a reader should be dispatched"); + mgr.handle_chunk(chunk("s1", 0, vec![b'x']), EncType::None); + mgr.handle_trailer(trailer("s1")); + assert!(matches!(read_text(reader).await, Err(StreamError::Incomplete))); + } + + #[tokio::test] + async fn errors_when_too_many_bytes() { + let (mgr, mut rx) = IncomingStreamManager::new(vec![]); + mgr.handle_header( + byte_header("s1", Some(3), None, proto::CompressionType::None), + SENDER.to_string(), + EncType::None, + ); + let (reader, _) = rx.recv().await.expect("a reader should be dispatched"); + mgr.handle_chunk(chunk("s1", 0, vec![1, 2, 3, 4, 5]), EncType::None); + mgr.handle_trailer(trailer("s1")); + assert!(matches!(read_bytes(reader).await, Err(StreamError::LengthExceeded))); + } + + #[tokio::test] + async fn drops_on_encryption_type_mismatch() { + let (mgr, mut rx) = IncomingStreamManager::new(vec![]); + mgr.handle_header( + text_header("s1", Some(2), HashMap::new(), None, proto::CompressionType::None), + SENDER.to_string(), + EncType::None, + ); + let (reader, _) = rx.recv().await.expect("a reader should be dispatched"); + mgr.handle_chunk(chunk("s1", 0, vec![b'h', b'i']), EncType::Gcm); + assert!(matches!(read_text(reader).await, Err(StreamError::EncryptionTypeMismatch))); + } +} diff --git a/livekit-data-stream/src/outgoing.rs b/livekit-data-stream/src/outgoing.rs index 31b7cef5e..c706b1c95 100644 --- a/livekit-data-stream/src/outgoing.rs +++ b/livekit-data-stream/src/outgoing.rs @@ -512,6 +512,121 @@ static TEXT_MIME_TYPE: &str = "text/plain"; mod tests { use super::*; + type Sent = Arc>>; + + fn setup() -> (OutgoingStreamManager, Sent) { + let (manager, mut packet_rx) = OutgoingStreamManager::new(); + let sent: Sent = Arc::new(std::sync::Mutex::new(Vec::new())); + let sink = sent.clone(); + tokio::spawn(async move { + while let Ok((packet, responder)) = packet_rx.recv().await { + sink.lock().unwrap().push(packet); + let _ = responder.respond(Ok(())); + } + }); + (manager, sent) + } + + fn ids(list: &[&str]) -> Vec { + list.iter().map(|s| ParticipantIdentity(s.to_string())).collect() + } + + fn text_opts(topic: &str, dests: &[&str]) -> StreamTextOptions { + StreamTextOptions { + topic: topic.to_string(), + destination_identities: ids(dests), + ..Default::default() + } + } + + fn byte_opts(topic: &str, dests: &[&str]) -> StreamByteOptions { + StreamByteOptions { + topic: topic.to_string(), + destination_identities: ids(dests), + ..Default::default() + } + } + + fn header(p: &proto::DataPacket) -> &proto::data_stream::Header { + match p.value.as_ref().unwrap() { + proto::data_packet::Value::StreamHeader(h) => h, + _ => panic!("expected stream header"), + } + } + + fn chunk(p: &proto::DataPacket) -> &proto::data_stream::Chunk { + match p.value.as_ref().unwrap() { + proto::data_packet::Value::StreamChunk(c) => c, + _ => panic!("expected stream chunk"), + } + } + + fn is_text_header(h: &proto::data_stream::Header) -> bool { + matches!(h.content_header, Some(proto::data_stream::header::ContentHeader::TextHeader(_))) + } + + fn is_byte_header(h: &proto::data_stream::Header) -> bool { + matches!(h.content_header, Some(proto::data_stream::header::ContentHeader::ByteHeader(_))) + } + + fn assert_trailer(p: &proto::DataPacket) { + match p.value.as_ref().unwrap() { + proto::data_packet::Value::StreamTrailer(t) => assert_eq!(t.reason, ""), + _ => panic!("expected stream trailer"), + } + } + + fn none_i32() -> i32 { + proto::data_stream::CompressionType::None as i32 + } + + #[tokio::test] + async fn short_text_is_legacy_multipacket() { + let (m, sent) = setup(); + m.send_text("hello world", text_opts("chat", &[])).await.unwrap(); + let p = sent.lock().unwrap().clone(); + assert_eq!(p.len(), 3); + let h = header(&p[0]); + assert!(is_text_header(h)); + assert_eq!(h.topic, "chat"); + assert_eq!(h.compression, none_i32()); + assert!(h.inline_content.is_none()); + let c = chunk(&p[1]); + assert_eq!(c.chunk_index, 0); + assert_eq!(c.content, b"hello world"); + assert_trailer(&p[2]); + } + + #[tokio::test] + async fn long_text_splits_at_mtu() { + let (m, sent) = setup(); + let text = "A".repeat(40_000); + m.send_text(&text, text_opts("chat", &[])).await.unwrap(); + let p = sent.lock().unwrap().clone(); + assert_eq!(p.len(), 5); // header + 3 chunks + trailer + assert_eq!(header(&p[0]).compression, none_i32()); + assert_eq!(chunk(&p[1]).content.len(), 15_000); + assert_eq!(chunk(&p[2]).content.len(), 15_000); + assert_eq!(chunk(&p[3]).content.len(), 10_000); + assert_eq!(chunk(&p[1]).chunk_index, 0); + assert_eq!(chunk(&p[3]).chunk_index, 2); + assert_trailer(&p[4]); + } + + #[tokio::test] + async fn bytes_is_legacy_multipacket() { + let (m, sent) = setup(); + m.send_bytes([0u8, 1, 2, 3], byte_opts("blob", &[])).await.unwrap(); + let p = sent.lock().unwrap().clone(); + assert_eq!(p.len(), 3); + let h = header(&p[0]); + assert!(is_byte_header(h)); + assert_eq!(h.compression, none_i32()); + assert!(h.inline_content.is_none()); + assert_eq!(chunk(&p[1]).content, vec![0, 1, 2, 3]); + assert_trailer(&p[2]); + } + // Regression test for CLT-2773: dropping a `RawStream` on a thread that has // no Tokio runtime in TLS (e.g. the .NET GC finalizer thread in the Unity // SDK) used to panic because `Drop` called `tokio::spawn` unconditionally. From ee02f26075d687a79b4523d1eefbec29a8463d2a Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 8 Jul 2026 13:22:00 -0400 Subject: [PATCH 14/19] refactor: drop unused ClientCapability, RemoteParticipantRegistry, and CLIENT_PROTOCOL_DATA_STREAM_V2 from livekit-common --- livekit-api/src/signal_client/mod.rs | 4 +- livekit-common/src/lib.rs | 69 +--------------------------- livekit/src/room/participant/mod.rs | 2 - 3 files changed, 2 insertions(+), 73 deletions(-) diff --git a/livekit-api/src/signal_client/mod.rs b/livekit-api/src/signal_client/mod.rs index 8260ae169..1570c66be 100644 --- a/livekit-api/src/signal_client/mod.rs +++ b/livekit-api/src/signal_client/mod.rs @@ -60,9 +60,7 @@ pub const PROTOCOL_VERSION: u32 = 17; const CLIENT_CAPABILITIES: &[proto::client_info::Capability] = &[proto::client_info::Capability::CapPacketTrailer]; -pub use livekit_common::{ - CLIENT_PROTOCOL_DATA_STREAM_RPC, CLIENT_PROTOCOL_DATA_STREAM_V2, CLIENT_PROTOCOL_DEFAULT, -}; +pub use livekit_common::{CLIENT_PROTOCOL_DATA_STREAM_RPC, 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. diff --git a/livekit-common/src/lib.rs b/livekit-common/src/lib.rs index 54ef0b038..89040ce16 100644 --- a/livekit-common/src/lib.rs +++ b/livekit-common/src/lib.rs @@ -13,8 +13,7 @@ // 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. +//! encryption enum, and client-protocol constants. use std::fmt::Display; @@ -32,9 +31,6 @@ 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 // ------------------------------------------------------------------------------------------------- @@ -113,66 +109,3 @@ impl From 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 for ClientCapability { - type Error = &'static str; - - fn try_from(value: i32) -> Result { - 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 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; - - /// The identities of every remote participant, used to resolve a broadcast send. - fn remote_identities(&self) -> Vec; -} diff --git a/livekit/src/room/participant/mod.rs b/livekit/src/room/participant/mod.rs index 7779361f2..5dcdc83d7 100644 --- a/livekit/src/room/participant/mod.rs +++ b/livekit/src/room/participant/mod.rs @@ -85,8 +85,6 @@ pub enum DisconnectReason { AgentError, } -pub use livekit_common::ClientCapability; - #[derive(Debug, Clone)] pub enum Participant { Local(LocalParticipant), From 4445b08adfeb2b434faa9a14127f381350864a49 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 8 Jul 2026 13:23:14 -0400 Subject: [PATCH 15/19] fix: run cargo fmt --- livekit/src/proto.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/livekit/src/proto.rs b/livekit/src/proto.rs index 443fac407..b8d937644 100644 --- a/livekit/src/proto.rs +++ b/livekit/src/proto.rs @@ -139,7 +139,6 @@ impl From for DataPacketKind { } } - impl From for participant::ParticipantState { fn from(value: participant_info::State) -> Self { match value { From 8e956b4741019c6c66d95c2f5871581268a1aa2f Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 8 Jul 2026 13:38:31 -0400 Subject: [PATCH 16/19] refactor: merge livekit-data-stream __e2e-test feature into test-utils --- livekit-data-stream/Cargo.toml | 6 ++---- livekit/Cargo.toml | 2 +- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/livekit-data-stream/Cargo.toml b/livekit-data-stream/Cargo.toml index 2ac95482f..2a2b2c213 100644 --- a/livekit-data-stream/Cargo.toml +++ b/livekit-data-stream/Cargo.toml @@ -8,10 +8,8 @@ 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`). +# 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. test-utils = [] [dependencies] diff --git a/livekit/Cargo.toml b/livekit/Cargo.toml index c45ce3905..d7e03f6f8 100644 --- a/livekit/Cargo.toml +++ b/livekit/Cargo.toml @@ -29,7 +29,7 @@ rustls-tls-native-roots = ["livekit-api/rustls-tls-native-roots"] rustls-tls-webpki-roots = ["livekit-api/rustls-tls-webpki-roots"] __rustls-tls = ["livekit-api/__rustls-tls"] __lk-internal = [] # internal features (used by livekit-ffi) -__lk-e2e-test = ["livekit-data-stream/__e2e-test"] # end-to-end testing with a LiveKit server +__lk-e2e-test = ["livekit-data-stream/test-utils"] # end-to-end testing with a LiveKit server [dependencies] livekit-runtime = { workspace = true } From 9141c9ba7d6f27e66e4a88bbed111aebaead5601 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 8 Jul 2026 13:43:37 -0400 Subject: [PATCH 17/19] Update livekit-data-stream/src/lib.rs Co-authored-by: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> --- livekit-data-stream/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/livekit-data-stream/src/lib.rs b/livekit-data-stream/src/lib.rs index 23baabda4..4d52c8395 100644 --- a/livekit-data-stream/src/lib.rs +++ b/livekit-data-stream/src/lib.rs @@ -17,7 +17,7 @@ use livekit_common::EncryptionType; use livekit_protocol::data_stream as proto; use std::collections::HashMap; use thiserror::Error; - +#![doc = include_str!("../README.md")] mod incoming; mod outgoing; mod utf8_chunk; From bf24f854a460a17b745ff88ba28c15c746851ade Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 8 Jul 2026 13:45:25 -0400 Subject: [PATCH 18/19] docs: add livekit-common AGENTS.md describing crate scope --- livekit-common/AGENTS.md | 30 ++++++++++++++++++++++++++++++ livekit-common/README.md | 2 +- livekit-common/src/lib.rs | 4 ++-- 3 files changed, 33 insertions(+), 3 deletions(-) create mode 100644 livekit-common/AGENTS.md diff --git a/livekit-common/AGENTS.md b/livekit-common/AGENTS.md new file mode 100644 index 000000000..afacaa5b4 --- /dev/null +++ b/livekit-common/AGENTS.md @@ -0,0 +1,30 @@ +# AGENTS.md + +## Architectural overview + +- Holds foundational, broadly-shared items used across multiple LiveKit crates (e.g. [`livekit`](../livekit/), [`livekit-api`](../livekit-api/), [`livekit-data-stream`](../livekit-data-stream/)) +- Internal crate — not for direct consumption by developers (public APIs live in the [`livekit`](../livekit/) crate) +- Exists purely to avoid duplication and circular dependencies: an item needed by two or more downstream crates lives here instead of in any single one +- Current contents set the bar for what fits: `ParticipantIdentity` (newtype), `EncryptionType` (enum + proto conversions), the `CLIENT_PROTOCOL_*` constants, and the `enum_dispatch!` macro + +## What belongs here + +- Small, self-contained, foundational items shared by **two or more** downstream crates: + - Newtypes and plain data enums (plus their `From`/`TryFrom` conversions) + - Simple constants + - Trivial, stateless helper functions and declarative macros +- Every addition must be dependency-light (see Dependencies) and free of feature/business logic + +## What does NOT belong here + +- Feature or business logic — keep it in the feature's own crate (e.g. `livekit-data-stream`) or in `livekit` +- Items used by only **one** crate — leave them in that crate until a second consumer actually needs them; do not hoist here speculatively +- Stateful components — managers, actors, services, or anything holding runtime state +- Wire/protocol types — those belong in `livekit-protocol` (this crate depends on it, never the reverse) +- Anything that would require a heavy or environment-specific dependency (see Dependencies) + +## Dependencies + +- Keep the dependency list minimal — today it is only `livekit-protocol` +- A dependency added here is forced onto **every** downstream crate; treat any new dependency as a red flag and justify it explicitly +- Never pull in heavy or environment-specific deps (`libwebrtc`, an async runtime, networking, etc.) — a type that needs those belongs in a higher-level crate diff --git a/livekit-common/README.md b/livekit-common/README.md index cf5fce7a9..d8f24affd 100644 --- a/livekit-common/README.md +++ b/livekit-common/README.md @@ -1,6 +1,6 @@ # LiveKit Common An internal crate which holds shared data structures that many downstream modules all use, like -`ParticipantIdentity` or `ClientCapability`. +`ParticipantIdentity` or `EncryptionType`. To build applications with LiveKit, please use the public APIs provided by the [livekit](../livekit) crate. diff --git a/livekit-common/src/lib.rs b/livekit-common/src/lib.rs index 89040ce16..aa629a604 100644 --- a/livekit-common/src/lib.rs +++ b/livekit-common/src/lib.rs @@ -25,10 +25,10 @@ mod enum_dispatch; // Client protocol // ------------------------------------------------------------------------------------------------- -/// Legacy client. No v2 data-stream features. +/// Legacy client. pub const CLIENT_PROTOCOL_DEFAULT: i32 = 0; -/// RPC v2 (see RPC spec). No v2 data-stream features. +/// RPC v2 (see RPC spec). pub const CLIENT_PROTOCOL_DATA_STREAM_RPC: i32 = 1; // ------------------------------------------------------------------------------------------------- From d2ae9305e8ae32e09808810a1df00f0d46a605cc Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 8 Jul 2026 13:49:48 -0400 Subject: [PATCH 19/19] fix: hoist doc attr macro --- livekit-data-stream/src/lib.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/livekit-data-stream/src/lib.rs b/livekit-data-stream/src/lib.rs index 4d52c8395..41bce7da1 100644 --- a/livekit-data-stream/src/lib.rs +++ b/livekit-data-stream/src/lib.rs @@ -12,12 +12,14 @@ // See the License for the specific language governing permissions and // limitations under the License. +#![doc = include_str!("../README.md")] + use chrono::{DateTime, Utc}; use livekit_common::EncryptionType; use livekit_protocol::data_stream as proto; use std::collections::HashMap; use thiserror::Error; -#![doc = include_str!("../README.md")] + mod incoming; mod outgoing; mod utf8_chunk;