diff --git a/Cargo.lock b/Cargo.lock index f6ac222fe..9c9ff3a86 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8746,26 +8746,6 @@ dependencies = [ "web-sys", ] -[[package]] -name = "wgpu_room" -version = "0.1.1" -dependencies = [ - "console-subscriber", - "eframe", - "egui", - "egui-wgpu", - "env_logger 0.11.10", - "futures", - "image", - "livekit", - "log", - "parking_lot", - "serde", - "tokio", - "wgpu", - "winit", -] - [[package]] name = "which" version = "4.4.2" diff --git a/Cargo.toml b/Cargo.toml index 93d4a11df..ac01ddead 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,7 +34,6 @@ members = [ "examples/screensharing", "examples/send_bytes", "examples/webhooks", - "examples/wgpu_room", ] [workspace.package] diff --git a/examples/wgpu_room/Cargo.toml b/examples/wgpu_room/Cargo.toml deleted file mode 100644 index b06952157..000000000 --- a/examples/wgpu_room/Cargo.toml +++ /dev/null @@ -1,25 +0,0 @@ -[package] -name = "wgpu_room" -version = "0.1.1" -edition.workspace = true -publish = false - -[features] -default = [] -tracing = ["console-subscriber", "tokio/tracing"] - -[dependencies] -tokio = { workspace = true, features = ["full", "parking_lot"] } -livekit = { workspace = true, features = ["rustls-tls-native-roots"] } -futures = { workspace = true } -parking_lot = { workspace = true, features = ["deadlock_detection"] } -eframe = { workspace = true, features = ["default_fonts", "wgpu", "persistence"] } -egui = { workspace = true } -egui-wgpu = { workspace = true } -image = { workspace = true } -wgpu = { workspace = true } -winit = { workspace = true, features = [ "android-native-activity" ] } -serde = { workspace = true, features = ["derive"] } -log = { workspace = true } -env_logger = { workspace = true } -console-subscriber = { workspace = true, features = ["parking_lot"], optional = true } diff --git a/examples/wgpu_room/README.md b/examples/wgpu_room/README.md new file mode 100644 index 000000000..711dba94f --- /dev/null +++ b/examples/wgpu_room/README.md @@ -0,0 +1,4 @@ +# This example has moved + +The `wgpu_room` example has been broken out into its own repository: +https://github.com/livekit-examples/rust-dev-client diff --git a/examples/wgpu_room/src/app.rs b/examples/wgpu_room/src/app.rs deleted file mode 100644 index 01002dbe9..000000000 --- a/examples/wgpu_room/src/app.rs +++ /dev/null @@ -1,670 +0,0 @@ -use crate::{ - data_track::{LocalDataTrackTile, RemoteDataTrackTile, MAX_VALUE, TIME_WINDOW}, - rpc_ui::RpcUiState, - service::{AsyncCmd, LkService, UiCmd}, - video_grid::VideoGrid, - video_renderer::VideoRenderer, -}; -use egui::{emath, epaint, pos2, Color32, CornerRadius, Rect, Stroke}; -use livekit::{e2ee::EncryptionType, prelude::*, track::VideoQuality, SimulateScenario}; -use std::collections::HashMap; -use std::sync::Arc; - -/// The state of the application are saved on app exit and restored on app start. -#[derive(serde::Deserialize, serde::Serialize)] -#[serde(default)] -struct AppState { - url: String, - token: String, - key: String, - auto_subscribe: bool, - enable_e2ee: bool, -} - -#[derive(Clone, Copy, PartialEq, Eq)] -enum RightTab { - Participants, - Rpc, -} - -pub struct LkApp { - async_runtime: tokio::runtime::Runtime, - state: AppState, - video_renderers: HashMap<(ParticipantIdentity, TrackSid), VideoRenderer>, - local_data_tracks: Vec, - remote_data_tracks: Vec, - connecting: bool, - connection_failure: Option, - render_state: egui_wgpu::RenderState, - service: LkService, - rpc_ui: RpcUiState, - right_tab: RightTab, -} - -impl Default for AppState { - fn default() -> Self { - Self { - url: "ws://localhost:7880".to_string(), - token: "".to_string(), - auto_subscribe: true, - enable_e2ee: false, - key: "".to_string(), - } - } -} - -impl LkApp { - pub fn new(cc: &eframe::CreationContext<'_>) -> Self { - let state = cc - .storage - .and_then(|storage| eframe::get_value(storage, eframe::APP_KEY)) - .unwrap_or_default(); - - let async_runtime = - tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap(); - - Self { - service: LkService::new(async_runtime.handle()), - async_runtime, - state, - video_renderers: HashMap::new(), - local_data_tracks: Vec::new(), - remote_data_tracks: Vec::new(), - connecting: false, - connection_failure: None, - render_state: cc.wgpu_render_state.clone().unwrap(), - rpc_ui: RpcUiState::default(), - right_tab: RightTab::Participants, - } - } - - fn event(&mut self, event: UiCmd) { - match event { - UiCmd::ConnectResult { result } => { - self.connecting = false; - if let Err(err) = result { - self.connection_failure = Some(err.to_string()); - } - } - UiCmd::DataTrackPublished { track } => { - self.local_data_tracks.push(LocalDataTrackTile::new(track)); - } - UiCmd::DataTrackUnpublished => { - self.local_data_tracks.clear(); - } - UiCmd::RpcSendResult { request_id, result } => { - self.rpc_ui.handle_send_result(request_id, result); - } - UiCmd::RoomEvent { event } => { - log::info!("{:?}", event); - match event { - RoomEvent::TrackSubscribed { track, publication: _, participant } => { - if let RemoteTrack::Video(ref video_track) = track { - let video_renderer = VideoRenderer::new( - self.async_runtime.handle(), - self.render_state.clone(), - video_track.rtc_track(), - ); - self.video_renderers - .insert((participant.identity(), track.sid()), video_renderer); - } - } - RoomEvent::TrackUnsubscribed { track, publication: _, participant } => { - self.video_renderers.remove(&(participant.identity(), track.sid())); - } - RoomEvent::LocalTrackPublished { track, publication: _, participant } => { - if let LocalTrack::Video(ref video_track) = track { - let video_renderer = VideoRenderer::new( - self.async_runtime.handle(), - self.render_state.clone(), - video_track.rtc_track(), - ); - self.video_renderers - .insert((participant.identity(), track.sid()), video_renderer); - } - } - RoomEvent::LocalTrackUnpublished { publication, participant } => { - self.video_renderers.remove(&(participant.identity(), publication.sid())); - } - RoomEvent::DataTrackPublished(track) => { - self.remote_data_tracks - .push(RemoteDataTrackTile::new(self.async_runtime.handle(), track)); - } - RoomEvent::Disconnected { reason: _ } => { - self.video_renderers.clear(); - self.local_data_tracks.clear(); - self.remote_data_tracks.clear(); - self.rpc_ui.on_disconnect(); - } - _ => {} - } - } - } - } - - fn top_panel(&mut self, ui: &mut egui::Ui) { - egui::MenuBar::new().ui(ui, |ui| { - ui.menu_button("Simulate", |ui| { - let scenarios = [ - SimulateScenario::SignalReconnect, - SimulateScenario::Speaker, - SimulateScenario::NodeFailure, - SimulateScenario::ServerLeave, - SimulateScenario::Migration, - SimulateScenario::ForceTcp, - SimulateScenario::ForceTls, - ]; - - for scenario in scenarios { - if ui.button(format!("{:?}", scenario)).clicked() { - let _ = self.service.send(AsyncCmd::SimulateScenario { scenario }); - } - } - }); - - ui.menu_button("Publish", |ui| { - if ui.button("Logo").clicked() { - let _ = self.service.send(AsyncCmd::ToggleLogo); - } - if ui.button("SineWave").clicked() { - let _ = self.service.send(AsyncCmd::ToggleSine); - } - if ui.button("DataTrack").clicked() { - let _ = self.service.send(AsyncCmd::ToggleDataTrack); - } - }); - - ui.menu_button("Debug", |ui| { - if ui.button("Log stats").clicked() { - let _ = self.service.send(AsyncCmd::LogStats); - } - }); - }); - } - - /// Connection form and room info - fn left_panel(&mut self, ui: &mut egui::Ui) { - let room = self.service.room(); - let connected = room.is_some() - && room.as_ref().unwrap().connection_state() == ConnectionState::Connected; - - ui.add_space(8.0); - ui.monospace("Livekit - Connect to a room"); - ui.add_space(8.0); - - ui.horizontal(|ui| { - ui.label("Url: "); - ui.text_edit_singleline(&mut self.state.url); - }); - - ui.horizontal(|ui| { - ui.label("Token: "); - ui.text_edit_singleline(&mut self.state.token); - }); - - ui.horizontal(|ui| { - ui.label("E2ee Key: "); - ui.text_edit_singleline(&mut self.state.key); - }); - - ui.horizontal(|ui| { - ui.add_enabled_ui(true, |ui| { - ui.checkbox(&mut self.state.enable_e2ee, "Enable E2ee"); - }); - }); - - ui.horizontal(|ui| { - ui.add_enabled_ui(!connected && !self.connecting, |ui| { - if ui.button("Connect").clicked() { - self.connecting = true; - self.connection_failure = None; - let _ = self.service.send(AsyncCmd::RoomConnect { - url: self.state.url.clone(), - token: self.state.token.clone(), - auto_subscribe: self.state.auto_subscribe, - enable_e2ee: self.state.enable_e2ee, - key: self.state.key.clone(), - }); - } - }); - - if self.connecting { - ui.spinner(); - } else if connected && ui.button("Disconnect").clicked() { - let _ = self.service.send(AsyncCmd::RoomDisconnect); - } - }); - - if ui.button("E2eeKeyRatchet").clicked() { - let _ = self.service.send(AsyncCmd::E2eeKeyRatchet); - } - - ui.horizontal(|ui| { - ui.add_enabled_ui(true, |ui| { - ui.checkbox(&mut self.state.auto_subscribe, "Auto Subscribe"); - }); - }); - - if let Some(err) = &self.connection_failure { - ui.colored_label(egui::Color32::RED, err); - } - - if let Some(room) = room.as_ref() { - ui.label(format!("Name: {}", room.name())); - //ui.label(format!("Sid: {}", String::from(room.sid().await))); - ui.label(format!("ConnectionState: {:?}", room.connection_state())); - ui.label(format!("ParticipantCount: {:?}", room.remote_participants().len() + 1)); - } - - egui::warn_if_debug_build(ui); - ui.separator(); - } - - fn right_panel(&mut self, ui: &mut egui::Ui) { - ui.horizontal(|ui| { - ui.selectable_value(&mut self.right_tab, RightTab::Participants, "Participants"); - ui.selectable_value(&mut self.right_tab, RightTab::Rpc, "RPC"); - }); - ui.separator(); - - let Some(room) = self.service.room() else { - ui.label("Not connected"); - return; - }; - - match self.right_tab { - RightTab::Participants => self.participants_tab(ui, &room), - RightTab::Rpc => { - let service = &self.service; - let rpc_ui = &mut self.rpc_ui; - egui::ScrollArea::vertical().show(ui, |ui| { - rpc_ui.show(ui, service, &room); - }); - } - } - } - - /// Show remote_participants and their tracks - fn participants_tab(&self, ui: &mut egui::Ui, room: &Arc) { - egui::ScrollArea::vertical().show(ui, |ui| { - // Iterate with sorted keys to avoid flickers (Because this is a immediate mode UI) - let participants = room.remote_participants(); - let mut sorted_participants = - participants.keys().cloned().collect::>(); - sorted_participants.sort_by(|a, b| a.as_str().cmp(b.as_str())); - - for psid in sorted_participants { - let participant = participants.get(&psid).unwrap(); - let tracks = participant.track_publications(); - let mut sorted_tracks = tracks.keys().cloned().collect::>(); - sorted_tracks.sort_by(|a, b| a.as_str().cmp(b.as_str())); - - ui.monospace(&participant.identity().0); - for tsid in sorted_tracks { - let publication = tracks.get(&tsid).unwrap().clone(); - - ui.horizontal(|ui| { - ui.label("Encrypted - "); - let enc_type = publication.encryption_type(); - if enc_type == EncryptionType::None { - ui.colored_label(egui::Color32::RED, format!("{:?}", enc_type)); - } else { - ui.colored_label(egui::Color32::GREEN, format!("{:?}", enc_type)); - } - }); - - ui.label(format!("{} - {:?}", publication.name(), publication.source())); - - ui.horizontal(|ui| { - ui.label("Simulcasted - "); - let is_simulcasted = publication.simulcasted(); - ui.label(if is_simulcasted { "Yes" } else { "No" }); - if is_simulcasted { - ui.menu_button("Set Quality", |ui| { - let publication = publication.clone(); - if ui.button("Low").clicked() { - let _ = self.service.send(AsyncCmd::SetVideoQuality { - publication, - quality: VideoQuality::Low, - }); - } else if ui.button("Medium").clicked() { - let _ = self.service.send(AsyncCmd::SetVideoQuality { - publication, - quality: VideoQuality::Medium, - }); - } else if ui.button("High").clicked() { - let _ = self.service.send(AsyncCmd::SetVideoQuality { - publication, - quality: VideoQuality::High, - }); - } - }); - } - }); - - ui.horizontal(|ui| { - if publication.is_muted() { - ui.colored_label(egui::Color32::DARK_GRAY, "Muted"); - } - - if publication.is_subscribed() { - ui.colored_label(egui::Color32::GREEN, "Subscribed"); - } else { - ui.colored_label(egui::Color32::RED, "Unsubscribed"); - } - - if publication.is_subscribed() { - if ui.button("Unsubscribe").clicked() { - let _ = - self.service.send(AsyncCmd::UnsubscribeTrack { publication }); - } - } else if ui.button("Subscribe").clicked() { - let _ = self.service.send(AsyncCmd::SubscribeTrack { publication }); - } - }); - } - ui.separator(); - } - }); - } - - /// Draw a grid of all track tiles (video + data) - fn central_panel(&mut self, ui: &mut egui::Ui) { - let room = self.service.room(); - let connected = room.is_some(); - - let has_tiles = !self.video_renderers.is_empty() - || !self.local_data_tracks.is_empty() - || !self.remote_data_tracks.is_empty(); - - if connected && !has_tiles { - ui.centered_and_justified(|ui| { - ui.label("No tracks subscribed"); - }); - return; - } - - egui::ScrollArea::vertical().show(ui, |ui| { - VideoGrid::new("default_grid").max_columns(6).show(ui, |ui| { - if connected { - let room = room.as_ref().unwrap(); - - for ((participant_id, _), video_renderer) in &self.video_renderers { - ui.video_frame(|ui| { - if let Some(p) = room.remote_participants().get(participant_id) { - draw_video(p.name().as_str(), p.is_speaking(), video_renderer, ui); - } else { - let lp = room.local_participant(); - draw_video( - lp.name().as_str(), - lp.is_speaking(), - video_renderer, - ui, - ); - } - }); - } - - for tile in &mut self.local_data_tracks { - ui.video_frame(|ui| draw_local_data_track(tile, ui)); - } - - for tile in &self.remote_data_tracks { - ui.video_frame(|ui| draw_remote_data_track(tile, ui)); - } - } else { - for _ in 0..5 { - ui.video_frame(|ui| { - egui::Frame::NONE.fill(ui.style().visuals.code_bg_color).show( - ui, - |ui| { - ui.allocate_space(ui.available_size()); - }, - ); - }); - } - } - }) - }); - } -} - -impl eframe::App for LkApp { - fn save(&mut self, storage: &mut dyn eframe::Storage) { - eframe::set_value(storage, eframe::APP_KEY, &self.state); - } - - fn ui(&mut self, root_ui: &mut egui::Ui, _frame: &mut eframe::Frame) { - if let Some(event) = self.service.try_recv() { - self.event(event); - } - - egui::Panel::top("top_panel").show(root_ui, |ui| { - self.top_panel(ui); - }); - - egui::Panel::left("left_panel").resizable(true).min_size(20.0).max_size(360.0).show( - root_ui, - |ui| { - self.left_panel(ui); - }, - ); - - /*egui::TopBottomPanel::bottom("bottom_panel") - .resizable(true) - .height_range(20.0..=256.0) - .show(ctx, |ui| { - self.bottom_panel(ui); - });*/ - - egui::Panel::right("right_panel").resizable(true).min_size(20.0).max_size(360.0).show( - root_ui, - |ui| { - self.right_panel(ui); - }, - ); - - egui::CentralPanel::default().show(root_ui, |ui| { - self.central_panel(ui); - }); - - root_ui.ctx().request_repaint(); - } -} - -fn draw_video(name: &str, speaking: bool, video_renderer: &VideoRenderer, ui: &mut egui::Ui) { - let rect = ui.available_rect_before_wrap(); - let inner_rect = rect.shrink(1.0); - - if speaking { - ui.painter().rect( - rect, - CornerRadius::default(), - egui::Color32::GREEN, - Stroke::NONE, - egui::StrokeKind::Inside, - ); - } - - let resolution = video_renderer.resolution(); - if let Some(tex) = video_renderer.texture_id() { - ui.painter().image( - tex, - inner_rect, - egui::Rect::from_min_max(egui::pos2(0.0, 0.0), egui::pos2(1.0, 1.0)), - egui::Color32::WHITE, - ); - } - - ui.painter().text( - egui::pos2(rect.min.x + 5.0, rect.max.y - 5.0), - egui::Align2::LEFT_BOTTOM, - format!("{}x{} {}", resolution.0, resolution.1, name), - egui::FontId::default(), - egui::Color32::WHITE, - ); -} - -struct DataTrackChart<'a> { - points: &'a parking_lot::Mutex>, - name: &'a str, - publisher_label: &'a str, - drag_value: Option<&'a mut i32>, -} - -impl<'a> DataTrackChart<'a> { - fn new( - points: &'a parking_lot::Mutex>, - name: &'a str, - publisher_label: &'a str, - ) -> Self { - Self { points, name, publisher_label, drag_value: None } - } - - fn interactive(mut self, value: &'a mut i32) -> Self { - self.drag_value = Some(value); - self - } -} - -impl egui::Widget for DataTrackChart<'_> { - fn ui(self, ui: &mut egui::Ui) -> egui::Response { - let mut drag_value = self.drag_value; - let interactive = drag_value.is_some(); - let sense = if interactive { egui::Sense::click_and_drag() } else { egui::Sense::hover() }; - - let desired_size = ui.available_size(); - let (rect, mut response) = ui.allocate_exact_size(desired_size, sense); - let painter = ui.painter(); - - let bg = Color32::from_rgb(0x1a, 0x1a, 0x2e); - painter.rect_filled(rect, CornerRadius::default(), bg); - - let v_margin = rect.height() * 0.15; - let h_margin = 8.0; - let label_width = 32.0; - let plot_rect = Rect::from_min_max( - pos2(rect.min.x + h_margin, rect.min.y + v_margin), - pos2(rect.max.x - h_margin - label_width, rect.max.y - v_margin), - ); - - let time_window_secs = TIME_WINDOW.as_secs_f32(); - let to_screen = emath::RectTransform::from_to( - Rect::from_x_y_ranges(time_window_secs..=0.0, MAX_VALUE..=0.0), - plot_rect, - ); - - let guide_color = Color32::from_rgb(0x40, 0x40, 0x50); - let max_y = (to_screen * pos2(0.0, MAX_VALUE)).y; - let min_y = (to_screen * pos2(0.0, 0.0)).y; - painter.line_segment( - [pos2(plot_rect.min.x, max_y), pos2(plot_rect.max.x, max_y)], - Stroke::new(1.0, guide_color), - ); - painter.line_segment( - [pos2(plot_rect.min.x, min_y), pos2(plot_rect.max.x, min_y)], - Stroke::new(1.0, guide_color), - ); - - if let Some(value) = &mut drag_value { - if let Some(pointer_pos) = response.interact_pointer_pos() { - let from_screen = to_screen.inverse(); - let logical = from_screen * pointer_pos; - let new_val = (logical.y as i32).clamp(0, MAX_VALUE as i32); - if **value != new_val { - **value = new_val; - response.mark_changed(); - } - } - } - - let now = std::time::Instant::now(); - let mut points = self.points.lock(); - while points.back().is_some_and(|(t, _)| now.duration_since(*t) > TIME_WINDOW) { - points.pop_back(); - } - - let is_interacting = response.interact_pointer_pos().is_some(); - let display_val = drag_value - .as_deref() - .copied() - .filter(|_| !points.is_empty() || is_interacting) - .or_else(|| points.front().map(|(_, v)| *v)); - - let line_color = Color32::from_rgb(0xFF, 0x44, 0x44); - - if !points.is_empty() { - let mut screen_points = Vec::with_capacity(points.len() + 1); - if let Some(val) = display_val { - screen_points.push(to_screen * pos2(0.0, val as f32)); - } - for &(t, val) in points.iter() { - let age = now.duration_since(t).as_secs_f32(); - screen_points.push(to_screen * pos2(age, val as f32)); - } - drop(points); - painter - .add(epaint::Shape::line(screen_points, epaint::PathStroke::new(2.0, line_color))); - ui.ctx().request_repaint(); - } else { - drop(points); - } - - if let Some(val) = display_val { - let newest_screen = to_screen * pos2(0.0, val as f32); - let is_active = interactive && (response.hovered() || response.dragged()); - let dot_radius = if is_active { 6.0 } else { 4.0 }; - painter.circle_filled(newest_screen, dot_radius, line_color); - if is_active { - painter.circle_stroke( - newest_screen, - dot_radius + 2.0, - Stroke::new(1.5, Color32::WHITE), - ); - } - - painter.text( - pos2(plot_rect.max.x + 8.0, newest_screen.y), - egui::Align2::LEFT_CENTER, - val.to_string(), - egui::FontId::monospace(14.0), - Color32::WHITE, - ); - } else { - let hint = if interactive { "Drag to Push Frames…" } else { "Waiting for Frames…" }; - painter.text( - rect.center(), - egui::Align2::CENTER_CENTER, - hint, - egui::FontId::proportional(18.0), - Color32::WHITE, - ); - } - - painter.text( - pos2(rect.min.x + 5.0, rect.max.y - 5.0), - egui::Align2::LEFT_BOTTOM, - format!("Data: {} ({})", self.name, self.publisher_label), - egui::FontId::default(), - Color32::WHITE, - ); - - if interactive { - response = response.on_hover_cursor(egui::CursorIcon::ResizeVertical); - } - - response - } -} - -fn draw_local_data_track(tile: &mut LocalDataTrackTile, ui: &mut egui::Ui) { - let chart = - DataTrackChart::new(&tile.points, &tile.name, "local").interactive(&mut tile.slider_value); - if ui.add(chart).changed() { - tile.push_value(); - } -} - -fn draw_remote_data_track(tile: &RemoteDataTrackTile, ui: &mut egui::Ui) { - ui.add(DataTrackChart::new(&tile.points, &tile.name, &tile.publisher_identity)); -} diff --git a/examples/wgpu_room/src/data_track.rs b/examples/wgpu_room/src/data_track.rs deleted file mode 100644 index 8f9ce0852..000000000 --- a/examples/wgpu_room/src/data_track.rs +++ /dev/null @@ -1,62 +0,0 @@ -use futures::StreamExt; -use livekit::prelude::*; -use parking_lot::Mutex; -use std::collections::VecDeque; -use std::sync::Arc; -use std::time::{Duration, Instant}; - -pub const TIME_WINDOW: Duration = Duration::from_secs(30); -pub const MAX_VALUE: f32 = 512.0; - -pub struct LocalDataTrackTile { - track: LocalDataTrack, - pub slider_value: i32, - pub points: Arc>>, - pub name: String, -} - -impl LocalDataTrackTile { - pub fn new(track: LocalDataTrack) -> Self { - let name = track.info().name().to_string(); - Self { track, slider_value: 0, points: Arc::new(Mutex::new(VecDeque::new())), name } - } - - pub fn push_value(&self) { - let frame = DataTrackFrame::new(self.slider_value.to_string().into_bytes()); - let _ = self.track.try_push(frame); - self.points.lock().push_front((Instant::now(), self.slider_value)); - } -} - -pub struct RemoteDataTrackTile { - pub points: Arc>>, - pub publisher_identity: String, - pub name: String, -} - -impl RemoteDataTrackTile { - pub fn new(async_handle: &tokio::runtime::Handle, track: RemoteDataTrack) -> Self { - let points = Arc::new(Mutex::new(VecDeque::new())); - let points_ref = points.clone(); - let publisher_identity = track.publisher_identity().to_string(); - let name = track.info().name().to_string(); - - async_handle.spawn(async move { - let mut stream = match track.subscribe().await { - Ok(s) => s, - Err(err) => { - log::error!("Failed to subscribe to data track: {err}"); - return; - } - }; - while let Some(frame) = stream.next().await { - let payload = frame.payload(); - let Ok(s) = std::str::from_utf8(&payload) else { continue }; - let Ok(value) = s.parse::() else { continue }; - points_ref.lock().push_front((Instant::now(), value)); - } - }); - - Self { points, publisher_identity, name } - } -} diff --git a/examples/wgpu_room/src/logo_track.rs b/examples/wgpu_room/src/logo_track.rs deleted file mode 100644 index 4e184f661..000000000 --- a/examples/wgpu_room/src/logo_track.rs +++ /dev/null @@ -1,204 +0,0 @@ -use image::ImageFormat; -use image::RgbaImage; -use livekit::options::TrackPublishOptions; -use livekit::options::VideoCodec; -use livekit::prelude::*; -use livekit::webrtc::video_source::RtcVideoSource; -use livekit::webrtc::video_source::VideoResolution; -use livekit::webrtc::{ - native::yuv_helper, - video_frame::{I420Buffer, VideoFrame, VideoRotation}, - video_source::native::NativeVideoSource, -}; -use parking_lot::Mutex; -use std::sync::Arc; -use std::time::Duration; -use tokio::sync::oneshot; -use tokio::task::JoinHandle; - -// The logo must not be bigger than the framebuffer -const PIXEL_SIZE: usize = 4; -const FRAME_RATE: u64 = 30; -const MOVE_SPEED: i32 = 16; -const FB_WIDTH: usize = 1280; -const FB_HEIGHT: usize = 720; - -#[derive(Clone)] -struct FrameData { - image: Arc, - framebuffer: Arc>>, - video_frame: Arc>>, - pos: (i32, i32), - direction: (i32, i32), -} - -struct TrackHandle { - close_tx: oneshot::Sender<()>, - track: LocalVideoTrack, - task: JoinHandle<()>, -} - -pub struct LogoTrack { - rtc_source: NativeVideoSource, - room: Arc, - handle: Option, -} - -impl LogoTrack { - pub fn new(room: Arc) -> Self { - Self { - rtc_source: NativeVideoSource::new( - VideoResolution { width: FB_WIDTH as u32, height: FB_HEIGHT as u32 }, - false, - ), - room, - handle: None, - } - } - - pub fn is_published(&self) -> bool { - self.handle.is_some() - } - - pub async fn publish(&mut self) -> Result<(), RoomError> { - self.unpublish().await?; - - let (close_tx, close_rx) = oneshot::channel(); - let track = LocalVideoTrack::create_video_track( - "livekit_logo", - RtcVideoSource::Native(self.rtc_source.clone()), - ); - - let task = tokio::spawn(Self::track_task(close_rx, self.rtc_source.clone())); - - self.room - .local_participant() - .publish_track( - LocalTrack::Video(track.clone()), - TrackPublishOptions { - source: TrackSource::Camera, - simulcast: true, - video_codec: VideoCodec::H265, - ..Default::default() - }, - ) - .await?; - - let handle = TrackHandle { close_tx, task, track }; - - self.handle = Some(handle); - Ok(()) - } - - pub async fn unpublish(&mut self) -> Result<(), RoomError> { - if let Some(handle) = self.handle.take() { - let _ = handle.close_tx.send(()); - let _ = handle.task.await; - - self.room.local_participant().unpublish_track(&handle.track.sid()).await?; - } - Ok(()) - } - - async fn track_task(mut close_rx: oneshot::Receiver<()>, rtc_source: NativeVideoSource) { - let mut interval = tokio::time::interval(Duration::from_millis(1000 / FRAME_RATE)); - - let image = tokio::task::spawn_blocking(|| { - image::load_from_memory_with_format(include_bytes!("moving-logo.png"), ImageFormat::Png) - .unwrap() - .to_rgba8() - }) - .await - .unwrap(); - - let mut data = FrameData { - image: Arc::new(image), - framebuffer: Arc::new(Mutex::new(vec![0u8; FB_WIDTH * FB_HEIGHT * 4])), - video_frame: Arc::new(Mutex::new(VideoFrame { - rotation: VideoRotation::VideoRotation0, - timestamp_us: 0, - frame_metadata: None, - buffer: I420Buffer::new(FB_WIDTH as u32, FB_HEIGHT as u32), - })), - pos: (0, 0), - direction: (1, 1), - }; - - loop { - tokio::select! { - _ = &mut close_rx => { - break; - } - _ = interval.tick() => {} - } - - data.pos.0 += data.direction.0 * MOVE_SPEED; - data.pos.1 += data.direction.1 * MOVE_SPEED; - - if data.pos.0 >= (FB_WIDTH - data.image.width() as usize) as i32 { - data.direction.0 = -1; - } else if data.pos.0 <= 0 { - data.direction.0 = 1; - } - - if data.pos.1 >= (FB_HEIGHT - data.image.height() as usize) as i32 { - data.direction.1 = -1; - } else if data.pos.1 <= 0 { - data.direction.1 = 1; - } - - tokio::task::spawn_blocking({ - let data = data.clone(); - let source = rtc_source.clone(); - move || { - let image = data.image.as_raw(); - let mut framebuffer = data.framebuffer.lock(); - let mut video_frame = data.video_frame.lock(); - let i420_buffer = &mut video_frame.buffer; - - let (stride_y, stride_u, stride_v) = i420_buffer.strides(); - let (data_y, data_u, data_v) = i420_buffer.data_mut(); - - framebuffer.fill(0); - for i in 0..data.image.height() as usize { - let x = data.pos.0 as usize; - let y = data.pos.1 as usize; - let frame_width = data.image.width() as usize; - let logo_stride = frame_width * PIXEL_SIZE; - let row_start = (x + ((i + y) * FB_WIDTH)) * PIXEL_SIZE; - let row_end = row_start + logo_stride; - - framebuffer[row_start..row_end].copy_from_slice( - &image[i * logo_stride..i * logo_stride + logo_stride], - ); - } - - yuv_helper::abgr_to_i420( - &framebuffer, - (FB_WIDTH * PIXEL_SIZE) as u32, - data_y, - stride_y, - data_u, - stride_u, - data_v, - stride_v, - FB_WIDTH as i32, - FB_HEIGHT as i32, - ); - - source.capture_frame(&*video_frame); - } - }) - .await - .unwrap(); - } - } -} - -impl Drop for LogoTrack { - fn drop(&mut self) { - if let Some(handle) = self.handle.take() { - let _ = handle.close_tx.send(()); - } - } -} diff --git a/examples/wgpu_room/src/main.rs b/examples/wgpu_room/src/main.rs deleted file mode 100644 index c85baccf3..000000000 --- a/examples/wgpu_room/src/main.rs +++ /dev/null @@ -1,44 +0,0 @@ -use eframe::Renderer; -use parking_lot::deadlock; -use std::thread; -use std::time::Duration; - -mod app; -mod data_track; -mod logo_track; -mod rpc_ui; -mod service; -mod sine_track; -mod video_grid; -mod video_renderer; - -fn main() { - env_logger::init(); - - #[cfg(feature = "tracing")] - console_subscriber::init(); - - // Create a background thread which checks for deadlocks every 10s - thread::spawn(move || loop { - thread::sleep(Duration::from_secs(10)); - let deadlocks = deadlock::check_deadlock(); - if deadlocks.is_empty() { - continue; - } - - log::error!("{} deadlocks detected", deadlocks.len()); - for (i, threads) in deadlocks.iter().enumerate() { - log::error!("Deadlock #{}", i); - for t in threads { - log::error!("Thread Id {:#?}: \n{:#?}", t.thread_id(), t.backtrace()); - } - } - }); - - eframe::run_native( - "LiveKit - Rust App", - eframe::NativeOptions { centered: true, renderer: Renderer::Wgpu, ..Default::default() }, - Box::new(|cc| Ok(Box::new(app::LkApp::new(cc)))), - ) - .unwrap(); -} diff --git a/examples/wgpu_room/src/moving-logo.png b/examples/wgpu_room/src/moving-logo.png deleted file mode 100644 index 5c47cc1ed..000000000 Binary files a/examples/wgpu_room/src/moving-logo.png and /dev/null differ diff --git a/examples/wgpu_room/src/rpc_ui.rs b/examples/wgpu_room/src/rpc_ui.rs deleted file mode 100644 index bb219aea0..000000000 --- a/examples/wgpu_room/src/rpc_ui.rs +++ /dev/null @@ -1,376 +0,0 @@ -use crate::service::{AsyncCmd, LkService}; -use egui::Color32; -use livekit::prelude::*; -use parking_lot::Mutex; -use std::collections::{BTreeMap, VecDeque}; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::Arc; -use std::time::{SystemTime, UNIX_EPOCH}; - -const MAX_INVOCATIONS: usize = 200; -const PAYLOAD_PREVIEW_CHARS: usize = 256; -const RESPONSE_PREVIEW_CHARS: usize = 40; - -static NEXT_SEND_ID: AtomicU64 = AtomicU64::new(1); - -pub struct RpcUiState { - send_destination: Option, - send_method: String, - send_payload: String, - send_in_flight: Option, - send_result: Option, - - register_method: String, - register_error: Option, - - handlers: BTreeMap>>, -} - -struct HandlerEntry { - method: String, - reply: String, - invocations: VecDeque, - invocation_count: u64, -} - -struct Invocation { - n: u64, - caller: String, - payload_len: usize, - received_at: SystemTime, - payload_preview: String, -} - -enum SendResult { - Ok(String), - Err { code: u32, message: String }, -} - -impl Default for RpcUiState { - fn default() -> Self { - Self { - send_destination: None, - send_method: String::new(), - send_payload: String::new(), - send_in_flight: None, - send_result: None, - register_method: String::new(), - register_error: None, - handlers: BTreeMap::new(), - } - } -} - -impl RpcUiState { - pub fn handle_send_result(&mut self, request_id: u64, result: Result) { - if self.send_in_flight == Some(request_id) { - self.send_in_flight = None; - } - self.send_result = Some(match result { - Ok(s) => SendResult::Ok(s), - Err(e) => SendResult::Err { code: e.code, message: e.message }, - }); - } - - pub fn on_disconnect(&mut self) { - self.handlers.clear(); - self.send_in_flight = None; - self.send_destination = None; - self.register_error = None; - } - - pub fn show(&mut self, ui: &mut egui::Ui, service: &LkService, room: &Arc) { - self.show_send(ui, service, room); - ui.add_space(8.0); - ui.separator(); - self.show_register(ui, service, room); - self.show_handler_cards(ui, service, room); - } - - fn show_send(&mut self, ui: &mut egui::Ui, service: &LkService, room: &Arc) { - ui.label(egui::RichText::new("Send RPC").strong()); - - let participants = room.remote_participants(); - let mut idents: Vec = participants.keys().cloned().collect(); - idents.sort_by(|a, b| a.as_str().cmp(b.as_str())); - - if let Some(sel) = self.send_destination.as_ref() { - if !participants.contains_key(sel) { - self.send_destination = None; - } - } - - ui.horizontal(|ui| { - ui.label("To:"); - let combo_label = - self.send_destination.as_ref().map(|i| i.as_str().to_string()).unwrap_or_else( - || { - if idents.is_empty() { - "(no remote participants)".to_string() - } else { - "(select)".to_string() - } - }, - ); - egui::ComboBox::from_id_salt("rpc_dest_combo").selected_text(combo_label).show_ui( - ui, - |ui| { - for ident in &idents { - ui.selectable_value( - &mut self.send_destination, - Some(ident.clone()), - ident.as_str(), - ); - } - }, - ); - }); - - ui.horizontal(|ui| { - ui.label("Method:"); - ui.add(egui::TextEdit::singleline(&mut self.send_method).desired_width(f32::INFINITY)); - }); - - ui.horizontal(|ui| { - ui.label("Payload:"); - if ui.small_button("Hello").clicked() { - self.send_payload = "Hello world".to_string(); - } - if ui.small_button("20k").clicked() { - self.send_payload = "X".repeat(20_000); - } - }); - let max_h = ui.text_style_height(&egui::TextStyle::Body) * 5.0 + 8.0; - egui::ScrollArea::vertical().id_salt("rpc_send_payload_scroll").max_height(max_h).show( - ui, - |ui| { - ui.add( - egui::TextEdit::multiline(&mut self.send_payload) - .desired_rows(2) - .desired_width(f32::INFINITY), - ); - }, - ); - - let can_send = self.send_in_flight.is_none() - && self.send_destination.is_some() - && !self.send_method.trim().is_empty(); - - ui.horizontal(|ui| { - ui.add_enabled_ui(can_send, |ui| { - if ui.button("Send").clicked() { - let dest = self.send_destination.as_ref().unwrap().0.clone(); - let method = self.send_method.clone(); - let payload = self.send_payload.clone(); - let request_id = NEXT_SEND_ID.fetch_add(1, Ordering::Relaxed); - self.send_in_flight = Some(request_id); - self.send_result = None; - let _ = service.send(AsyncCmd::RpcSendRequest { - destination: dest, - method, - payload, - request_id, - }); - } - }); - if self.send_in_flight.is_some() { - ui.spinner(); - } - }); - - if self.send_in_flight.is_some() { - let dest = - self.send_destination.as_ref().map(|i| i.as_str().to_string()).unwrap_or_default(); - ui.colored_label(Color32::GRAY, format!("Sending to {} {}...", dest, self.send_method)); - } else { - match &self.send_result { - Some(SendResult::Ok(s)) => { - ui.colored_label(Color32::LIGHT_GREEN, format!("OK: {}", preview_response(s))); - } - Some(SendResult::Err { code, message }) => { - ui.colored_label(Color32::LIGHT_RED, format!("Error {}: {}", code, message)); - } - None => {} - } - } - } - - fn show_register(&mut self, ui: &mut egui::Ui, service: &LkService, room: &Arc) { - ui.label(egui::RichText::new("Handlers").strong()); - - let mut do_register = false; - ui.horizontal(|ui| { - ui.label("Topic:"); - let resp = - ui.add(egui::TextEdit::singleline(&mut self.register_method).desired_width(120.0)); - if resp.lost_focus() && ui.input(|i| i.key_pressed(egui::Key::Enter)) { - do_register = true; - } - if ui.button("Register Handler").clicked() { - do_register = true; - } - }); - - if do_register { - self.register_error = None; - let method = self.register_method.trim().to_string(); - if method.is_empty() { - self.register_error = Some("Topic is empty".to_string()); - } else if self.handlers.contains_key(&method) { - self.register_error = Some(format!("Handler already registered for '{}'", method)); - } else { - let entry = Arc::new(Mutex::new(HandlerEntry { - method: method.clone(), - reply: String::new(), - invocations: VecDeque::new(), - invocation_count: 0, - })); - let entry_for_cb = entry.clone(); - let _guard = service.runtime().enter(); - room.local_participant().register_rpc_method(method.clone(), move |data| { - let entry_for_cb = entry_for_cb.clone(); - Box::pin(async move { - let reply = { - let mut g = entry_for_cb.lock(); - push_invocation(&mut g, &data); - g.reply.clone() - }; - Ok(reply) - }) - }); - self.handlers.insert(method, entry); - self.register_method.clear(); - } - } - - if let Some(err) = &self.register_error { - ui.colored_label(Color32::LIGHT_RED, err); - } - } - - fn show_handler_cards(&mut self, ui: &mut egui::Ui, service: &LkService, room: &Arc) { - let methods: Vec = self.handlers.keys().cloned().collect(); - let mut to_remove: Option = None; - - for method in methods { - let entry = self.handlers.get(&method).unwrap().clone(); - ui.add_space(6.0); - egui::Frame::group(ui.style()).show(ui, |ui| { - let mut guard = entry.lock(); - ui.horizontal(|ui| { - ui.monospace(egui::RichText::new(&guard.method).strong()); - ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| { - if ui.button("Unregister").clicked() { - to_remove = Some(guard.method.clone()); - } - }); - }); - - ui.horizontal(|ui| { - ui.label("Reply:"); - if ui.small_button("Hello").clicked() { - guard.reply = "Hello world".to_string(); - } - if ui.small_button("20k").clicked() { - guard.reply = "X".repeat(20_000); - } - }); - let max_h = ui.text_style_height(&egui::TextStyle::Body) * 5.0 + 8.0; - egui::ScrollArea::vertical() - .id_salt(format!("rpc_handler_reply_scroll_{}", guard.method)) - .max_height(max_h) - .show(ui, |ui| { - ui.add( - egui::TextEdit::multiline(&mut guard.reply) - .desired_rows(1) - .desired_width(f32::INFINITY), - ); - }); - - ui.label(format!("Invocations ({})", guard.invocation_count)); - - if guard.invocations.is_empty() { - ui.colored_label(Color32::GRAY, "No invocations yet"); - } else { - for inv in guard.invocations.iter() { - let meta = format!( - "#{} | {} | {} | {}", - inv.n, - inv.caller, - format_size(inv.payload_len), - format_ts(inv.received_at), - ); - ui.add(egui::Label::new( - egui::RichText::new(meta).small().color(Color32::GRAY), - )); - ui.add(egui::Label::new( - egui::RichText::new(&inv.payload_preview).monospace(), - )); - ui.separator(); - } - } - }); - } - - if let Some(m) = to_remove { - let _guard = service.runtime().enter(); - room.local_participant().unregister_rpc_method(m.clone()); - self.handlers.remove(&m); - } - } -} - -fn push_invocation(entry: &mut HandlerEntry, data: &RpcInvocationData) { - entry.invocation_count += 1; - let payload_len = data.payload.as_bytes().len(); - let payload_preview = truncate_chars(&data.payload, PAYLOAD_PREVIEW_CHARS); - entry.invocations.push_back(Invocation { - n: entry.invocation_count, - caller: data.caller_identity.as_str().to_string(), - payload_len, - received_at: SystemTime::now(), - payload_preview, - }); - while entry.invocations.len() > MAX_INVOCATIONS { - entry.invocations.pop_front(); - } -} - -fn truncate_chars(s: &str, max_chars: usize) -> String { - let mut iter = s.chars(); - let head: String = iter.by_ref().take(max_chars).collect(); - if iter.next().is_some() { - format!("{}...", head) - } else { - head - } -} - -fn preview_response(s: &str) -> String { - let bytes = s.as_bytes().len(); - let mut iter = s.chars(); - let head: String = iter.by_ref().take(RESPONSE_PREVIEW_CHARS).collect(); - if iter.next().is_some() { - format!("{}... ({}B)", head, bytes) - } else { - head - } -} - -fn format_size(bytes: usize) -> String { - if bytes < 1024 { - format!("{}B", bytes) - } else { - format!("{:.2}KB", bytes as f64 / 1024.0) - } -} - -fn format_ts(ts: SystemTime) -> String { - let d = ts.duration_since(UNIX_EPOCH).unwrap_or_default(); - let total = d.as_secs(); - let h = (total / 3600) % 24; - let m = (total / 60) % 60; - let s = total % 60; - let ms = d.subsec_millis(); - format!("{:02}:{:02}:{:02}.{:03}Z", h, m, s, ms) -} diff --git a/examples/wgpu_room/src/service.rs b/examples/wgpu_room/src/service.rs deleted file mode 100644 index 92ec20fe6..000000000 --- a/examples/wgpu_room/src/service.rs +++ /dev/null @@ -1,264 +0,0 @@ -use crate::{ - logo_track::LogoTrack, - sine_track::{SineParameters, SineTrack}, -}; -use livekit::{ - e2ee::{key_provider::*, E2eeOptions, EncryptionType}, - prelude::*, - track::VideoQuality, - SimulateScenario, -}; -use parking_lot::Mutex; -use std::sync::Arc; -use tokio::sync::mpsc::{self, error::SendError}; - -#[derive(Debug)] -pub enum AsyncCmd { - RoomConnect { url: String, token: String, auto_subscribe: bool, enable_e2ee: bool, key: String }, - RoomDisconnect, - SimulateScenario { scenario: SimulateScenario }, - ToggleLogo, - ToggleSine, - ToggleDataTrack, - SubscribeTrack { publication: RemoteTrackPublication }, - UnsubscribeTrack { publication: RemoteTrackPublication }, - SetVideoQuality { publication: RemoteTrackPublication, quality: VideoQuality }, - E2eeKeyRatchet, - LogStats, - RpcSendRequest { destination: String, method: String, payload: String, request_id: u64 }, -} - -#[derive(Debug)] -pub enum UiCmd { - ConnectResult { result: RoomResult<()> }, - RoomEvent { event: RoomEvent }, - DataTrackPublished { track: LocalDataTrack }, - DataTrackUnpublished, - RpcSendResult { request_id: u64, result: Result }, -} - -/// AppService is the "asynchronous" part of our application, where we connect to a room and -/// handle events. -pub struct LkService { - cmd_tx: mpsc::UnboundedSender, - ui_rx: mpsc::UnboundedReceiver, - handle: tokio::task::JoinHandle<()>, - inner: Arc, - runtime: tokio::runtime::Handle, -} - -struct ServiceInner { - ui_tx: mpsc::UnboundedSender, - room: Mutex>>, -} - -impl LkService { - /// Create a new AppService and return a channel that informs the UI of events. - pub fn new(async_handle: &tokio::runtime::Handle) -> Self { - let (ui_tx, ui_rx) = mpsc::unbounded_channel(); - let (cmd_tx, cmd_rx) = mpsc::unbounded_channel(); - - let inner = Arc::new(ServiceInner { ui_tx, room: Default::default() }); - let handle = async_handle.spawn(service_task(inner.clone(), cmd_rx)); - - Self { cmd_tx, ui_rx, handle, inner, runtime: async_handle.clone() } - } - - pub fn room(&self) -> Option> { - self.inner.room.lock().clone() - } - - pub fn runtime(&self) -> &tokio::runtime::Handle { - &self.runtime - } - - pub fn send(&self, cmd: AsyncCmd) -> Result<(), SendError> { - self.cmd_tx.send(cmd) - } - - pub fn try_recv(&mut self) -> Option { - self.ui_rx.try_recv().ok() - } - - #[allow(dead_code)] - pub async fn close(self) { - drop(self.cmd_tx); - let _ = self.handle.await; - } -} - -async fn service_task(inner: Arc, mut cmd_rx: mpsc::UnboundedReceiver) { - struct RunningState { - room: Arc, - logo_track: LogoTrack, - sine_track: SineTrack, - data_track: Option, - } - - let mut running_state = None; - - while let Some(event) = cmd_rx.recv().await { - match event { - AsyncCmd::RoomConnect { url, token, auto_subscribe, enable_e2ee, key } => { - log::info!("connecting to room: {}", url); - - let key_provider = - KeyProvider::with_shared_key(KeyProviderOptions::default(), key.into_bytes()); - let e2ee = enable_e2ee - .then_some(E2eeOptions { encryption_type: EncryptionType::Gcm, key_provider }); - - let mut options = RoomOptions::default(); - options.auto_subscribe = auto_subscribe; - options.e2ee = e2ee; - - let res = Room::connect(&url, &token, options).await; - - if let Ok((new_room, events)) = res { - log::info!("connected to room: {}", new_room.name()); - tokio::spawn(room_task(inner.clone(), events)); - - let new_room = Arc::new(new_room); - running_state = Some(RunningState { - room: new_room.clone(), - logo_track: LogoTrack::new(new_room.clone()), - sine_track: SineTrack::new(new_room.clone(), SineParameters::default()), - data_track: None, - }); - - // Allow direct access to the room from the UI (Used for sync access) - inner.room.lock().replace(new_room); - - let _ = inner.ui_tx.send(UiCmd::ConnectResult { result: Ok(()) }); - } else if let Err(err) = res { - log::error!("failed to connect to room: {:?}", err); - let _ = inner.ui_tx.send(UiCmd::ConnectResult { result: Err(err) }); - } - } - AsyncCmd::RoomDisconnect => { - if let Some(state) = running_state.take() { - *inner.room.lock() = None; - if let Err(err) = state.room.close().await { - log::error!("failed to disconnect from room: {:?}", err); - } - } - } - AsyncCmd::SimulateScenario { scenario } => { - if let Some(state) = running_state.as_ref() { - if let Err(err) = state.room.simulate_scenario(scenario).await { - log::error!("failed to simulate scenario: {:?}", err); - } - } - } - AsyncCmd::ToggleLogo => { - if let Some(state) = running_state.as_mut() { - if state.logo_track.is_published() { - state.logo_track.unpublish().await.unwrap(); - } else { - state.logo_track.publish().await.unwrap(); - } - } - } - AsyncCmd::ToggleSine => { - if let Some(state) = running_state.as_mut() { - if state.sine_track.is_published() { - state.sine_track.unpublish().await.unwrap(); - } else { - state.sine_track.publish().await.unwrap(); - } - } - } - AsyncCmd::ToggleDataTrack => { - if let Some(state) = running_state.as_mut() { - if let Some(track) = state.data_track.take() { - track.unpublish(); - let _ = inner.ui_tx.send(UiCmd::DataTrackUnpublished); - } else { - match state.room.local_participant().publish_data_track("slider").await { - Ok(track) => { - let _ = inner - .ui_tx - .send(UiCmd::DataTrackPublished { track: track.clone() }); - state.data_track = Some(track); - } - Err(err) => log::error!("failed to publish data track: {err}"), - } - } - } - } - AsyncCmd::SubscribeTrack { publication } => { - publication.set_subscribed(true); - } - AsyncCmd::UnsubscribeTrack { publication } => { - publication.set_subscribed(false); - } - AsyncCmd::SetVideoQuality { publication, quality } => { - publication.set_video_quality(quality); - } - AsyncCmd::E2eeKeyRatchet => { - if let Some(state) = running_state.as_ref() { - let e2ee_manager = state.room.e2ee_manager(); - if let Some(key_provider) = e2ee_manager.key_provider() { - key_provider.ratchet_shared_key(0); - } - } - } - AsyncCmd::RpcSendRequest { destination, method, payload, request_id } => { - if let Some(state) = running_state.as_ref() { - let local = state.room.local_participant(); - let ui_tx = inner.ui_tx.clone(); - tokio::spawn(async move { - let result = local - .perform_rpc( - PerformRpcData::new(destination, method).with_payload(payload), - ) - .await; - let _ = ui_tx.send(UiCmd::RpcSendResult { request_id, result }); - }); - } else { - let _ = inner.ui_tx.send(UiCmd::RpcSendResult { - request_id, - result: Err(RpcError { - code: RpcErrorCode::SendFailed as u32, - message: "Not connected".to_string(), - data: None, - }), - }); - } - } - AsyncCmd::LogStats => { - if let Some(state) = running_state.as_ref() { - for (_, publication) in state.room.local_participant().track_publications() { - if let Some(track) = publication.track() { - log::info!( - "track stats: LOCAL {:?} {:?}", - track.sid(), - track.get_stats().await, - ); - } - } - - for (_, participant) in state.room.remote_participants() { - for (_, publication) in participant.track_publications() { - if let Some(track) = publication.track() { - log::info!( - "track stats: {:?} {:?} {:?}", - participant.identity(), - track.sid(), - track.get_stats().await, - ); - } - } - } - } - } - } - } -} - -/// Task basically used to forward room events to the UI. -/// It will automatically close when the room is disconnected. -async fn room_task(inner: Arc, mut events: mpsc::UnboundedReceiver) { - while let Some(event) = events.recv().await { - let _ = inner.ui_tx.send(UiCmd::RoomEvent { event }); - } -} diff --git a/examples/wgpu_room/src/sine_track.rs b/examples/wgpu_room/src/sine_track.rs deleted file mode 100644 index 1825bb8a7..000000000 --- a/examples/wgpu_room/src/sine_track.rs +++ /dev/null @@ -1,132 +0,0 @@ -use livekit::options::TrackPublishOptions; -use livekit::webrtc::audio_frame::AudioFrame; -use livekit::webrtc::audio_source::RtcAudioSource; -use livekit::webrtc::prelude::AudioSourceOptions; -use livekit::{prelude::*, webrtc::audio_source::native::NativeAudioSource}; -use std::sync::Arc; -use tokio::sync::oneshot; -use tokio::task::JoinHandle; - -#[derive(Clone)] -pub struct SineParameters { - pub sample_rate: u32, - pub freq: f64, - pub amplitude: f64, - pub num_channels: u32, -} - -impl Default for SineParameters { - fn default() -> Self { - Self { sample_rate: 48000, freq: 440.0, amplitude: 1.0, num_channels: 2 } - } -} - -struct TrackHandle { - close_tx: oneshot::Sender<()>, - track: LocalAudioTrack, - task: JoinHandle<()>, -} - -pub struct SineTrack { - rtc_source: NativeAudioSource, - params: SineParameters, - room: Arc, - handle: Option, -} - -impl SineTrack { - pub fn new(room: Arc, params: SineParameters) -> Self { - Self { - rtc_source: NativeAudioSource::new( - AudioSourceOptions::default(), - params.sample_rate, - params.num_channels, - 1000, - ), - params, - room, - handle: None, - } - } - - pub fn is_published(&self) -> bool { - self.handle.is_some() - } - - pub async fn publish(&mut self) -> Result<(), RoomError> { - let (close_tx, close_rx) = oneshot::channel(); - let track = LocalAudioTrack::create_audio_track( - "sine_wave", - RtcAudioSource::Native(self.rtc_source.clone()), - ); - - let task = - tokio::spawn(Self::track_task(close_rx, self.rtc_source.clone(), self.params.clone())); - - self.room - .local_participant() - .publish_track( - LocalTrack::Audio(track.clone()), - TrackPublishOptions { source: TrackSource::Microphone, ..Default::default() }, - ) - .await?; - - let handle = TrackHandle { close_tx, track, task }; - - self.handle = Some(handle); - Ok(()) - } - - pub async fn unpublish(&mut self) -> Result<(), RoomError> { - if let Some(handle) = self.handle.take() { - handle.close_tx.send(()).ok(); - handle.task.await.ok(); - self.room.local_participant().unpublish_track(&handle.track.sid()).await?; - } - - Ok(()) - } - - async fn track_task( - mut close_rx: oneshot::Receiver<()>, - rtc_source: NativeAudioSource, - params: SineParameters, - ) { - let num_channels = params.num_channels as usize; - let samples_count = (params.sample_rate / 100) as usize * num_channels; - let mut samples_10ms = vec![0; samples_count]; - let mut phase = 0; - loop { - if close_rx.try_recv().is_ok() { - break; - } - - for i in (0..samples_count).step_by(num_channels) { - let val = params.amplitude - * f64::sin( - std::f64::consts::PI - * 2.0 - * params.freq - * (phase as f64 / params.sample_rate as f64), - ); - - phase += 1; - - for c in 0..num_channels { - // WebRTC uses 16-bit signed PCM - samples_10ms[i + c] = (val * 32768.0) as i16; - } - } - - rtc_source - .capture_frame(&AudioFrame { - data: samples_10ms.as_slice().into(), - sample_rate: params.sample_rate, - num_channels: params.num_channels, - samples_per_channel: samples_count as u32 / params.num_channels, - }) - .await - .unwrap(); - } - } -} diff --git a/examples/wgpu_room/src/video_grid.rs b/examples/wgpu_room/src/video_grid.rs deleted file mode 100644 index e1bec97e0..000000000 --- a/examples/wgpu_room/src/video_grid.rs +++ /dev/null @@ -1,178 +0,0 @@ -use std::cmp; - -#[derive(Debug, Clone, Default, PartialEq)] -struct State { - num_videos: u32, -} - -impl State { - pub fn load(ctx: &egui::Context, id: egui::Id) -> Option { - ctx.data(|i| i.get_temp(id)) - } - - pub fn store(self, ctx: &egui::Context, id: egui::Id) { - ctx.data_mut(|i| i.insert_temp(id, self)); - } -} - -pub const DEFAULT_VIDEO_SIZE: egui::Vec2 = egui::vec2(320.0 / 1.4, 180.0 / 1.4); -pub const DEFAULT_MAX_COLUMNS: u32 = 4; -pub const DEFAULT_SPACING: f32 = 8.0; - -pub struct VideoGrid { - id: egui::Id, - - // Current frame - available_rect: egui::Rect, - prev_state: State, - curr_state: State, - video_index: u32, // Kinda "cursor" - - // Options - min_video_size: egui::Vec2, - max_columns: u32, - spacing: f32, -} - -impl VideoGrid { - pub fn new(id_source: impl std::hash::Hash + std::fmt::Debug) -> Self { - Self { - id: egui::Id::new(id_source), - available_rect: egui::Rect::NAN, - prev_state: State::default(), - curr_state: State::default(), - video_index: 0, - min_video_size: DEFAULT_VIDEO_SIZE, - max_columns: DEFAULT_MAX_COLUMNS, - spacing: DEFAULT_SPACING, - } - } - - pub fn show( - mut self, - ui: &mut egui::Ui, - grid: impl FnOnce(&mut VideoGridContext) -> R, - ) -> egui::InnerResponse { - // TODO(theomonnom): Should I care about the current egui layout? - - let prev_state = State::load(ui.ctx(), self.id); - let is_first_frame = prev_state.is_none(); - - self.prev_state = prev_state.unwrap_or_default(); - self.available_rect = ui.available_rect_before_wrap(); - - ui.ctx().check_for_id_clash(self.id, self.available_rect, "VideoGrid"); - - ui.scope_builder(egui::UiBuilder::new().max_rect(self.available_rect), |ui| { - if is_first_frame { - ui.set_invisible(); - } - - let mut ctx = VideoGridContext { layout: &mut self, ui }; - let res = grid(&mut ctx); - - // Save the new state - if self.curr_state != self.prev_state { - self.curr_state.clone().store(ui.ctx(), self.id); - ui.ctx().request_repaint(); - } - - res - }) - } - - fn next_frame_rect(&mut self) -> egui::Rect { - assert!(self.available_rect.is_finite()); - assert!(self.spacing <= self.min_video_size.x); - - // increment the amount of videos for the next frame - self.curr_state.num_videos += 1; - - let num_videos = self.prev_state.num_videos; - if num_videos == 0 { - return egui::Rect::NOTHING; - } - - let max_columns = self.max_columns; - let minimum_size = self.min_video_size; - let available_size = self.available_rect.size(); - - let calc_min_width = - |columns: u32| columns as f32 * minimum_size.x + (columns - 1) as f32 * self.spacing; - - let total_columns = { - let mut est = (available_size.x / minimum_size.x) as u32 + 1; - if available_size.x < calc_min_width(est) { - est -= 1; - } - cmp::max(1, cmp::min(est, max_columns)) - }; - - let aspect_ratio = minimum_size.x / minimum_size.y; - let remaining_width = available_size.x - calc_min_width(total_columns); - let w = minimum_size.x + remaining_width / total_columns as f32; - let h = w / aspect_ratio; - - let x_index = self.video_index % total_columns; - let y_index = self.video_index / total_columns; - - let x = { - let mut x = x_index as f32 * (w + self.spacing); - - // vertically center the last row - let total_rows = num_videos / total_columns + 1; - if (y_index + 1) == total_rows { - let nb_items = num_videos - (total_rows - 1) * total_columns; // nb. of items on the last row - x += (total_columns - nb_items) as f32 * (w + self.spacing) / 2.0; - } - - x - }; - let y = y_index as f32 * (h + self.spacing); - - let min = egui::pos2(x, y) + self.available_rect.left_top().to_vec2(); - let max = egui::pos2(w, h) + min.to_vec2(); - - self.video_index += 1; - - egui::Rect { min, max } - } -} - -#[allow(dead_code)] -impl VideoGrid { - pub fn min_video_size(mut self, min_video_size: egui::Vec2) -> Self { - self.min_video_size = min_video_size; - self - } - - pub fn max_columns(mut self, max_columns: u32) -> Self { - self.max_columns = max_columns; - self - } - - pub fn spacing(mut self, spacing: f32) -> Self { - self.spacing = spacing; - self - } -} - -pub struct VideoGridContext<'a> { - layout: &'a mut VideoGrid, - ui: &'a mut egui::Ui, -} - -impl<'a> VideoGridContext<'a> { - pub fn video_frame(&mut self, add_contents: impl FnOnce(&mut egui::Ui)) -> egui::Response { - let frame_rect = self.layout.next_frame_rect(); - - if self.ui.is_visible() { - let mut child_ui = self.ui.new_child( - egui::UiBuilder::new().max_rect(frame_rect).layout(egui::Layout::default()), - ); - add_contents(&mut child_ui); - } - - self.ui.allocate_rect(frame_rect, egui::Sense::hover()) - } -} diff --git a/examples/wgpu_room/src/video_renderer.rs b/examples/wgpu_room/src/video_renderer.rs deleted file mode 100644 index 5fc4b5f77..000000000 --- a/examples/wgpu_room/src/video_renderer.rs +++ /dev/null @@ -1,160 +0,0 @@ -use futures::StreamExt; -use livekit::webrtc::native::yuv_helper; -use livekit::webrtc::prelude::*; -use livekit::webrtc::video_stream::native::NativeVideoStream; -use parking_lot::Mutex; -use std::{ops::DerefMut, sync::Arc}; - -pub struct VideoRenderer { - internal: Arc>, - - #[allow(dead_code)] - rtc_track: RtcVideoTrack, -} - -struct RendererInternal { - render_state: egui_wgpu::RenderState, - width: u32, - height: u32, - rgba_data: Vec, - texture: Option, - texture_view: Option, - egui_texture: Option, -} - -impl VideoRenderer { - pub fn new( - async_handle: &tokio::runtime::Handle, - render_state: egui_wgpu::RenderState, - rtc_track: RtcVideoTrack, - ) -> Self { - let internal = Arc::new(Mutex::new(RendererInternal { - render_state, - width: 0, - height: 0, - rgba_data: Vec::default(), - texture: None, - texture_view: None, - egui_texture: None, - })); - - // TODO(theomonnom) Gracefully close the thread - let mut video_sink = NativeVideoStream::new(rtc_track.clone()); - - std::thread::spawn({ - let async_handle = async_handle.clone(); - let internal = internal.clone(); - move || { - while let Some(frame) = async_handle.block_on(video_sink.next()) { - // Process the frame - let mut internal = internal.lock(); - let buffer = frame.buffer.to_i420(); - - let width: u32 = buffer.width(); - let height: u32 = buffer.height(); - - internal.ensure_texture_size(width, height); - - let rgba_ptr = internal.rgba_data.deref_mut(); - let rgba_stride = buffer.width() * 4; - - let (stride_y, stride_u, stride_v) = buffer.strides(); - let (data_y, data_u, data_v) = buffer.data(); - - yuv_helper::i420_to_abgr( - data_y, - stride_y, - data_u, - stride_u, - data_v, - stride_v, - rgba_ptr, - rgba_stride, - buffer.width() as i32, - buffer.height() as i32, - ); - - internal.render_state.queue.write_texture( - eframe::wgpu::TexelCopyTextureInfo { - texture: internal.texture.as_ref().unwrap(), - mip_level: 0, - origin: eframe::wgpu::Origin3d::default(), - aspect: eframe::wgpu::TextureAspect::default(), - }, - &internal.rgba_data, - eframe::wgpu::TexelCopyBufferLayout { - bytes_per_row: Some(width * 4), - ..Default::default() - }, - eframe::wgpu::Extent3d { width, height, ..Default::default() }, - ); - } - } - }); - - Self { rtc_track, internal } - } - - // Returns the last frame resolution - pub fn resolution(&self) -> (u32, u32) { - let internal = self.internal.lock(); - (internal.width, internal.height) - } - - // Returns the texture id, can be used to draw the texture on the UI - pub fn texture_id(&self) -> Option { - self.internal.lock().egui_texture - } -} - -impl RendererInternal { - fn ensure_texture_size(&mut self, width: u32, height: u32) { - if self.width == width && self.height == height { - return; - } - - self.width = width; - self.height = height; - self.rgba_data.resize((width * height * 4) as usize, 0); - - self.texture = - Some(self.render_state.device.create_texture(&eframe::wgpu::TextureDescriptor { - label: Some("lk-videotexture"), - usage: eframe::wgpu::TextureUsages::TEXTURE_BINDING - | eframe::wgpu::TextureUsages::COPY_DST, - dimension: eframe::wgpu::TextureDimension::D2, - size: eframe::wgpu::Extent3d { width, height, ..Default::default() }, - sample_count: 1, - mip_level_count: 1, - format: eframe::wgpu::TextureFormat::Rgba8Unorm, - view_formats: &[eframe::wgpu::TextureFormat::Rgba8Unorm], - })); - - self.texture_view = Some(self.texture.as_mut().unwrap().create_view( - &eframe::wgpu::TextureViewDescriptor { - label: Some("lk-videotexture-view"), - format: Some(eframe::wgpu::TextureFormat::Rgba8Unorm), - dimension: Some(eframe::wgpu::TextureViewDimension::D2), - mip_level_count: Some(1), - array_layer_count: Some(1), - ..Default::default() - }, - )); - - if let Some(texture_id) = self.egui_texture { - // Update the existing texture - self.render_state.renderer.write().update_egui_texture_from_wgpu_texture( - &self.render_state.device, - self.texture_view.as_ref().unwrap(), - eframe::wgpu::FilterMode::Linear, - texture_id, - ); - } else { - self.egui_texture = Some(self.render_state.renderer.write().register_native_texture( - &self.render_state.device, - self.texture_view.as_ref().unwrap(), - eframe::wgpu::FilterMode::Linear, - )); - } - } -}