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`) diff --git a/Cargo.lock b/Cargo.lock index 388960a99..16498e82f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2656,11 +2656,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" @@ -3919,6 +3930,8 @@ dependencies = [ "libloading 0.8.9", "libwebrtc", "livekit-api", + "livekit-common", + "livekit-data-stream", "livekit-datatrack", "livekit-protocol", "livekit-runtime", @@ -3951,6 +3964,7 @@ dependencies = [ "http 1.4.0", "isahc", "jsonwebtoken", + "livekit-common", "livekit-protocol", "livekit-runtime", "log", @@ -3973,6 +3987,32 @@ dependencies = [ "url", ] +[[package]] +name = "livekit-common" +version = "0.1.0" +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.10" @@ -5936,6 +5976,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" @@ -8131,7 +8177,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 d31c33d06..ad55c34b0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,8 @@ members = [ "livekit", "livekit-api", "livekit-protocol", + "livekit-common", + "livekit-data-stream", "livekit-ffi", "livekit-uniffi", "livekit-datatrack", @@ -50,6 +52,8 @@ livekit = { version = "0.7.51", path = "livekit" } livekit-api = { version = "0.5.5", path = "livekit-api" } livekit-ffi = { version = "0.12.69", path = "livekit-ffi" } livekit-datatrack = { version = "0.1.10", 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" } # default-features off so each consumer selects its runtime explicitly # (livekit-runtime/tokio | /async | /dispatcher); otherwise the default `tokio` diff --git a/livekit-api/Cargo.toml b/livekit-api/Cargo.toml index baca4a6b8..0b9e02f99 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 8d3e20a9e..36d0870c0 100644 --- a/livekit-api/src/signal_client/mod.rs +++ b/livekit-api/src/signal_client/mod.rs @@ -60,11 +60,7 @@ 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; +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/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/Cargo.toml b/livekit-common/Cargo.toml new file mode 100644 index 000000000..c3ec0491d --- /dev/null +++ b/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 } diff --git a/livekit-common/README.md b/livekit-common/README.md new file mode 100644 index 000000000..d8f24affd --- /dev/null +++ b/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 `EncryptionType`. + +To build applications with LiveKit, please use the public APIs provided by the [livekit](../livekit) crate. 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 new file mode 100644 index 000000000..aa629a604 --- /dev/null +++ b/livekit-common/src/lib.rs @@ -0,0 +1,111 @@ +// 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 enum, and client-protocol constants. + +use std::fmt::Display; + +use livekit_protocol as proto; + +mod enum_dispatch; + +// ------------------------------------------------------------------------------------------------- +// Client protocol +// ------------------------------------------------------------------------------------------------- + +/// Legacy client. +pub const CLIENT_PROTOCOL_DEFAULT: i32 = 0; + +/// RPC v2 (see RPC spec). +pub const CLIENT_PROTOCOL_DATA_STREAM_RPC: i32 = 1; + +// ------------------------------------------------------------------------------------------------- +// ParticipantIdentity +// ------------------------------------------------------------------------------------------------- + +#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)] +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)] +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, + } + } +} diff --git a/livekit-data-stream/Cargo.toml b/livekit-data-stream/Cargo.toml new file mode 100644 index 000000000..2a2b2c213 --- /dev/null +++ b/livekit-data-stream/Cargo.toml @@ -0,0 +1,31 @@ +[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] +# 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] +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..2ff6b9376 --- /dev/null +++ b/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. diff --git a/livekit/src/room/data_stream/incoming.rs b/livekit-data-stream/src/incoming.rs similarity index 59% rename from livekit/src/room/data_stream/incoming.rs rename to livekit-data-stream/src/incoming.rs index 213a68c07..163f0dd82 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, @@ -148,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 { @@ -217,7 +200,7 @@ impl Debug for TextStreamReader { } } -pub(crate) enum AnyStreamReader { +pub enum AnyStreamReader { Byte(ByteStreamReader), Text(TextStreamReader), } @@ -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 { @@ -382,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/src/room/data_stream/mod.rs b/livekit-data-stream/src/lib.rs similarity index 90% rename from livekit/src/room/data_stream/mod.rs rename to livekit-data-stream/src/lib.rs index 1be6cce7b..41bce7da1 100644 --- a/livekit/src/room/data_stream/mod.rs +++ b/livekit-data-stream/src/lib.rs @@ -12,28 +12,37 @@ // See the License for the specific language governing permissions and // limitations under the License. +#![doc = include_str!("../README.md")] + 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; - -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) +mod utf8_chunk; + +pub use incoming::{ + AnyStreamReader, ByteStreamReader, IncomingStreamManager, StreamReader, TextStreamReader, +}; +pub use outgoing::{ + ByteStreamWriter, OutgoingStreamManager, StreamByteOptions, StreamTextOptions, StreamWriter, + TextStreamWriter, +}; + +/// 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,7 +261,7 @@ pub(crate) enum AnyStreamInfo { } impl AnyStreamInfo { - enum_dispatch!( + livekit_common::enum_dispatch!( [Byte, Text]; pub fn id(self: &Self) -> &str; pub fn total_length(self: &Self) -> Option; diff --git a/livekit/src/room/data_stream/outgoing.rs b/livekit-data-stream/src/outgoing.rs similarity index 82% rename from livekit/src/room/data_stream/outgoing.rs rename to livekit-data-stream/src/outgoing.rs index 5c3e4c431..c706b1c95 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) @@ -280,13 +279,13 @@ pub struct StreamTextOptions { } #[derive(Clone)] -pub(crate) struct OutgoingStreamManager { +pub 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) @@ -513,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. @@ -522,7 +636,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 8dcd216b4..a9dd5da19 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/test-utils"] # 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 78e690d73..b8d937644 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 { @@ -141,36 +139,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 { match value { diff --git a/livekit/src/room/e2ee/mod.rs b/livekit/src/room/e2ee/mod.rs index e1235d81d..43d4a908c 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 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/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 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);