Data streams v2

.changeset/data_streams_v2.md

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

examples/basic_text_stream/src/main.rs

@@ -28,7 +28,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 async fn run_sender(room: Room) -> Result<(), Box<dyn Error>> {
     println!("Running as sender");
     loop {
-        let options = StreamTextOptions { topic: TOPIC.to_string(), ..Default::default() };
+        let options = StreamTextOptions::new_with_topic(TOPIC);
         let writer = room.local_participant().stream_text(options).await?;
         println!("Opened new stream");
 

examples/encrypted_text_stream/src/main.rs

@@ -86,10 +86,7 @@ async fn run_interactive_chat(
                         }
 
                         // Send the message
-                        let options = StreamTextOptions {
-                            topic: "lk.chat".to_string(),
-                            ..Default::default()
-                        };
+                        let options = StreamTextOptions::new_with_topic("lk.chat");
 
                         match room.local_participant().send_text(&input, options).await {
                             Ok(_) => {

examples/send_bytes/src/main.rs

@@ -43,7 +43,7 @@ async fn run_sender(room: Room) -> Result<(), Box<dyn Error>> {
 
         println!("[tx] {}", packet);
 
-        let options = StreamByteOptions { topic: LED_CONTROL_TOPIC.into(), ..Default::default() };
+        let options = StreamByteOptions::new_with_topic(LED_CONTROL_TOPIC);
         let be_bytes = packet.into_bits().to_be_bytes();
         room.local_participant().send_bytes(&be_bytes, options).await?;
 

livekit-api/src/signal_client/mod.rs

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

livekit-common/src/lib.rs

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

livekit-data-stream/AGENTS.md

@@ -0,0 +1,44 @@
+# AGENTS.md
+
+## Architectural overview
+
+- This crate encapsulates the business logic and public APIs for the data streams feature: [text streams](https://docs.livekit.io/transport/data/text-streams/) and [byte streams](https://docs.livekit.io/transport/data/byte-streams/)
+- Not for direct consumption by developers
+- Unlike most SDK features which live directly in the [`livekit`](../livekit/) crate, data streams are intentionally isolated here for several reasons:
+  - Enforces decoupling from other components (e.g., data channel, signaling client, etc.)
+  - Enables proper integration testing
+  - Enables shared implementation amongst multiple _consumers_:
+    - [`livekit`](../livekit/): Rust client SDK
+    - [`livekit-uniffi`](../livekit-uniffi/): will eventually power downstream client SDKs such as Swift and Kotlin
+
+## Incoming vs. outgoing split
+
+- The crate is organized into two halves:
+  - `incoming/`: receiving streams from remote participants (produces readers)
+  - `outgoing/`: sending streams to remote participants (produces writers)
+- The two halves never communicate with each other or share state
+- Shared types live at the crate root rather than inside either side:
+  - wire/domain packet types (`Header`, `Chunk`, `Trailer`, `Packet`, `StreamId`, ...) in `types/`
+  - `ByteStreamInfo` / `TextStreamInfo` in `info.rs`
+  - `StreamError`, `StreamResult`, `StreamProgress`, `SendError` in `utils.rs`
+  - helpers such as UTF-8-aware chunking in `utf8_chunk.rs`
+
+### The halves are deliberately *not* symmetric
+
+Unlike the mirror-image `local/`/`remote/` split in [`livekit-datatrack`](../livekit-datatrack/AGENTS.md), `incoming/` and `outgoing/` do **not** share a parallel shape. Each has a `manager`, but they are built differently on purpose:
+
+- The **incoming** manager (`incoming/manager.rs`) is an **actor**. It owns all receive-side state on a single task and is driven by `InputEvent`s fed over a channel, emitting `OutputEvent`s for the host to surface (see `incoming/events.rs`). This is required because inbound packets arrive from the engine's event loop in a context that cannot `.await` into the manager, inbound chunks must never be dropped (a dropped chunk is an unrecoverable `MissedChunk`), and processing must not head-of-line-block that loop. Owning its state directly also lets its handlers `.await` decompression on the run-loop task without holding a lock across the await point.
+- The **outgoing** manager (`outgoing/manager.rs`) is a plain struct with `async` methods that callers `.await` directly. It runs in an already-async context (publishing), so awaiting into it is fine and the actor machinery (events, channels, a run loop) would add substantial complexity for no benefit.
+
+This asymmetry is a conscious choice, **not** an oversight or unfinished work. Do not "symmetrize" the outgoing side into an actor preemptively. When changing behavior on one side, there is often no mirror to update on the other — verify rather than assume.
+
+## Boundaries
+
+- Two public modules get exported from this crate (see `lib.rs`):
+  1. `api`: public APIs that get re-exported by _consumers_ and made available to developers (readers, writers, stream options, stream infos, `StreamError`, ...)
+  2. `backend`: managers and supporting wire types used internally by _consumers_ to power the feature (`backend::incoming`, `backend::outgoing`, and the domain packet types)
+- The incoming actor's events are decoupled from protocol messages for several reasons:
+  - Protobuf is a wire format and cannot express Rust-level invariants
+  - Events can carry in-process types proto cannot (e.g., the channel senders used to feed a reader)
+  - Allows the protocol to evolve independently
+- Wire <-> domain conversion lives in `types/packet.rs`, which owns the `From`/`TryFrom` impls between `livekit_protocol::data_stream` messages and this crate's own packet types. Consumers convert inbound proto into domain `Packet`s before feeding the incoming actor, so the incoming side only ever sees domain types.

livekit-data-stream/Cargo.toml

@@ -8,9 +8,7 @@ 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 = []
+test-utils = ["dep:rand"]
 
 [dependencies]
 livekit-common = { workspace = true }
@@ -20,12 +18,16 @@ 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"] }
+tokio-stream = { workspace = true, features = ["sync"] }
+futures-util = { workspace = true, default-features = false, features = ["sink", "io"] }
 prost = "0.12"
 chrono = "0.4.38"
-flate2 = "1"
 bmrng = "0.5.2"
 uuid = { version = "1", features = ["v4"] }
+async-compression = { version = "0.4.42", features = ["deflate", "futures-io"] }
+from_variants.workspace = true
+rand = { workspace = true, optional = true }
 
 [dev-dependencies]
 tokio = { workspace = true, default-features = false, features = ["macros", "rt", "rt-multi-thread", "time"] }
+rand = { workspace = true }