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/wgpu_room/src/app.rs

@@ -1,4 +1,5 @@
 use crate::{
+    data_streams_ui::DataStreamsUiState,
     data_track::{LocalDataTrackTile, RemoteDataTrackTile, MAX_VALUE, TIME_WINDOW},
     rpc_ui::RpcUiState,
     service::{AsyncCmd, LkService, UiCmd},
@@ -25,6 +26,7 @@ struct AppState {
 enum RightTab {
     Participants,
     Rpc,
+    DataStreams,
 }
 
 pub struct LkApp {
@@ -38,6 +40,7 @@ pub struct LkApp {
     render_state: egui_wgpu::RenderState,
     service: LkService,
     rpc_ui: RpcUiState,
+    data_streams_ui: DataStreamsUiState,
     right_tab: RightTab,
 }
 
@@ -74,6 +77,7 @@ impl LkApp {
             connection_failure: None,
             render_state: cc.wgpu_render_state.clone().unwrap(),
             rpc_ui: RpcUiState::default(),
+            data_streams_ui: DataStreamsUiState::default(),
             right_tab: RightTab::Participants,
         }
     }
@@ -130,11 +134,28 @@ impl LkApp {
                         self.remote_data_tracks
                             .push(RemoteDataTrackTile::new(self.async_runtime.handle(), track));
                     }
+                    RoomEvent::TextStreamOpened { reader, topic, participant_identity } => {
+                        self.data_streams_ui.on_text_stream(
+                            reader,
+                            topic,
+                            participant_identity,
+                            &self.service,
+                        );
+                    }
+                    RoomEvent::ByteStreamOpened { reader, topic, participant_identity } => {
+                        self.data_streams_ui.on_byte_stream(
+                            reader,
+                            topic,
+                            participant_identity,
+                            &self.service,
+                        );
+                    }
                     RoomEvent::Disconnected { reason: _ } => {
                         self.video_renderers.clear();
                         self.local_data_tracks.clear();
                         self.remote_data_tracks.clear();
                         self.rpc_ui.on_disconnect();
+                        self.data_streams_ui.on_disconnect();
                     }
                     _ => {}
                 }
@@ -264,6 +285,7 @@ impl LkApp {
         ui.horizontal(|ui| {
             ui.selectable_value(&mut self.right_tab, RightTab::Participants, "Participants");
             ui.selectable_value(&mut self.right_tab, RightTab::Rpc, "RPC");
+            ui.selectable_value(&mut self.right_tab, RightTab::DataStreams, "Data Streams");
         });
         ui.separator();
 
@@ -281,6 +303,13 @@ impl LkApp {
                     rpc_ui.show(ui, service, &room);
                 });
             }
+            RightTab::DataStreams => {
+                let service = &self.service;
+                let data_streams_ui = &mut self.data_streams_ui;
+                egui::ScrollArea::vertical().show(ui, |ui| {
+                    data_streams_ui.show(ui, service, &room);
+                });
+            }
         }
     }
 
@@ -300,6 +329,15 @@ impl LkApp {
                 sorted_tracks.sort_by(|a, b| a.as_str().cmp(b.as_str()));
 
                 ui.monospace(&participant.identity().0);
+                ui.label(format!("Client protocol: {}", participant.client_protocol()));
+                let caps = participant.capabilities();
+                let caps_str = if caps.is_empty() {
+                    "(none)".to_string()
+                } else {
+                    caps.iter().map(|c| format!("{:?}", c)).collect::<Vec<_>>().join(", ")
+                };
+                ui.label(format!("Capabilities: {}", caps_str));
+
                 for tsid in sorted_tracks {
                     let publication = tracks.get(&tsid).unwrap().clone();
 

examples/wgpu_room/src/main.rs

@@ -5,6 +5,7 @@
 use std::time::Duration;
 
 mod app;
+mod data_streams_ui;
 mod data_track;
 mod logo_track;
 mod rpc_ui;

livekit-api/src/signal_client/mod.rs

@@ -57,18 +57,26 @@ const VALIDATE_TIMEOUT: Duration = Duration::from_secs(3);
 pub const PROTOCOL_VERSION: u32 = 17;
 
 /// Capabilities the Rust SDK advertises to the SFU at connect time.
-const CLIENT_CAPABILITIES: &[proto::client_info::Capability] =
-    &[proto::client_info::Capability::CapPacketTrailer];
+///
+/// `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,
+];
 
 /// 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;
+/// `ClientInfo.client_protocol` value indicating support for data streams v2
+/// (inline single-packet sends; compression is gated separately via capabilities).
+pub const CLIENT_PROTOCOL_DATA_STREAM_V2: i32 = 2;
 
 /// The client protocol which is sent to other clients and indicates the set of apis that other
 /// clients should assume this client supports.
-const CLIENT_PROTOCOL_VERSION: i32 = CLIENT_PROTOCOL_DATA_STREAM_RPC;
+const CLIENT_PROTOCOL_VERSION: i32 = CLIENT_PROTOCOL_DATA_STREAM_V2;
 
 #[derive(Error, Debug)]
 pub enum SignalError {

livekit-ffi/src/conversion/data_stream.rs

@@ -72,6 +72,8 @@ impl From<proto::StreamTextOptions> for StreamTextOptions {
             reply_to_stream_id: options.reply_to_stream_id,
             attached_stream_ids: options.attached_stream_ids,
             generated: options.generated,
+            // Compression opt-out is not yet exposed over FFI; default to on.
+            compress: None,
         }
     }
 }
@@ -90,6 +92,8 @@ impl From<proto::StreamByteOptions> for StreamByteOptions {
             name: options.name,
             mime_type: options.mime_type,
             total_length: options.total_length,
+            // Compression opt-out is not yet exposed over FFI; default to on.
+            compress: None,
         }
     }
 }

livekit/Cargo.toml

@@ -52,6 +52,7 @@ semver = "1.0"
 libloading = { version = "0.8.6" }
 bytes = "1.10.1"
 bmrng = "0.5.2"
+flate2 = "1"
 base64 = "0.22"
 
 [dev-dependencies]

livekit/src/prelude.rs

@@ -22,8 +22,8 @@ pub use crate::{
     },
     id::*,
     participant::{
-        ConnectionQuality, DisconnectReason, LocalParticipant, Participant, PerformRpcData,
-        RemoteParticipant, RpcError, RpcErrorCode, RpcInvocationData,
+        ClientCapability, ConnectionQuality, DisconnectReason, LocalParticipant, Participant,
+        PerformRpcData, RemoteParticipant, RpcError, RpcErrorCode, RpcInvocationData,
     },
     publication::{LocalTrackPublication, RemoteTrackPublication, TrackPublication},
     track::{