Data streams v2

.changeset/data_streams_v2.md

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

Cargo.toml

@@ -4,6 +4,7 @@ members = [
     "livekit",
     "livekit-api",
     "livekit-protocol",
+    "livekit-common",
     "livekit-ffi",
     "livekit-uniffi",
     "livekit-datatrack",
@@ -51,6 +52,7 @@ livekit = { version = "0.7.49", path = "livekit" }
 livekit-api = { version = "0.5.4", path = "livekit-api" }
 livekit-ffi = { version = "0.12.67", 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.9", path = "livekit-protocol" }
 livekit-runtime = { version = "0.4.0", path = "livekit-runtime" }
 soxr-sys = { version = "0.1.3", path = "soxr-sys" }

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();