From c7c82cc69336f45f2555932bc7d82bede826f6e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Thu, 29 Jun 2023 22:52:43 +0200 Subject: [PATCH] use callbacks on internal events (#100) --- Cargo.lock | 70 +++- examples/Cargo.lock | 33 +- examples/wgpu_room/Cargo.toml | 9 +- examples/wgpu_room/src/app.rs | 137 +++++--- examples/wgpu_room/src/logo_track.rs | 2 +- livekit-ffi/Cargo.toml | 2 +- livekit-ffi/src/lib.rs | 7 +- livekit-ffi/src/server/audio_frame.rs | 10 +- livekit-ffi/src/server/mod.rs | 84 +++-- livekit-ffi/src/server/room.rs | 65 ++-- livekit-ffi/src/server/tests.rs | 2 +- livekit-ffi/src/server/video_frame.rs | 10 +- livekit/src/lib.rs | 2 - livekit/src/prelude.rs | 4 +- livekit/src/proto.rs | 12 +- livekit/src/room/mod.rs | 185 ++++------ .../src/room/participant/local_participant.rs | 160 +++++---- livekit/src/room/participant/mod.rs | 283 +++++++-------- .../room/participant/remote_participant.rs | 282 ++++++++++----- livekit/src/room/publication/local.rs | 121 +++---- livekit/src/room/publication/mod.rs | 278 ++++++--------- livekit/src/room/publication/remote.rs | 273 ++++++++++----- livekit/src/room/track/audio_track.rs | 43 +++ livekit/src/room/track/local_audio_track.rs | 168 ++++----- livekit/src/room/track/local_track.rs | 28 +- livekit/src/room/track/local_video_track.rs | 168 ++++----- livekit/src/room/track/mod.rs | 327 +++++------------- livekit/src/room/track/remote_audio_track.rs | 62 ++-- livekit/src/room/track/remote_track.rs | 30 +- livekit/src/room/track/remote_video_track.rs | 62 ++-- livekit/src/room/track/video_track.rs | 44 +++ livekit/src/rtc_engine/rtc_session.rs | 5 +- webrtc-sys/src/webrtc.cpp | 1 - 33 files changed, 1522 insertions(+), 1447 deletions(-) create mode 100644 livekit/src/room/track/audio_track.rs create mode 100644 livekit/src/room/track/video_track.rs diff --git a/Cargo.lock b/Cargo.lock index a55de55..fd10881 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,15 @@ # It is not intended for manual editing. version = 3 +[[package]] +name = "addr2line" +version = "0.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a76fd60b23679b7d19bd066031410fb7e458ccc5e958eb5c325888ce4baedc97" +dependencies = [ + "gimli", +] + [[package]] name = "adler" version = "1.0.2" @@ -40,6 +49,21 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa" +[[package]] +name = "backtrace" +version = "0.3.67" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "233d376d6d185f2a3093e58f283f60f880315b6c60075b01f36b3b85154564ca" +dependencies = [ + "addr2line", + "cc", + "cfg-if", + "libc", + "miniz_oxide 0.6.2", + "object", + "rustc-demangle", +] + [[package]] name = "base64" version = "0.21.2" @@ -357,7 +381,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3b9429470923de8e8cbd4d2dc513535400b4b3fef0319fb5c4e1f520a7bef743" dependencies = [ "crc32fast", - "miniz_oxide", + "miniz_oxide 0.7.1", ] [[package]] @@ -500,6 +524,12 @@ dependencies = [ "wasi", ] +[[package]] +name = "gimli" +version = "0.27.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c80984affa11d98d1b88b66ac8853f143217b399d3c74116778ff8fdb4ed2e" + [[package]] name = "glob" version = "0.3.1" @@ -908,6 +938,15 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "miniz_oxide" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b275950c28b37e794e8c55d88aeb5e139d0ce23fdbbeda68f8d7174abdf9e8fa" +dependencies = [ + "adler", +] + [[package]] name = "miniz_oxide" version = "0.7.1" @@ -962,6 +1001,15 @@ dependencies = [ "libc", ] +[[package]] +name = "object" +version = "0.30.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03b4680b86d9cfafba8fc491dc9b6df26b68cf40e9e6cd73909194759a63c385" +dependencies = [ + "memchr", +] + [[package]] name = "once_cell" version = "1.18.0" @@ -1038,10 +1086,13 @@ version = "0.9.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "93f00c865fe7cabf650081affecd3871070f26767e7b2070a3ffae14c654b447" dependencies = [ + "backtrace", "cfg-if", "libc", + "petgraph", "redox_syscall 0.3.5", "smallvec", + "thread-id", "windows-targets", ] @@ -1313,6 +1364,12 @@ dependencies = [ "winapi", ] +[[package]] +name = "rustc-demangle" +version = "0.1.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d626bb9dae77e28219937af045c257c28bfd3f69333c512553507f5f9798cb76" + [[package]] name = "rustix" version = "0.37.19" @@ -1617,6 +1674,17 @@ dependencies = [ "syn 2.0.18", ] +[[package]] +name = "thread-id" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ee93aa2b8331c0fec9091548843f2c90019571814057da3b783f9de09349d73" +dependencies = [ + "libc", + "redox_syscall 0.2.16", + "winapi", +] + [[package]] name = "time" version = "0.3.22" diff --git a/examples/Cargo.lock b/examples/Cargo.lock index 15c35b1..5f311a9 100644 --- a/examples/Cargo.lock +++ b/examples/Cargo.lock @@ -641,7 +641,8 @@ checksum = "9ea835d29036a4087793836fa931b08837ad5e957da9e23886b29586fb9b6650" [[package]] name = "ecolor" version = "0.22.0" -source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e479a7fa3f23d4e794f8b2f8b3568dd4e47886ad1b12c9c095e141cb591eb63" dependencies = [ "bytemuck", ] @@ -649,7 +650,8 @@ dependencies = [ [[package]] name = "egui" version = "0.22.0" -source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3aef8ec3ae1b772f340170c65bf27d5b8c28f543a0116c844d2ac08d01123e7" dependencies = [ "ahash 0.8.3", "epaint", @@ -660,7 +662,8 @@ dependencies = [ [[package]] name = "egui-wgpu" version = "0.22.0" -source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33caaedd8283779c787298af23d8754a7e88421ff32e89ad0040c855fc0b0224" dependencies = [ "bytemuck", "epaint", @@ -674,7 +677,8 @@ dependencies = [ [[package]] name = "egui-winit" version = "0.22.0" -source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a49155fd4a0a4fb21224407a91de0030847972ef90fc64edb63621caea61cb2" dependencies = [ "arboard", "egui", @@ -695,7 +699,8 @@ checksum = "7fcaabb2fef8c910e7f4c7ce9f67a1283a1715879a7c230ca9d6d1ae31f16d91" [[package]] name = "emath" version = "0.22.0" -source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3857d743a6e0741cdd60b622a74c7a36ea75f5f8f11b793b41d905d2c9721a4b" dependencies = [ "bytemuck", ] @@ -725,7 +730,8 @@ dependencies = [ [[package]] name = "epaint" version = "0.22.0" -source = "git+https://github.com/emilk/egui#307565efa55158cfa6b82d2e8fdc4c4914b954ed" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09333964d4d57f40a85338ba3ca5ed4716070ab184dcfed966b35491c5c64f3b" dependencies = [ "ab_glyph", "ahash 0.8.3", @@ -2016,10 +2022,13 @@ version = "0.9.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9069cbb9f99e3a5083476ccb29ceb1de18b9118cafa53e90c9551235de2b9521" dependencies = [ + "backtrace", "cfg-if", "libc", + "petgraph", "redox_syscall 0.2.16", "smallvec", + "thread-id", "windows-sys 0.45.0", ] @@ -2753,6 +2762,17 @@ dependencies = [ "syn 2.0.16", ] +[[package]] +name = "thread-id" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ee93aa2b8331c0fec9091548843f2c90019571814057da3b783f9de09349d73" +dependencies = [ + "libc", + "redox_syscall 0.2.16", + "winapi", +] + [[package]] name = "thread_local" version = "1.1.7" @@ -3420,6 +3440,7 @@ dependencies = [ "futures", "image", "livekit", + "log", "parking_lot", "tokio", "tracing", diff --git a/examples/wgpu_room/Cargo.toml b/examples/wgpu_room/Cargo.toml index 596d9d4..221e3b3 100644 --- a/examples/wgpu_room/Cargo.toml +++ b/examples/wgpu_room/Cargo.toml @@ -11,8 +11,9 @@ livekit = { path = "../../livekit", version = "0.1.1" } futures = "0.3" wgpu = "0.16" winit = "0.28" -parking_lot = { version = "0.12.1", features=["send_guard"] } +parking_lot = { version = "0.12.1", features=["deadlock_detection"] } image = "0.24" -egui = { git = "https://github.com/emilk/egui" } -egui-wgpu = { git = "https://github.com/emilk/egui", features = ["winit"] } -egui-winit = { git = "https://github.com/emilk/egui" } +egui = "0.22" +egui-wgpu = { version = "0.22", features = ["winit"] } +egui-winit = "0.22" +log = "0.4" \ No newline at end of file diff --git a/examples/wgpu_room/src/app.rs b/examples/wgpu_room/src/app.rs index 1cd36b8..1f12cf5 100644 --- a/examples/wgpu_room/src/app.rs +++ b/examples/wgpu_room/src/app.rs @@ -9,37 +9,38 @@ use futures::StreamExt; use livekit::prelude::*; use livekit::webrtc::audio_stream::native::NativeAudioStream; use livekit::SimulateScenario; +use parking_lot::deadlock; use parking_lot::Mutex; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::sync::{ atomic::{AtomicBool, Ordering}, Arc, }; - +use std::thread; +use std::time::Duration; +use tokio::sync::Mutex as AsyncMutex; use tokio::sync::{mpsc, oneshot}; // Useful default constants for developing const DEFAULT_URL: &str = "ws://localhost:7880"; const DEFAULT_TOKEN : &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5MDY2MTMyODgsImlzcyI6IkFQSVRzRWZpZFpqclFvWSIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NzI2MTMyODgsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb20iOiJ0ZXN0Iiwicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZSwicm9vbUxpc3QiOnRydWV9fQ.uSNIangMRu8jZD5mnRYoCHjcsQWCrJXgHCs0aNIgBFY"; -// eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5MDY2MTM0MzcsImlzcyI6IkFQSVRzRWZpZFpqclFvWSIsIm5hbWUiOiJ3ZWIiLCJuYmYiOjE2NzI2MTM0MzcsInN1YiI6IndlYiIsInZpZGVvIjp7InJvb20iOiJ0ZXN0Iiwicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZSwicm9vbUxpc3QiOnRydWV9fQ.DFTXt60n1kzGq4cSuOhbFBTQW2nd3rlcXKQ54sXsP8s - use winit::{ event::*, event_loop::{ControlFlow, EventLoop}, window::{WindowBuilder, WindowId}, }; -struct Session { +struct SessionHandle { room: Arc, - logo_track: LogoTrack, - sine_track: SineTrack, + logo_track: Arc>, + sine_track: Arc>, close_tx: oneshot::Sender<()>, - handle: tokio::task::JoinHandle<()>, + task_handle: tokio::task::JoinHandle<()>, } struct AppState { - session: Mutex>, + session: Mutex>, connecting: AtomicBool, } @@ -54,11 +55,14 @@ struct App { cmd_tx: mpsc::UnboundedSender, cmd_rx: mpsc::UnboundedReceiver, - // UI State + // Ui State lk_url: String, lk_token: String, connection_failure: Option, room_state: ConnectionState, + + // Log events + events: Vec, } pub fn run(rt: tokio::runtime::Runtime) { @@ -97,8 +101,26 @@ pub fn run(rt: tokio::runtime::Runtime) { lk_token: DEFAULT_TOKEN.to_owned(), connection_failure: None, room_state: ConnectionState::Connected, + events: Vec::new(), }; + // 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()); + } + } + }); + // Async event loop tokio::spawn(async move { while let Some(event) = async_cmd_rx.recv().await { @@ -112,19 +134,19 @@ pub fn run(rt: tokio::runtime::Runtime) { let (close_tx, close_rx) = oneshot::channel(); let logo_track = LogoTrack::new(room.clone()); let sine_track = SineTrack::new(room.clone()); - let handle = tokio::spawn(room_task( + let task_handle = tokio::spawn(room_task( state.clone(), room_events, close_rx, ui_cmd_tx.clone(), )); - *state.session.lock() = Some(Session { + *state.session.lock() = Some(SessionHandle { room, - logo_track, - sine_track, + logo_track: Arc::new(AsyncMutex::new(logo_track)), + sine_track: Arc::new(AsyncMutex::new(sine_track)), close_tx, - handle, + task_handle, }); let _ = ui_cmd_tx.send(UiCmd::ConnectResult { result: Ok(()) }); @@ -135,20 +157,37 @@ pub fn run(rt: tokio::runtime::Runtime) { state.connecting.store(false, Ordering::SeqCst); } AsyncCmd::RoomDisconnect => { - if let Some(session) = state.session.lock().take() { - let _ = session.room.close().await; - let _ = session.close_tx.send(()); - let _ = session.handle.await; + let handle = state.session.lock().take(); + if let Some(handle) = handle { + let _ = handle.room.close().await; + let _ = handle.close_tx.send(()); + let _ = handle.task_handle.await; } } AsyncCmd::SimulateScenario { scenario } => { - if let Some(session) = state.session.lock().as_ref() { - let _ = session.room.simulate_scenario(scenario).await; + let room = { + state + .session + .lock() + .as_ref() + .map(|handle| handle.room.clone()) + }; + + if let Some(room) = room { + let _ = room.simulate_scenario(scenario).await; } } AsyncCmd::ToggleLogo => { - if let Some(session) = state.session.lock().as_mut() { - let logo_track = &mut session.logo_track; + let logo_track = { + state + .session + .lock() + .as_ref() + .map(|handle| handle.logo_track.clone()) + }; + + if let Some(logo_track) = logo_track { + let mut logo_track = logo_track.lock().await; if !logo_track.is_published() { logo_track.publish().await.unwrap(); } else { @@ -157,8 +196,16 @@ pub fn run(rt: tokio::runtime::Runtime) { } } AsyncCmd::ToggleSine => { - if let Some(session) = state.session.lock().as_mut() { - let sine_track = &mut session.sine_track; + let sine_track = { + state + .session + .lock() + .as_ref() + .map(|handle| handle.sine_track.clone()) + }; + + if let Some(sine_track) = sine_track { + let mut sine_track = sine_track.lock().await; if !sine_track.is_published() { sine_track.publish().await.unwrap(); } else { @@ -209,6 +256,8 @@ impl App { } } UiCmd::RoomEvent { event } => { + self.events.push(format!("{:?}", event)); + match event { RoomEvent::TrackSubscribed { track, participant, .. @@ -229,7 +278,7 @@ impl App { NativeAudioStream::new(audio_track.rtc_track()); while let Some(_frame) = stream.next().await { - // Received audio frames + // TODO(theomonnom): Play audio using the libwebrtc ADM playout devices } }); } @@ -295,12 +344,10 @@ impl App { fn ui(&mut self, ui: &mut egui::Ui) { egui::TopBottomPanel::top("top_panel").show(ui.ctx(), |ui| { egui::menu::bar(ui, |ui| { - ui.menu_button("Tools", |ui| { - if ui.button("Logs").clicked() {} - if ui.button("Profiler").clicked() {} - if ui.button("WebRTC Stats").clicked() {} - if ui.button("Events").clicked() {} - }); + ui.menu_button( + "Tools", + |ui| if ui.button("WebRTC Stats (TODO)").clicked() {}, + ); ui.menu_button("Simulate", |ui| { if ui.button("SignalReconnect").clicked() { let _ = self.cmd_tx.send(AsyncCmd::SimulateScenario { @@ -399,21 +446,33 @@ impl App { { // Room Info - if let Some(session) = self.state.session.lock().as_ref() { - ui.label(format!("Name: {}", session.room.name())); - ui.label(format!("SID: {}", session.room.sid())); + if let Some(handle) = self.state.session.lock().as_ref() { + ui.label(format!("Name: {}", handle.room.name())); + ui.label(format!("SID: {}", handle.room.sid())); ui.label(format!( "ConnectionState: {:?}", - session.room.connection_state() + handle.room.connection_state() )); ui.label(format!( "ParticipantCount: {:?}", - session.room.participants().len() + 1 + handle.room.participants().len() + 1 )); } } }); + egui::TopBottomPanel::bottom("bottom_panel") + .default_height(128.0) + .show(ui.ctx(), |ui| { + ui.label("Events"); + + egui::ScrollArea::vertical().show(ui, |ui| { + for event in &self.events { + ui.label(event); + } + }); + }); + egui::CentralPanel::default().show(ui.ctx(), |ui| { egui::ScrollArea::vertical().show(ui, |ui| { VideoGrid::new("default_grid") @@ -455,8 +514,8 @@ impl App { } let name = - self.state.session.lock().as_ref().and_then(|session| { - session + self.state.session.lock().as_ref().and_then(|handle| { + handle .room .participants() .get(participant_sid) diff --git a/examples/wgpu_room/src/logo_track.rs b/examples/wgpu_room/src/logo_track.rs index f53b198..10c6e6e 100644 --- a/examples/wgpu_room/src/logo_track.rs +++ b/examples/wgpu_room/src/logo_track.rs @@ -72,7 +72,7 @@ impl LogoTrack { .publish_track( LocalTrack::Video(track.clone()), TrackPublishOptions { - source: TrackSource::Camera, + source: TrackSource::Camera, // Using camera so it is shown on everyone's screen (Unknown is sometimes not shown) ..Default::default() }, ) diff --git a/livekit-ffi/Cargo.toml b/livekit-ffi/Cargo.toml index 2ed4fbb..4398720 100644 --- a/livekit-ffi/Cargo.toml +++ b/livekit-ffi/Cargo.toml @@ -19,7 +19,7 @@ livekit = { path = "../livekit", version = "0.1.1" } livekit-protocol = { path = "../livekit-protocol", version = "0.1.0" } tokio = { version = "1.0", features = ["full"] } futures-util = { version = "0.3", default-features = false, features = ["sink"] } -parking_lot = { version = "0.12.1", features=["send_guard"] } +parking_lot = { version = "0.12.1", features=["deadlock_detection"] } prost = "0.11.0" prost-types = "0.11.1" lazy_static = "1.4.0" diff --git a/livekit-ffi/src/lib.rs b/livekit-ffi/src/lib.rs index 7843859..ee98d62 100644 --- a/livekit-ffi/src/lib.rs +++ b/livekit-ffi/src/lib.rs @@ -62,7 +62,7 @@ pub extern "C" fn livekit_ffi_request( let handle_id = server::FFI_SERVER.next_id(); server::FFI_SERVER - .ffi_handles() + .ffi_handles .insert(handle_id, Box::new(res)); handle_id @@ -71,8 +71,5 @@ pub extern "C" fn livekit_ffi_request( #[no_mangle] pub extern "C" fn livekit_ffi_drop_handle(handle_id: FfiHandleId) -> bool { // Free the memory - server::FFI_SERVER - .ffi_handles() - .remove(&handle_id) - .is_some() + server::FFI_SERVER.ffi_handles.remove(&handle_id).is_some() } diff --git a/livekit-ffi/src/server/audio_frame.rs b/livekit-ffi/src/server/audio_frame.rs index eb8a32a..82c436b 100644 --- a/livekit-ffi/src/server/audio_frame.rs +++ b/livekit-ffi/src/server/audio_frame.rs @@ -39,7 +39,7 @@ impl FfiAudioSream { .id as FfiHandleId; let track = server - .ffi_handles() + .ffi_handles .get(&handle_id) .ok_or(FfiError::InvalidRequest("track not found"))?; @@ -77,7 +77,7 @@ impl FfiAudioSream { // Store the new audio stream and return the info let info = proto::AudioStreamInfo::from(&audio_stream); server - .ffi_handles() + .ffi_handles .insert(audio_stream.handle_id, Box::new(audio_stream)); Ok(info) @@ -110,7 +110,7 @@ impl FfiAudioSream { let handle_id = server.next_id(); let buffer_info = proto::AudioFrameBufferInfo::from(handle_id, &frame); - server.ffi_handles().insert(handle_id, Box::new(frame)); + server.ffi_handles.insert(handle_id, Box::new(frame)); if let Err(err) = server.send_event(proto::ffi_event::Message::AudioStreamEvent( proto::AudioStreamEvent { @@ -164,7 +164,7 @@ impl FfiAudioSource { let source_info = proto::AudioSourceInfo::from(&audio_source); server - .ffi_handles() + .ffi_handles .insert(audio_source.handle_id, Box::new(audio_source)); Ok(source_info) @@ -184,7 +184,7 @@ impl FfiAudioSource { .id as FfiHandleId; let frame = server - .ffi_handles() + .ffi_handles .get(&buffer_handle) .ok_or(FfiError::InvalidRequest("handle not found"))?; diff --git a/livekit-ffi/src/server/mod.rs b/livekit-ffi/src/server/mod.rs index f86aac8..ba17699 100644 --- a/livekit-ffi/src/server/mod.rs +++ b/livekit-ffi/src/server/mod.rs @@ -8,7 +8,6 @@ use livekit::webrtc::prelude::*; use livekit::webrtc::video_frame::{native::I420BufferExt, BoxVideoFrameBuffer, I420Buffer}; use parking_lot::Mutex; use prost::Message; -use std::collections::HashMap; use std::slice; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; @@ -29,12 +28,12 @@ pub struct FfiConfig { } pub struct FfiServer { - rooms: Mutex>, /// Store all Ffi handles inside an HashMap, if this isn't efficient enough /// We can still use Box::into_raw & Box::from_raw in the future (but keep it safe for now) - ffi_handles: DashMap, + pub ffi_handles: DashMap, + pub async_runtime: tokio::runtime::Runtime, + next_id: AtomicUsize, - async_runtime: tokio::runtime::Runtime, config: Mutex>, } @@ -42,8 +41,30 @@ impl Default for FfiServer { fn default() -> Self { env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init(); + // Create a background thread which checks for deadlocks every 10s + { + use parking_lot::deadlock; + use std::thread; + use std::time::Duration; + + 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()); + } + } + }); + } + Self { - rooms: Default::default(), ffi_handles: Default::default(), next_id: AtomicUsize::new(1), // 0 is invalid async_runtime: tokio::runtime::Builder::new_multi_thread() @@ -60,13 +81,19 @@ impl Default for FfiServer { impl FfiServer { pub async fn dispose(&'static self) { // Close all rooms - for (_, room_handle) in self.rooms.lock().drain() { - let room = self.ffi_handles.remove(&room_handle); - if let Some(room) = room { - let ffi_room = room.1.downcast::().unwrap(); - ffi_room.close().await; + log::info!("disposing the FfiServer, closing all rooms..."); + + let mut rooms = Vec::new(); + for handle in self.ffi_handles.iter_mut() { + if let Some(handle) = handle.value().downcast_ref::() { + rooms.push(handle.clone()); } } + + for room in rooms { + room.close().await; + } + // Drop all handles self.ffi_handles.clear(); @@ -78,14 +105,6 @@ impl FfiServer { self.next_id.fetch_add(1, Ordering::Relaxed) } - pub fn ffi_handles(&'static self) -> &DashMap { - &self.ffi_handles - } - - pub fn rooms(&'static self) -> &Mutex> { - &self.rooms - } - pub fn send_event(&'static self, message: proto::ffi_event::Message) -> FfiResult<()> { let callback_fn = self .config @@ -128,14 +147,13 @@ impl FfiServer { ) -> FfiResult { *self.config.lock() = None; - let close = self.dispose(); if !dispose.r#async { - self.async_runtime.block_on(close); + self.async_runtime.block_on(self.dispose()); Ok(proto::DisposeResponse::default()) } else { let async_id = self.next_id(); self.async_runtime.spawn(async move { - close.await; + self.dispose().await; }); Ok(proto::DisposeResponse { async_id: Some(proto::FfiAsyncId { @@ -197,17 +215,19 @@ impl FfiServer { .ok_or(FfiError::InvalidRequest("room_handle is empty"))? .id as FfiHandleId; - let ffi_room = self - .ffi_handles - .remove(&room_handle) - .ok_or(FfiError::InvalidRequest("room not found"))? - .1; - - let ffi_room = ffi_room - .downcast::() - .map_err(|_| FfiError::InvalidRequest("room is not a FfiRoom"))?; - self.async_runtime.spawn(async move { + let mut ffi_room = self + .ffi_handles + .get_mut(&room_handle) + .ok_or(FfiError::InvalidRequest("room not found")) + .unwrap(); + + let ffi_room = ffi_room + .value_mut() + .downcast_mut::() + .ok_or(FfiError::InvalidRequest("room is not a FfiRoom")) + .unwrap(); + ffi_room.close().await; let _ = self.send_event(proto::ffi_event::Message::Disconnect( proto::DisconnectCallback { @@ -313,7 +333,7 @@ impl FfiServer { .ok_or(FfiError::InvalidRequest("room not found"))?; let ffi_room = ffi_room - .downcast_ref::() + .downcast_ref::() .ok_or(FfiError::InvalidRequest("room is not a FfiRoom"))?; // Push the data to an async queue (avoid blocking and keep the order) diff --git a/livekit-ffi/src/server/room.rs b/livekit-ffi/src/server/room.rs index 68374a7..a204a05 100644 --- a/livekit-ffi/src/server/room.rs +++ b/livekit-ffi/src/server/room.rs @@ -1,11 +1,14 @@ use crate::server::FfiServer; use crate::{proto, FfiAsyncId, FfiError, FfiHandleId, FfiResult}; use livekit::prelude::*; +use parking_lot::Mutex; use std::slice; use std::sync::Arc; use tokio::sync::{broadcast, mpsc}; use tokio::task::JoinHandle; +pub type HandleType = Arc; + struct DataPacket { data: Vec, kind: DataPacketKind, @@ -13,11 +16,15 @@ struct DataPacket { async_id: FfiAsyncId, } -pub struct FfiRoom { - room: Arc, +struct Handle { event_handle: JoinHandle<()>, data_handle: JoinHandle<()>, close_tx: broadcast::Sender<()>, +} + +pub struct FfiRoom { + room: Arc, + handle: Mutex>, data_tx: mpsc::UnboundedSender, } @@ -49,16 +56,17 @@ impl FfiRoom { .async_runtime .spawn(data_task(server, room.clone(), data_rx, close_rx)); - let ffi_room = Self { + let ffi_room = Arc::new(Self { room: room.clone(), - event_handle, - data_handle, - close_tx, + handle: Mutex::new(Some(Handle { + event_handle, + data_handle, + close_tx, + })), data_tx, - }; + }); - server.ffi_handles().insert(next_id, Box::new(ffi_room)); - server.rooms().lock().insert(room.sid(), next_id); + server.ffi_handles.insert(next_id, Box::new(ffi_room)); let room_info = proto::RoomInfo::from_room(next_id, &room); Ok(room_info) @@ -92,11 +100,15 @@ impl FfiRoom { }) } - pub async fn close(self) { + pub async fn close(&self) { let _ = self.room.close().await; - let _ = self.close_tx.send(()); - let _ = self.event_handle.await; - let _ = self.data_handle.await; + + let handle = self.handle.lock().take(); + if let Some(handle) = handle { + let _ = handle.close_tx.send(()); + let _ = handle.event_handle.await; + let _ = handle.data_handle.await; + } } pub fn room(&self) -> &Arc { @@ -135,23 +147,16 @@ async fn data_task( async fn room_task( server: &'static FfiServer, - room: Arc, + _room: Arc, room_handle: FfiHandleId, mut events: mpsc::UnboundedReceiver, mut close_rx: broadcast::Receiver<()>, ) { - server - .async_runtime - .spawn(participant_task(Participant::Local( - room.local_participant(), - ))); - loop { tokio::select! { Some(event) = events.recv() => { - let message = match event { + if let Some(message)= match event { RoomEvent::ParticipantConnected(participant) => { - server.async_runtime.spawn(participant_task(Participant::Remote(participant.clone()))); Some(proto::room_event::Message::ParticipantConnected( proto::ParticipantConnected { info: Some(proto::ParticipantInfo::from(&participant)), @@ -190,7 +195,7 @@ async fn room_task( } => { let handle_id = server.next_id() as FfiHandleId; let track_info = proto::TrackInfo::from_remote_track(handle_id, &track); - server.ffi_handles().insert(handle_id, Box::new(Track::from(track))); + server.ffi_handles.insert(handle_id, Box::new(Track::from(track))); Some(proto::room_event::Message::TrackSubscribed( proto::TrackSubscribed { @@ -210,12 +215,11 @@ async fn room_task( }, )), _ => None - }; - - if message.is_some() { + } { + // Send the event to the FfiClient let _ = server.send_event(proto::ffi_event::Message::RoomEvent(proto::RoomEvent{ room_handle: Some(room_handle.into()), - message + message: Some(message) })); } @@ -226,10 +230,3 @@ async fn room_task( }; } } - -async fn participant_task(participant: Participant) { - let mut participant_events = participant.register_observer(); - while let Some(_event) = participant_events.recv().await { - // TODO(theomonnom): convert event to proto - } -} diff --git a/livekit-ffi/src/server/tests.rs b/livekit-ffi/src/server/tests.rs index a5dd9a9..bdc0c34 100644 --- a/livekit-ffi/src/server/tests.rs +++ b/livekit-ffi/src/server/tests.rs @@ -109,7 +109,7 @@ impl TestScope { impl Drop for TestScope { fn drop(&mut self) { // At the end of a test, no more handle should exist - assert!(server::FFI_SERVER.ffi_handles().is_empty()); + assert!(server::FFI_SERVER.ffi_handles.is_empty()); } } diff --git a/livekit-ffi/src/server/video_frame.rs b/livekit-ffi/src/server/video_frame.rs index 2861f97..269416a 100644 --- a/livekit-ffi/src/server/video_frame.rs +++ b/livekit-ffi/src/server/video_frame.rs @@ -39,7 +39,7 @@ impl FfiVideoStream { .id as FfiHandleId; let track = server - .ffi_handles() + .ffi_handles .get(&handle_id) .ok_or(FfiError::InvalidRequest("track not found"))?; @@ -75,7 +75,7 @@ impl FfiVideoStream { // Store the new video stream and return the info let info = proto::VideoStreamInfo::from(&stream); server - .ffi_handles() + .ffi_handles .insert(stream.handle_id, Box::new(stream)); Ok(info) @@ -110,7 +110,7 @@ impl FfiVideoStream { let buffer_info = proto::VideoFrameBufferInfo::from(handle_id, &frame.buffer); server - .ffi_handles() + .ffi_handles .insert(handle_id, Box::new(frame.buffer)); if let Err(err) = server.send_event(proto::ffi_event::Message::VideoStreamEvent( @@ -167,7 +167,7 @@ impl FfiVideoSource { let source_info = proto::VideoSourceInfo::from(&video_source); server - .ffi_handles() + .ffi_handles .insert(video_source.handle_id, Box::new(video_source)); Ok(source_info) @@ -191,7 +191,7 @@ impl FfiVideoSource { .id as FfiHandleId; let buffer = server - .ffi_handles() + .ffi_handles .get(&buffer_handle) .ok_or(FfiError::InvalidRequest("handle not found"))?; diff --git a/livekit/src/lib.rs b/livekit/src/lib.rs index d05e6b9..7d07edd 100644 --- a/livekit/src/lib.rs +++ b/livekit/src/lib.rs @@ -1,5 +1,3 @@ -extern crate core; - pub mod proto; mod room; mod rtc_engine; diff --git a/livekit/src/prelude.rs b/livekit/src/prelude.rs index ff989eb..84a082d 100644 --- a/livekit/src/prelude.rs +++ b/livekit/src/prelude.rs @@ -1,4 +1,4 @@ -pub use crate::participant::{LocalParticipant, Participant, ParticipantEvent, RemoteParticipant}; +pub use crate::participant::{LocalParticipant, Participant, RemoteParticipant}; pub use crate::{ ConnectionState, DataPacketKind, Room, RoomError, RoomEvent, RoomOptions, RoomResult, @@ -8,7 +8,7 @@ pub use crate::publication::{LocalTrackPublication, RemoteTrackPublication, Trac pub use crate::track::{ AudioTrack, LocalAudioTrack, LocalTrack, LocalVideoTrack, RemoteAudioTrack, RemoteTrack, - RemoteVideoTrack, StreamState, Track, TrackEvent, TrackKind, TrackSource, VideoTrack, + RemoteVideoTrack, StreamState, Track, TrackDimension, TrackKind, TrackSource, VideoTrack, }; pub use crate::id::*; diff --git a/livekit/src/proto.rs b/livekit/src/proto.rs index 03544dc..71cc5d2 100644 --- a/livekit/src/proto.rs +++ b/livekit/src/proto.rs @@ -1,7 +1,17 @@ -use crate::{track, DataPacketKind}; +use crate::{participant, track, DataPacketKind}; use livekit_protocol::*; // Conversions +impl From for participant::ConnectionQuality { + fn from(value: ConnectionQuality) -> Self { + match value { + ConnectionQuality::Excellent => Self::Excellent, + ConnectionQuality::Good => Self::Good, + ConnectionQuality::Poor => Self::Poor, + } + } +} + impl TryFrom for track::TrackKind { type Error = &'static str; diff --git a/livekit/src/room/mod.rs b/livekit/src/room/mod.rs index 005d974..56e1f8b 100644 --- a/livekit/src/room/mod.rs +++ b/livekit/src/room/mod.rs @@ -6,11 +6,12 @@ use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RtcEngine}; use livekit_api::signal_client::SignalOptions; use livekit_protocol as proto; use livekit_protocol::observer::Dispatcher; -use parking_lot::{Mutex, RwLock, RwLockReadGuard}; +use parking_lot::RwLock; use std::collections::HashMap; use std::fmt::Debug; use std::sync::Arc; use thiserror::Error; +use tokio::sync::Mutex as AsyncMutex; use tokio::sync::{mpsc, oneshot}; use tokio::task::JoinHandle; @@ -37,14 +38,26 @@ pub enum RoomError { } #[derive(Clone, Debug)] +#[non_exhaustive] pub enum RoomEvent { ParticipantConnected(RemoteParticipant), ParticipantDisconnected(RemoteParticipant), + LocalTrackPublished { + publication: LocalTrackPublication, + }, + LocalTrackUnpublished { + publication: LocalTrackPublication, + }, TrackSubscribed { track: RemoteTrack, publication: RemoteTrackPublication, participant: RemoteParticipant, }, + TrackUnsubscribed { + track: RemoteTrack, + publication: RemoteTrackPublication, + participant: RemoteParticipant, + }, TrackPublished { publication: RemoteTrackPublication, participant: RemoteParticipant, @@ -53,15 +66,10 @@ pub enum RoomEvent { publication: RemoteTrackPublication, participant: RemoteParticipant, }, - TrackUnsubscribed { - track: RemoteTrack, - publication: RemoteTrackPublication, - participant: RemoteParticipant, - }, TrackSubscriptionFailed { + participant: RemoteParticipant, error: track::TrackError, sid: TrackSid, - participant: RemoteParticipant, }, TrackMuted { participant: Participant, @@ -128,7 +136,7 @@ struct RoomHandle { pub struct Room { inner: Arc, - handle: Mutex>, + handle: AsyncMutex>, } impl Debug for Room { @@ -178,7 +186,6 @@ impl Room { metadata: room_info.metadata, }), participants: Default::default(), - participants_tasks: Default::default(), active_speakers: Default::default(), rtc_engine, local_participant, @@ -200,7 +207,7 @@ impl Room { let session = Self { inner, - handle: Mutex::new(Some(RoomHandle { + handle: AsyncMutex::new(Some(RoomHandle { session_task, close_emitter, })), @@ -211,10 +218,10 @@ impl Room { } pub async fn close(&self) -> RoomResult<()> { - if let Some(handle) = self.handle.lock().take() { + if let Some(handle) = self.handle.lock().await.take() { self.inner.close().await; - handle.close_emitter.send(()).ok(); - handle.session_task.await.ok(); + let _ = handle.close_emitter.send(()); + let _ = handle.session_task.await; Ok(()) } else { Err(RoomError::AlreadyClosed) @@ -245,8 +252,8 @@ impl Room { self.inner.info.read().state } - pub fn participants(&self) -> RwLockReadGuard> { - self.inner.participants.read() + pub fn participants(&self) -> HashMap { + self.inner.participants.read().clone() } pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> { @@ -268,7 +275,6 @@ pub(crate) struct RoomSession { active_speakers: RwLock>, local_participant: LocalParticipant, participants: RwLock>, - participants_tasks: RwLock, oneshot::Sender<()>)>>, } impl Debug for RoomSession { @@ -304,70 +310,6 @@ impl RoomSession { } } - /// Forward participant events to the room dispatcher - async fn participant_task( - self: Arc, - participant: Participant, - mut participant_events: mpsc::UnboundedReceiver, - mut close_rx: oneshot::Receiver<()>, - ) { - loop { - tokio::select! { - res = participant_events.recv() => { - if let Some(event) = res { - if let Err(err) = self.on_participant_event(&participant, event).await { - log::error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err); - } - } - }, - _ = &mut close_rx => { - log::trace!("closing participant_task for {:?}", participant.sid()); - break; - }, - } - } - } - - async fn on_participant_event( - self: &Arc, - participant: &Participant, - event: ParticipantEvent, - ) -> RoomResult<()> { - if let Participant::Remote(remote_participant) = participant { - match event { - ParticipantEvent::TrackPublished { publication } => { - self.dispatcher.dispatch(&RoomEvent::TrackPublished { - participant: remote_participant.clone(), - publication, - }); - } - ParticipantEvent::TrackUnpublished { publication } => { - self.dispatcher.dispatch(&RoomEvent::TrackUnpublished { - participant: remote_participant.clone(), - publication, - }); - } - ParticipantEvent::TrackSubscribed { track, publication } => { - self.dispatcher.dispatch(&RoomEvent::TrackSubscribed { - participant: remote_participant.clone(), - track, - publication, - }); - } - ParticipantEvent::TrackUnsubscribed { track, publication } => { - self.dispatcher.dispatch(&RoomEvent::TrackUnsubscribed { - participant: remote_participant.clone(), - track, - publication, - }); - } - _ => {} - }; - } - - Ok(()) - } - async fn on_engine_event(self: &Arc, event: EngineEvent) -> RoomResult<()> { match event { EngineEvent::ParticipantUpdate { updates } => self.handle_participant_update(updates), @@ -432,7 +374,7 @@ impl RoomSession { participant: participant.clone(), }); - participant.on_data_received(payload, kind); + //participant.on_data_received(payload, kind); } } EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers), @@ -570,7 +512,8 @@ impl RoomSession { fn handle_restarting(self: &Arc) { // Remove existing participants/subscriptions on full reconnect - for (_, participant) in self.participants.read().iter() { + let participants = self.participants.read().clone(); + for (_, participant) in participants.iter() { self.clone() .handle_participant_disconnect(participant.clone()); } @@ -621,44 +564,64 @@ impl RoomSession { metadata, ); - // Create the participant task - let (close_tx, close_rx) = oneshot::channel(); - let participant_task = tokio::spawn(self.clone().participant_task( - Participant::Remote(participant.clone()), - participant.register_observer(), - close_rx, - )); - self.participants_tasks - .write() - .insert(sid.clone(), (participant_task, close_tx)); + let dispatcher = self.dispatcher.clone(); + participant.on_track_published(move |participant, publication| { + dispatcher.dispatch(&RoomEvent::TrackPublished { + participant, + publication, + }); + }); + + let dispatcher = self.dispatcher.clone(); + participant.on_track_unpublished(move |participant, publication| { + dispatcher.dispatch(&RoomEvent::TrackUnpublished { + participant, + publication, + }); + }); + + let dispatcher = self.dispatcher.clone(); + participant.on_track_subscribed(move |participant, publication, track| { + dispatcher.dispatch(&RoomEvent::TrackSubscribed { + participant, + track, + publication, + }); + }); + + let dispatcher = self.dispatcher.clone(); + participant.on_track_unsubscribed(move |participant, publication, track| { + dispatcher.dispatch(&RoomEvent::TrackUnsubscribed { + participant, + track, + publication, + }); + }); + + let dispatcher = self.dispatcher.clone(); + participant.on_track_subscription_failed(move |participant, sid, error| { + dispatcher.dispatch(&RoomEvent::TrackSubscriptionFailed { + participant, + sid, + error, + }); + }); self.participants.write().insert(sid, participant.clone()); + participant } /// A participant has disconnected /// Cleanup the participant and emit an event fn handle_participant_disconnect(self: Arc, remote_participant: RemoteParticipant) { - tokio::spawn(async move { - for (sid, _) in &*remote_participant.tracks() { - remote_participant.unpublish_track(&sid); - } + for (sid, _) in remote_participant.tracks() { + remote_participant.unpublish_track(&sid); + } - // Close the participant task - let ptask = self - .participants_tasks - .write() - .remove(&remote_participant.sid()); - - if let Some((task, close_tx)) = ptask { - let _ = close_tx.send(()); - let _ = task.await; - } - - self.participants.write().remove(&remote_participant.sid()); - self.dispatcher - .dispatch(&RoomEvent::ParticipantDisconnected(remote_participant)); - }); + self.participants.write().remove(&remote_participant.sid()); + self.dispatcher + .dispatch(&RoomEvent::ParticipantDisconnected(remote_participant)); } fn get_participant(&self, sid: &ParticipantSid) -> Option { diff --git a/livekit/src/room/participant/local_participant.rs b/livekit/src/room/participant/local_participant.rs index 1c3963a..3e4307a 100644 --- a/livekit/src/room/participant/local_participant.rs +++ b/livekit/src/room/participant/local_participant.rs @@ -1,5 +1,5 @@ use super::ConnectionQuality; -use super::ParticipantInternal; +use super::ParticipantInner; use crate::options; use crate::options::compute_video_encodings; use crate::options::video_layers_from_encodings; @@ -9,15 +9,27 @@ use crate::rtc_engine::RtcEngine; use crate::DataPacketKind; use livekit_protocol as proto; use livekit_webrtc::rtp_parameters::RtpEncodingParameters; -use parking_lot::RwLockReadGuard; +use parking_lot::Mutex; use std::collections::HashMap; use std::fmt::Debug; use std::sync::Arc; -use tokio::sync::mpsc; + +#[derive(Default)] +struct LocalEvents { + local_track_published: + Mutex>>, + local_track_unpublished: + Mutex>>, +} + +struct LocalInfo { + events: LocalEvents, +} #[derive(Clone)] pub struct LocalParticipant { - inner: Arc, + inner: Arc, + local: Arc, } impl Debug for LocalParticipant { @@ -39,12 +51,54 @@ impl LocalParticipant { metadata: String, ) -> Self { Self { - inner: Arc::new(ParticipantInternal::new( - rtc_engine, sid, identity, name, metadata, - )), + inner: super::new_inner(rtc_engine, sid, identity, name, metadata), + local: Arc::new(LocalInfo { + events: LocalEvents::default(), + }), } } + pub(crate) fn update_info(self: &Self, info: proto::ParticipantInfo) { + super::update_info(&self.inner, &Participant::Local(self.clone()), info); + } + + pub(crate) fn set_speaking(&self, speaking: bool) { + super::set_speaking(&self.inner, &Participant::Local(self.clone()), speaking); + } + + pub(crate) fn set_audio_level(&self, level: f32) { + super::set_audio_level(&self.inner, &Participant::Local(self.clone()), level); + } + + pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) { + super::set_connection_quality(&self.inner, &Participant::Local(self.clone()), quality); + } + + #[allow(dead_code)] + pub(crate) fn on_local_track_published( + &self, + handler: impl Fn(LocalParticipant, LocalTrackPublication) + Send + 'static, + ) { + *self.local.events.local_track_published.lock() = Some(Box::new(handler)); + } + + #[allow(dead_code)] + pub(crate) fn on_local_track_unpublished( + &self, + handler: impl Fn(LocalParticipant, LocalTrackPublication) + Send + 'static, + ) { + *self.local.events.local_track_unpublished.lock() = Some(Box::new(handler)); + } + + pub(crate) fn add_publication(&self, publication: TrackPublication) { + super::add_publication(&self.inner, &Participant::Local(self.clone()), publication); + } + + #[allow(dead_code)] + pub(crate) fn remove_publication(&self, sid: &TrackSid) { + super::remove_publication(&self.inner, &Participant::Local(self.clone()), sid); + } + pub async fn publish_track( &self, track: LocalTrack, @@ -87,11 +141,7 @@ impl LocalParticipant { } } let track_info = self.inner.rtc_engine.add_track(req).await?; - let publication = LocalTrackPublication::new( - track_info.clone(), - Arc::downgrade(&self.inner), - track.clone(), - ); + let publication = LocalTrackPublication::new(track_info.clone(), track.clone()); track.update_info(track_info); // Update sid + source log::debug!("publishing track with cid {:?}", track.rtc_track().id()); @@ -101,8 +151,7 @@ impl LocalParticipant { .create_sender(track.clone(), options, encodings) .await?; - track.update_transceiver(Some(transceiver)); - //track.start(); + track.set_transceiver(Some(transceiver)); track.enable(); tokio::spawn({ @@ -112,14 +161,12 @@ impl LocalParticipant { } }); - self.inner - .add_publication(TrackPublication::Local(publication.clone())); + self.add_publication(TrackPublication::Local(publication.clone())); - self.inner - .dispatcher - .dispatch(&ParticipantEvent::LocalTrackPublished { - publication: publication.clone(), - }); + if let Some(local_track_published) = self.local.events.local_track_published.lock().as_ref() + { + local_track_published(self.clone(), publication.clone()); + } Ok(publication) } @@ -129,20 +176,21 @@ impl LocalParticipant { track: TrackSid, _stop_on_unpublish: bool, ) -> RoomResult { - let mut tracks = self.inner.tracks.write(); - if let Some(TrackPublication::Local(publication)) = tracks.remove(&track) { + let publication = self.inner.tracks.write().remove(&track); + if let Some(TrackPublication::Local(publication)) = publication { let track = publication.track(); let sender = track.transceiver().unwrap().sender(); self.inner.rtc_engine.remove_track(sender).await?; - track.update_transceiver(None); + track.set_transceiver(None); - self.inner - .dispatcher - .dispatch(&ParticipantEvent::LocalTrackUnpublished { - publication: publication.clone(), - }); - // publication.update_track(None); + if let Some(local_track_unpublished) = + self.local.events.local_track_unpublished.lock().as_ref() + { + local_track_unpublished(self.clone(), publication.clone()); + } + + publication.set_track(None); tokio::spawn({ let rtc_engine = self.inner.rtc_engine.clone(); @@ -179,7 +227,6 @@ impl LocalParticipant { .map_err(Into::into) } - #[inline] pub fn get_track_publication(&self, sid: &TrackSid) -> Option { self.inner.tracks.read().get(sid).map(|track| { if let TrackPublication::Local(local) = track { @@ -190,68 +237,35 @@ impl LocalParticipant { }) } - #[inline] pub fn sid(&self) -> ParticipantSid { - self.inner.sid() + self.inner.info.read().sid.clone() } - #[inline] pub fn identity(&self) -> ParticipantIdentity { - self.inner.identity() + self.inner.info.read().identity.clone() } - #[inline] pub fn name(&self) -> String { - self.inner.name() + self.inner.info.read().name.clone() } - #[inline] pub fn metadata(&self) -> String { - self.inner.metadata() + self.inner.info.read().metadata.clone() } - #[inline] pub fn is_speaking(&self) -> bool { - self.inner.is_speaking() + self.inner.info.read().speaking } - #[inline] - pub fn tracks(&self) -> RwLockReadGuard> { - self.inner.tracks() + pub fn tracks(&self) -> HashMap { + self.inner.tracks.read().clone() } - #[inline] pub fn audio_level(&self) -> f32 { - self.inner.audio_level() + self.inner.info.read().audio_level } - #[inline] pub fn connection_quality(&self) -> ConnectionQuality { - self.inner.connection_quality() - } - - #[inline] - pub fn register_observer(&self) -> mpsc::UnboundedReceiver { - self.inner.register_observer() - } - - #[inline] - pub(crate) fn update_info(self: &Self, info: proto::ParticipantInfo) { - self.inner.update_info(info); - } - - #[inline] - pub(crate) fn set_speaking(&self, speaking: bool) { - self.inner.set_speaking(speaking); - } - - #[inline] - pub(crate) fn set_audio_level(&self, level: f32) { - self.inner.set_audio_level(level); - } - - #[inline] - pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) { - self.inner.set_connection_quality(quality); + self.inner.info.read().connection_quality } } diff --git a/livekit/src/room/participant/mod.rs b/livekit/src/room/participant/mod.rs index 4f72636..352ec45 100644 --- a/livekit/src/room/participant/mod.rs +++ b/livekit/src/room/participant/mod.rs @@ -1,15 +1,11 @@ use crate::prelude::*; use crate::rtc_engine::RtcEngine; -use crate::track::TrackError; use livekit_protocol as proto; use livekit_protocol::enum_dispatch; -use livekit_protocol::observer::Dispatcher; -use parking_lot::{RwLock, RwLockReadGuard}; +use parking_lot::{Mutex, RwLock}; use std::collections::HashMap; use std::fmt::Debug; use std::sync::Arc; -use std::thread::JoinHandle; -use tokio::sync::{mpsc, oneshot}; mod local_participant; mod remote_participant; @@ -17,52 +13,7 @@ mod remote_participant; pub use local_participant::*; pub use remote_participant::*; -#[derive(Debug, Clone)] -pub enum ParticipantEvent { - TrackPublished { - publication: RemoteTrackPublication, - }, - TrackUnpublished { - publication: RemoteTrackPublication, - }, - TrackSubscribed { - track: RemoteTrack, - publication: RemoteTrackPublication, - }, - TrackUnsubscribed { - track: RemoteTrack, - publication: RemoteTrackPublication, - }, - TrackSubscriptionFailed { - error: TrackError, - sid: TrackSid, - }, - DataReceived { - payload: Arc>, - kind: DataPacketKind, - }, - SpeakingChanged { - speaking: bool, - }, - TrackMuted { - publication: TrackPublication, - }, - TrackUnmuted { - publication: TrackPublication, - }, - ConnectionQualityChanged { - quality: ConnectionQuality, - }, - LocalTrackPublished { - publication: LocalTrackPublication, - }, - LocalTrackUnpublished { - publication: LocalTrackPublication, - }, -} - #[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[repr(u8)] pub enum ConnectionQuality { Unknown, Excellent, @@ -70,27 +21,6 @@ pub enum ConnectionQuality { Poor, } -impl From for ConnectionQuality { - fn from(value: u8) -> Self { - match value { - 1 => Self::Excellent, - 2 => Self::Good, - 3 => Self::Poor, - _ => Self::Unknown, - } - } -} - -impl From for ConnectionQuality { - fn from(value: proto::ConnectionQuality) -> Self { - match value { - proto::ConnectionQuality::Excellent => Self::Excellent, - proto::ConnectionQuality::Good => Self::Good, - proto::ConnectionQuality::Poor => Self::Poor, - } - } -} - #[derive(Debug, Clone)] pub enum Participant { Local(LocalParticipant), @@ -107,18 +37,20 @@ impl Participant { pub fn is_speaking(self: &Self) -> bool; pub fn audio_level(self: &Self) -> f32; pub fn connection_quality(self: &Self) -> ConnectionQuality; - pub fn tracks(self: &Self) -> RwLockReadGuard>; - pub fn register_observer(self: &Self) -> mpsc::UnboundedReceiver; + pub fn tracks(self: &Self) -> HashMap; + pub(crate) fn update_info(self: &Self, info: proto::ParticipantInfo) -> (); + + // Internal functions called by the Room when receiving the associated signal messages pub(crate) fn set_speaking(self: &Self, speaking: bool) -> (); pub(crate) fn set_audio_level(self: &Self, level: f32) -> (); pub(crate) fn set_connection_quality(self: &Self, quality: ConnectionQuality) -> (); - pub(crate) fn update_info(self: &Self, info: proto::ParticipantInfo) -> (); + pub(crate) fn add_publication(self: &Self, publication: TrackPublication) -> (); + pub(crate) fn remove_publication(self: &Self, sid: &TrackSid) -> (); ); } -#[derive(Debug)] -pub(crate) struct ParticipantInfo { +struct ParticipantInfo { pub sid: ParticipantSid, pub identity: ParticipantIdentity, pub name: String, @@ -128,101 +60,118 @@ pub(crate) struct ParticipantInfo { pub connection_quality: ConnectionQuality, } -#[derive(Debug)] -pub(crate) struct ParticipantInternal { - pub(super) rtc_engine: Arc, - pub(super) dispatcher: Dispatcher, +#[derive(Default)] +struct ParticipantEvents { + track_muted: Mutex>>, + track_unmuted: Mutex>>, +} + +pub(super) struct ParticipantInner { + rtc_engine: Arc, info: RwLock, tracks: RwLock>, - tracks_tasks: RwLock, oneshot::Sender<()>)>>, + events: Arc, } -impl ParticipantInternal { - pub fn new( - rtc_engine: Arc, - sid: ParticipantSid, - identity: ParticipantIdentity, - name: String, - metadata: String, - ) -> Self { - Self { - rtc_engine, - info: RwLock::new(ParticipantInfo { - sid, - identity, - name, - metadata, - speaking: false, - audio_level: 0.0, - connection_quality: ConnectionQuality::Unknown, - }), - dispatcher: Default::default(), - tracks: Default::default(), - tracks_tasks: Default::default(), +pub(super) fn new_inner( + rtc_engine: Arc, + sid: ParticipantSid, + identity: ParticipantIdentity, + name: String, + metadata: String, +) -> Arc { + Arc::new(ParticipantInner { + rtc_engine, + info: RwLock::new(ParticipantInfo { + sid, + identity, + name, + metadata, + speaking: false, + audio_level: 0.0, + connection_quality: ConnectionQuality::Unknown, + }), + tracks: Default::default(), + events: Default::default(), + }) +} + +pub(super) fn update_info( + inner: &Arc, + _participant: &Participant, + new_info: proto::ParticipantInfo, +) { + let mut info = inner.info.write(); + info.sid = new_info.sid.into(); + info.name = new_info.name; + info.identity = new_info.identity.into(); + info.metadata = new_info.metadata; // TODO(theomonnom): callback MetadataChanged +} + +pub(super) fn set_speaking( + inner: &Arc, + _participant: &Participant, + speaking: bool, +) { + inner.info.write().speaking = speaking; +} + +pub(super) fn set_audio_level( + inner: &Arc, + _participant: &Participant, + audio_level: f32, +) { + inner.info.write().audio_level = audio_level; +} + +pub(super) fn set_connection_quality( + inner: &Arc, + _participant: &Participant, + quality: ConnectionQuality, +) { + inner.info.write().connection_quality = quality; +} + +pub(super) fn remove_publication( + inner: &Arc, + _participant: &Participant, + sid: &TrackSid, +) -> Option { + let mut tracks = inner.tracks.write(); + let publication = tracks.remove(sid); + if let Some(publication) = publication.clone() { + // remove events + publication.on_muted(|_, _| {}); + publication.on_unmuted(|_, _| {}); + } else { + // shouldn't happen (internal) + log::warn!("could not find publication to remove: {}", sid); + } + + publication +} + +pub(super) fn add_publication( + inner: &Arc, + participant: &Participant, + publication: TrackPublication, +) { + let mut tracks = inner.tracks.write(); + tracks.insert(publication.sid(), publication.clone()); + + let events = inner.events.clone(); + let particiant = participant.clone(); + publication.on_muted(move |publication, track| { + if let Some(cb) = events.track_muted.lock().as_ref() { + cb(particiant.clone(), publication, track); } - } + }); - pub fn update_info(&self, new_info: proto::ParticipantInfo) { - let mut info = self.info.write(); - info.sid = new_info.sid.into(); - info.name = new_info.name; - info.identity = new_info.identity.into(); - info.metadata = new_info.metadata; // TODO(theomonnom): callback MetadataChanged - } - - pub fn sid(&self) -> ParticipantSid { - self.info.read().sid.clone() - } - - pub fn identity(&self) -> ParticipantIdentity { - self.info.read().identity.clone() - } - - pub fn name(&self) -> String { - self.info.read().name.clone() - } - - pub fn metadata(&self) -> String { - self.info.read().metadata.clone() - } - - pub fn is_speaking(&self) -> bool { - self.info.read().speaking - } - - pub fn tracks(&self) -> RwLockReadGuard> { - self.tracks.read() - } - - pub fn audio_level(&self) -> f32 { - self.info.read().audio_level - } - - pub fn connection_quality(&self) -> ConnectionQuality { - self.info.read().connection_quality - } - - pub fn register_observer(&self) -> mpsc::UnboundedReceiver { - self.dispatcher.register() - } - - pub fn set_speaking(&self, speaking: bool) { - self.info.write().speaking = speaking; - } - - pub fn set_audio_level(&self, audio_level: f32) { - self.info.write().audio_level = audio_level; - } - - pub fn set_connection_quality(&self, quality: ConnectionQuality) { - self.info.write().connection_quality = quality; - } - - pub fn remove_publication(&self, sid: &TrackSid) { - self.tracks.write().remove(sid); - } - - pub fn add_publication(&self, publication: TrackPublication) { - self.tracks.write().insert(publication.sid(), publication); - } + let events = inner.events.clone(); + let participant = participant.clone(); + publication.on_unmuted(move |publication, track| { + if let Some(cb) = events.track_unmuted.lock().as_ref() { + cb(participant.clone(), publication, track); + } + }); } diff --git a/livekit/src/room/participant/remote_participant.rs b/livekit/src/room/participant/remote_participant.rs index 6acc0f3..2d99e9a 100644 --- a/livekit/src/room/participant/remote_participant.rs +++ b/livekit/src/room/participant/remote_participant.rs @@ -1,23 +1,39 @@ use super::TrackKind; -use super::{ConnectionQuality, ParticipantInternal}; +use super::{ConnectionQuality, ParticipantInner}; +use crate::prelude::*; use crate::rtc_engine::RtcEngine; use crate::track::TrackError; -use crate::{prelude::*, DataPacketKind}; use livekit_protocol as proto; use livekit_webrtc::prelude::*; -use parking_lot::RwLockReadGuard; +use parking_lot::Mutex; use std::collections::{HashMap, HashSet}; use std::fmt::Debug; use std::sync::Arc; use std::time::Duration; -use tokio::sync::mpsc; use tokio::time::timeout; const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5); +#[derive(Default)] +struct RemoteEvents { + track_published: Mutex>>, + track_unpublished: Mutex>>, + track_subscribed: + Mutex>>, + track_unsubscribed: + Mutex>>, + track_subscription_failed: + Mutex>>, +} + +struct RemoteInfo { + events: Arc, +} + #[derive(Clone)] pub struct RemoteParticipant { - inner: Arc, + inner: Arc, + remote: Arc, } impl Debug for RemoteParticipant { @@ -39,23 +55,13 @@ impl RemoteParticipant { metadata: String, ) -> Self { Self { - inner: Arc::new(ParticipantInternal::new( - rtc_engine, sid, identity, name, metadata, - )), + inner: super::new_inner(rtc_engine, sid, identity, name, metadata), + remote: Arc::new(RemoteInfo { + events: Default::default(), + }), } } - /// Called by the RoomSession when receiving data from the RtcSession - /// It is just used to emit the Data event on the participant dispatcher. - pub(crate) fn on_data_received(&self, data: Arc>, kind: DataPacketKind) { - self.inner - .dispatcher - .dispatch(&ParticipantEvent::DataReceived { - payload: data, - kind, - }); - } - pub(crate) async fn add_subscribed_media_track( &self, sid: TrackSid, @@ -106,7 +112,6 @@ impl RemoteParticipant { log::debug!("starting track: {:?}", sid); - remote_publication.update_track(Some(track.clone().into())); //track.set_muted(remote_publication.is_muted()); track.update_info(proto::TrackInfo { sid: remote_publication.sid().to_string(), @@ -116,26 +121,22 @@ impl RemoteParticipant { ..Default::default() }); - self.inner - .add_publication(TrackPublication::Remote(remote_publication.clone())); - // track.start(); + self.add_publication(TrackPublication::Remote(remote_publication.clone())); track.enable(); - self.inner - .dispatcher - .dispatch(&ParticipantEvent::TrackSubscribed { - track, - publication: remote_publication, - }); + remote_publication.set_track(Some(track.into())); // This will fire TrackSubscribed on the publication } else { log::error!("could not find published track with sid: {:?}", sid); - self.inner - .dispatcher - .dispatch(&ParticipantEvent::TrackSubscriptionFailed { - sid: sid.clone(), - error: TrackError::TrackNotFound(sid.clone().to_string()), - }); + if let Some(track_subscription_failed) = + self.remote.events.track_subscription_failed.lock().as_ref() + { + track_subscription_failed( + self.clone(), + sid.clone(), + TrackError::TrackNotFound(sid.0), + ); + } } } @@ -144,44 +145,37 @@ impl RemoteParticipant { // Unsubscribe to the track if needed if let Some(track) = publication.track() { track.disable(); - - self.inner - .dispatcher - .dispatch(&ParticipantEvent::TrackUnsubscribed { - track: track.clone(), - publication: publication.clone(), - }); + publication.set_track(None); // This will fire TrackUnsubscribed on the publication } - self.inner.remove_publication(sid); + self.remove_publication(sid); - self.inner - .dispatcher - .dispatch(&ParticipantEvent::TrackUnpublished { - publication: publication.clone(), - }); - - publication.update_track(None); + if let Some(track_unpublished) = self.remote.events.track_unpublished.lock().as_ref() { + track_unpublished(self.clone(), publication.clone()); + } } } pub(crate) fn update_info(&self, info: proto::ParticipantInfo) { - self.inner.update_info(info.clone()); + super::update_info( + &self.inner, + &Participant::Remote(self.clone()), + info.clone(), + ); let mut valid_tracks = HashSet::::new(); for track in info.tracks { if let Some(publication) = self.get_track_publication(&track.sid.clone().into()) { publication.update_info(track.clone()); } else { - let publication = - RemoteTrackPublication::new(track.clone(), Arc::downgrade(&self.inner), None); - self.inner - .add_publication(TrackPublication::Remote(publication.clone())); + let publication = RemoteTrackPublication::new(track.clone(), None); + + self.add_publication(TrackPublication::Remote(publication.clone())); // This is a new track, dispatch publish event - self.inner - .dispatcher - .dispatch(&ParticipantEvent::TrackPublished { publication }); + if let Some(track_published) = self.remote.events.track_published.lock().as_ref() { + track_published(self.clone(), publication); + } } valid_tracks.insert(track.sid.into()); @@ -197,7 +191,131 @@ impl RemoteParticipant { } } - #[inline] + pub(crate) fn on_track_published( + &self, + track_published: impl Fn(RemoteParticipant, RemoteTrackPublication) + Send + 'static, + ) { + *self.remote.events.track_published.lock() = Some(Box::new(track_published)); + } + + pub(crate) fn on_track_unpublished( + &self, + track_unpublished: impl Fn(RemoteParticipant, RemoteTrackPublication) + Send + 'static, + ) { + *self.remote.events.track_unpublished.lock() = Some(Box::new(track_unpublished)); + } + + pub(crate) fn on_track_subscribed( + &self, + track_subscribed: impl Fn(RemoteParticipant, RemoteTrackPublication, RemoteTrack) + + Send + + 'static, + ) { + *self.remote.events.track_subscribed.lock() = Some(Box::new(track_subscribed)); + } + + pub(crate) fn on_track_unsubscribed( + &self, + track_unsubscribed: impl Fn(RemoteParticipant, RemoteTrackPublication, RemoteTrack) + + Send + + 'static, + ) { + *self.remote.events.track_unsubscribed.lock() = Some(Box::new(track_unsubscribed)); + } + + pub(crate) fn on_track_subscription_failed( + &self, + track_subscription_failed: impl Fn(RemoteParticipant, TrackSid, TrackError) + Send + 'static, + ) { + *self.remote.events.track_subscription_failed.lock() = + Some(Box::new(track_subscription_failed)); + } + + pub(crate) fn set_speaking(&self, speaking: bool) { + super::set_speaking(&self.inner, &Participant::Remote(self.clone()), speaking); + } + + pub(crate) fn set_audio_level(&self, level: f32) { + super::set_audio_level(&self.inner, &Participant::Remote(self.clone()), level); + } + + pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) { + super::set_connection_quality(&self.inner, &Participant::Remote(self.clone()), quality); + } + + pub(crate) fn add_publication(&self, publication: TrackPublication) { + super::add_publication( + &self.inner, + &Participant::Remote(self.clone()), + publication.clone(), + ); + + let TrackPublication::Remote(publication) = publication else { + panic!("expected remote publication"); + }; + + publication.on_subscription_update_needed({ + let rtc_engine = self.inner.rtc_engine.clone(); + let psid = self.sid().0.clone(); + move |publication| { + let rtc_engine = rtc_engine.clone(); + let psid = psid.clone(); + tokio::spawn(async move { + let tsid = publication.sid().0.clone(); + let update_subscription = proto::UpdateSubscription { + track_sids: vec![tsid.clone()], + subscribe: publication.is_subscribed(), + participant_tracks: vec![proto::ParticipantTracks { + participant_sid: psid, + track_sids: vec![tsid.clone()], + }], + }; + + let _ = rtc_engine + .send_request(proto::signal_request::Message::Subscription( + update_subscription, + )) + .await; + }); + } + }); + + publication.on_subscribed({ + let events = self.remote.events.clone(); + let participant = self.clone(); + move |publication, track| { + if let Some(track_subscribed) = events.track_subscribed.lock().as_ref() { + track_subscribed(participant.clone(), publication, track); + } + } + }); + + publication.on_unsubscribed({ + let events = self.remote.events.clone(); + let participant = self.clone(); + move |publication, track| { + if let Some(track_unsubscribed) = events.track_unsubscribed.lock().as_ref() { + track_unsubscribed(participant.clone(), publication, track); + } + } + }); + } + + pub(crate) fn remove_publication(&self, sid: &TrackSid) { + let publication = + super::remove_publication(&self.inner, &Participant::Remote(self.clone()), sid); + + if let Some(publication) = publication { + let TrackPublication::Remote(publication) = publication else { + panic!("expected remote publication"); + }; + + publication.on_subscription_update_needed(|_| {}); + publication.on_subscribed(|_, _| {}); + publication.on_unsubscribed(|_, _| {}); + } + } + pub fn get_track_publication(&self, sid: &TrackSid) -> Option { self.inner.tracks.read().get(sid).map(|track| { if let TrackPublication::Remote(remote) = track { @@ -207,63 +325,35 @@ impl RemoteParticipant { }) } - #[inline] pub fn sid(&self) -> ParticipantSid { - self.inner.sid() + self.inner.info.read().sid.clone() } - #[inline] pub fn identity(&self) -> ParticipantIdentity { - self.inner.identity() + self.inner.info.read().identity.clone() } - #[inline] pub fn name(&self) -> String { - self.inner.name() + self.inner.info.read().name.clone() } - #[inline] pub fn metadata(&self) -> String { - self.inner.metadata() + self.inner.info.read().metadata.clone() } - #[inline] pub fn is_speaking(&self) -> bool { - self.inner.is_speaking() + self.inner.info.read().speaking } - #[inline] - pub fn tracks(&self) -> RwLockReadGuard> { - self.inner.tracks() + pub fn tracks(&self) -> HashMap { + self.inner.tracks.read().clone() } - #[inline] pub fn audio_level(&self) -> f32 { - self.inner.audio_level() + self.inner.info.read().audio_level } - #[inline] pub fn connection_quality(&self) -> ConnectionQuality { - self.inner.connection_quality() - } - - #[inline] - pub fn register_observer(&self) -> mpsc::UnboundedReceiver { - self.inner.register_observer() - } - - #[inline] - pub(crate) fn set_speaking(&self, speaking: bool) { - self.inner.set_speaking(speaking); - } - - #[inline] - pub(crate) fn set_audio_level(&self, level: f32) { - self.inner.set_audio_level(level); - } - - #[inline] - pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) { - self.inner.set_connection_quality(quality); + self.inner.info.read().connection_quality } } diff --git a/livekit/src/room/publication/local.rs b/livekit/src/room/publication/local.rs index 2e3002a..1c2a71f 100644 --- a/livekit/src/room/publication/local.rs +++ b/livekit/src/room/publication/local.rs @@ -1,127 +1,100 @@ use super::TrackPublicationInner; -use crate::id::TrackSid; -use crate::participant::ParticipantInternal; -use crate::track::{LocalTrack, TrackDimension, TrackKind, TrackSource}; +use crate::prelude::*; use livekit_protocol as proto; -use std::sync::{Arc, Weak}; +use std::fmt::Debug; +use std::sync::Arc; -#[derive(Debug)] -struct LocalTrackPublicationInner { - publication_inner: TrackPublicationInner, +#[derive(Clone)] +pub struct LocalTrackPublication { + inner: Arc, } -#[derive(Clone, Debug)] -pub struct LocalTrackPublication { - inner: Arc, +impl Debug for LocalTrackPublication { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("LocalTrackPublication") + .field("sid", &self.sid()) + .field("name", &self.name()) + .field("kind", &self.kind()) + .finish() + } } impl LocalTrackPublication { - pub(crate) fn new( - info: proto::TrackInfo, - participant: Weak, - track: LocalTrack, - ) -> Self { + pub(crate) fn new(info: proto::TrackInfo, track: LocalTrack) -> Self { Self { - inner: Arc::new(LocalTrackPublicationInner { - publication_inner: TrackPublicationInner::new( - info, - participant, - Some(track.into()), - ), - }), + inner: super::new_inner(info, Some(track.into())), } } - pub async fn mute(&self) {} + pub(crate) fn on_muted(&self, f: impl Fn(TrackPublication, Track) + Send + 'static) { + *self.inner.events.muted.lock() = Some(Box::new(f)); + } - pub async fn unmute(&self) {} + pub(crate) fn on_unmuted(&self, f: impl Fn(TrackPublication, Track) + Send + 'static) { + *self.inner.events.unmuted.lock() = Some(Box::new(f)); + } - pub async fn pause_upstream(&self) {} + pub(crate) fn set_track(&self, track: Option) { + super::set_track(&self.inner, &TrackPublication::Local(self.clone()), track); + } - pub async fn resume_upstream(&self) {} + #[allow(dead_code)] + pub(crate) fn update_info(&self, info: proto::TrackInfo) { + super::update_info(&self.inner, &TrackPublication::Local(self.clone()), info); + } - /*pub fn set_muted(&self, muted: bool) { - if self.is_muted() == muted { - return; - } + pub fn mute(&self) { + self.track().mute(); + } - self.track().rtc_track().set_enabled(!muted); + pub fn unmute(&self) { + self.track().unmute(); + } - let participant = self.inner.publication_inner.participant().upgrade(); - if participant.is_none() { - log::warn!("publication's participant is invalid, set_muted failed"); - return; - } - let participant = participant.unwrap(); - - // Engine update muted - - // Participant MUTED/UNMUTED event - }*/ - - #[inline] pub fn sid(&self) -> TrackSid { - self.inner.publication_inner.sid() + self.inner.info.read().sid.clone() } - #[inline] pub fn name(&self) -> String { - self.inner.publication_inner.name() + self.inner.info.read().name.clone() } - #[inline] pub fn kind(&self) -> TrackKind { - self.inner.publication_inner.kind() + self.inner.info.read().kind } - #[inline] pub fn source(&self) -> TrackSource { - self.inner.publication_inner.source() + self.inner.info.read().source } - #[inline] pub fn simulcasted(&self) -> bool { - self.inner.publication_inner.simulcasted() + self.inner.info.read().simulcasted } - #[inline] pub fn dimension(&self) -> TrackDimension { - self.inner.publication_inner.dimension() + self.inner.info.read().dimension } - #[inline] pub fn track(&self) -> LocalTrack { self.inner - .publication_inner - .track() + .info + .read() + .track + .clone() .unwrap() .try_into() .unwrap() } - #[inline] pub fn mime_type(&self) -> String { - self.inner.publication_inner.mime_type() + self.inner.info.read().mime_type.clone() } - #[inline] pub fn is_muted(&self) -> bool { - self.inner.publication_inner.is_muted() + self.inner.info.read().muted } - #[inline] pub fn is_remote(&self) -> bool { false } - - /*#[inline] - pub(crate) fn update_track(&self, track: Option) { - self.inner.publication_inner.update_track(track); - }*/ - - #[allow(dead_code)] - #[inline] - pub(crate) fn update_info(&self, info: proto::TrackInfo) { - self.inner.publication_inner.update_info(info); - } } diff --git a/livekit/src/room/publication/mod.rs b/livekit/src/room/publication/mod.rs index 61aa16b..9d9ea3b 100644 --- a/livekit/src/room/publication/mod.rs +++ b/livekit/src/room/publication/mod.rs @@ -1,14 +1,10 @@ use super::track::TrackDimension; -use crate::participant::ParticipantInternal; use crate::prelude::*; use crate::track::Track; use livekit_protocol as proto; use livekit_protocol::enum_dispatch; use parking_lot::{Mutex, RwLock}; -use proto::observer::Dispatcher; use std::sync::Arc; -use std::sync::Weak; -use tokio::sync::Notify; mod local; mod remote; @@ -16,23 +12,6 @@ mod remote; pub use local::*; pub use remote::*; -#[derive(Debug, Clone)] -pub enum PublicationEvent { - Muted, - Unmuted, - Subscribed, - Unsubscribed, - SubscriptionStatusChanged { - old_state: SubscriptionStatus, - new_state: SubscriptionStatus, - }, - SubscriptionPermissionChanged { - old_state: PermissionStatus, - new_state: PermissionStatus, - }, - SubscriptionFailed, -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SubscriptionStatus { Desired, @@ -64,8 +43,20 @@ impl TrackPublication { pub fn mime_type(self: &Self) -> String; pub fn is_muted(self: &Self) -> bool; pub fn is_remote(self: &Self) -> bool; + + pub(crate) fn on_muted(self: &Self, on_mute: impl Fn(TrackPublication, Track) + Send + 'static) -> (); + pub(crate) fn on_unmuted(self: &Self, on_unmute: impl Fn(TrackPublication, Track) + Send + 'static) -> (); + pub(crate) fn update_info(self: &Self, info: proto::TrackInfo) -> (); ); + #[allow(dead_code)] + pub(crate) fn set_track(&self, track: Option) { + match self { + TrackPublication::Local(p) => p.set_track(track), + TrackPublication::Remote(p) => p.set_track(track.map(|t| t.try_into().unwrap())), + } + } + pub fn track(&self) -> Option { match self { TrackPublication::Local(p) => Some(p.track().into()), @@ -74,173 +65,106 @@ impl TrackPublication { } } -#[derive(Debug)] -pub(crate) struct PublicationInfo { - track: Option, - name: String, - sid: TrackSid, - kind: TrackKind, - source: TrackSource, - simulcasted: bool, - dimension: TrackDimension, - mime_type: String, - muted: bool, +struct PublicationInfo { + pub track: Option, + pub name: String, + pub sid: TrackSid, + pub kind: TrackKind, + pub source: TrackSource, + pub simulcasted: bool, + pub dimension: TrackDimension, + pub mime_type: String, + pub muted: bool, } -#[derive(Debug)] -pub(crate) struct TrackPublicationInner { +#[derive(Default)] +struct PublicationEvents { + muted: Mutex>>, + unmuted: Mutex>>, +} + +pub(super) struct TrackPublicationInner { info: RwLock, - dispatcher: Dispatcher, - participant: Weak, - //forward_handle: Mutex>>, - forward_close: Arc, + events: Arc, } -impl TrackPublicationInner { - pub fn new( - info: proto::TrackInfo, - participant: Weak, - track: Option, - ) -> Self { - let info = PublicationInfo { - track, - name: info.name, - sid: info.sid.into(), - kind: proto::TrackType::from_i32(info.r#type) - .unwrap() - .try_into() - .unwrap(), - source: proto::TrackSource::from_i32(info.source) - .unwrap() - .try_into() - .unwrap(), - simulcasted: info.simulcast, - dimension: TrackDimension(info.width, info.height), - mime_type: info.mime_type, - muted: info.muted, - }; +pub(super) fn new_inner( + info: proto::TrackInfo, + track: Option, +) -> Arc { + let info = PublicationInfo { + track, + name: info.name, + sid: info.sid.into(), + kind: proto::TrackType::from_i32(info.r#type) + .unwrap() + .try_into() + .unwrap(), + source: proto::TrackSource::from_i32(info.source) + .unwrap() + .try_into() + .unwrap(), + simulcasted: info.simulcast, + dimension: TrackDimension(info.width, info.height), + mime_type: info.mime_type, + muted: info.muted, + }; - Self { - info: RwLock::new(info), - dispatcher: Default::default(), - participant, - //forward_handle: Default::default(), - forward_close: Default::default(), - } + Arc::new(TrackPublicationInner { + info: RwLock::new(info), + events: Default::default(), + }) +} + +pub(super) fn update_info( + inner: &TrackPublicationInner, + _publication: &TrackPublication, + new_info: proto::TrackInfo, +) { + let mut info = inner.info.write(); + info.name = new_info.name; + info.sid = new_info.sid.into(); + info.dimension = TrackDimension(new_info.width, new_info.height); + info.mime_type = new_info.mime_type; + info.kind = TrackKind::try_from(proto::TrackType::from_i32(new_info.r#type).unwrap()).unwrap(); + info.source = TrackSource::from(proto::TrackSource::from_i32(new_info.source).unwrap()); + info.simulcasted = new_info.simulcast; +} + +pub(super) fn set_track( + inner: &TrackPublicationInner, + publication: &TrackPublication, + track: Option, +) { + let mut info = inner.info.write(); + if let Some(prev_track) = info.track.as_ref() { + prev_track.on_muted(|_| {}); + prev_track.on_unmuted(|_| {}); } - // Forward track events to the publication events - // e.g: this also allow us to access the signal_client and notify the server if - // a local track changed mute state - async fn track_forward_task( - close_notifier: Weak, - track: Track, - dispatcher: Dispatcher, - ) { - let mut track_events = track.register_observer(); - loop { - let notifier = close_notifier.upgrade(); - if notifier.is_none() { - break; - } - let notified = notifier.as_ref().unwrap().notified(); + info.track = track.clone(); - tokio::select! { - _ = notified => { - break; - } - Some(event) = track_events.recv() => { - match event { - TrackEvent::Muted => { - dispatcher.dispatch(&PublicationEvent::Muted); - } - TrackEvent::Unmuted => { - dispatcher.dispatch(&PublicationEvent::Unmuted); - } - } + if let Some(track) = track.as_ref() { + info.sid = track.sid(); + + track.on_muted({ + let events = inner.events.clone(); + let publication = publication.clone(); + move |track| { + if let Some(on_muted) = events.muted.lock().as_ref() { + on_muted(publication.clone(), track); } } - } - } + }); - pub fn update_track(&self, track: Option) { - //let forward_task = self.forward_handle.lock().take(); - //if let Some(task) = forward_task { - // Make sure to close the old forwarder before changing the track - self.forward_close.notify_waiters(); - //let _ = task.await; - // } - - let mut info = self.info.write(); - info.track = track.clone(); - - if let Some(track) = track { - let _handle = tokio::spawn(Self::track_forward_task( - Arc::downgrade(&self.forward_close), - track, - self.dispatcher.clone(), - )); - //let mut forward_handle = self.forward_handle.lock(); - //*forward_handle = Some(handle); - } - } - - // Called when updating a participant info - pub fn update_info(&self, new_info: proto::TrackInfo) { - let mut info = self.info.write(); - info.name = new_info.name; - info.sid = new_info.sid.into(); - info.dimension = TrackDimension(new_info.width, new_info.height); - info.mime_type = new_info.mime_type; - info.kind = - TrackKind::try_from(proto::TrackType::from_i32(new_info.r#type).unwrap()).unwrap(); - info.source = TrackSource::from(proto::TrackSource::from_i32(new_info.source).unwrap()); - info.simulcasted = new_info.simulcast; - - // TODO MUTE ????????????????? - // info.muted = new_info.muted; - // if let Some(track) = info.track.as_ref() { - // track.set_muted(info.muted); - // } - } - - pub fn participant(&self) -> Weak { - self.participant.clone() - } - - pub fn sid(&self) -> TrackSid { - self.info.read().sid.clone() - } - - pub fn name(&self) -> String { - self.info.read().name.clone() - } - - pub fn kind(&self) -> TrackKind { - self.info.read().kind - } - - pub fn source(&self) -> TrackSource { - self.info.read().source - } - - pub fn simulcasted(&self) -> bool { - self.info.read().simulcasted - } - - pub fn dimension(&self) -> TrackDimension { - self.info.read().dimension.clone() - } - - pub fn mime_type(&self) -> String { - self.info.read().mime_type.clone() - } - - pub fn track(&self) -> Option { - self.info.read().track.clone() - } - - pub fn is_muted(&self) -> bool { - self.info.read().muted + track.on_unmuted({ + let events = inner.events.clone(); + let publication = publication.clone(); + move |track| { + if let Some(on_unmuted) = events.unmuted.lock().as_ref() { + on_unmuted(publication.clone(), track); + } + } + }); } } diff --git a/livekit/src/room/publication/remote.rs b/livekit/src/room/publication/remote.rs index 7ca4149..5cbd746 100644 --- a/livekit/src/room/publication/remote.rs +++ b/livekit/src/room/publication/remote.rs @@ -1,101 +1,203 @@ -use super::{PermissionStatus, SubscriptionStatus, TrackPublicationInner}; -use crate::id::TrackSid; -use crate::participant::ParticipantInternal; -use crate::publication::PublicationEvent; -use crate::track::{RemoteTrack, Track, TrackDimension, TrackKind, TrackSource}; +use super::{PermissionStatus, SubscriptionStatus, TrackPublication, TrackPublicationInner}; +use crate::prelude::*; use livekit_protocol as proto; -use parking_lot::RwLock; -use std::sync::{Arc, Weak}; +use parking_lot::{Mutex, RwLock}; +use std::fmt::Debug; +use std::sync::Arc; + +#[derive(Default)] +struct RemoteEvents { + subscribed: Mutex>>, + unsubscribed: Mutex>>, + subscription_status_changed: Mutex< + Option>, + >, // Old status, new status + permission_status_changed: Mutex< + Option>, + >, // Old status, new status + subscription_update_needed: Mutex>>, +} #[derive(Debug)] struct RemoteInfo { subscribed: bool, allowed: bool, - // TODO(theomonnom): other remote info } -#[derive(Debug)] struct RemoteInner { - publication_inner: TrackPublicationInner, info: RwLock, + events: RemoteEvents, } -#[derive(Clone, Debug)] +#[derive(Clone)] pub struct RemoteTrackPublication { - inner: Arc, + inner: Arc, + remote: Arc, +} + +impl Debug for RemoteTrackPublication { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RemoteTrackPublication") + .field("is_subscribed", &self.is_subscribed()) + .field("is_allowed", &self.is_allowed()) + .finish() + } } impl RemoteTrackPublication { - pub(crate) fn new( - info: proto::TrackInfo, - participant: Weak, - track: Option, - ) -> Self { + pub(crate) fn new(info: proto::TrackInfo, track: Option) -> Self { Self { - inner: Arc::new(RemoteInner { - publication_inner: TrackPublicationInner::new( - info, - participant, - track.map(Into::into), - ), + inner: super::new_inner(info, track.map(Into::into)), + remote: Arc::new(RemoteInner { info: RwLock::new(RemoteInfo { subscribed: false, allowed: false, }), + events: Default::default(), }), } } - pub fn set_subscribed(&self, subscribed: bool) { + /// This is called by the RemoteParticipant when it successfully subscribe to the track or when + /// the track is being unsubscribed. + /// We register the mute events from the track here so we can forward them. + pub(crate) fn set_track(&self, track: Option) { let old_subscription_state = self.subscription_status(); let old_permission_state = self.permission_status(); - let mut info = self.inner.info.write(); + + let prev_track = self.track(); + + if let Some(prev_track) = prev_track { + if let Some(unsubscribed) = self.remote.events.unsubscribed.lock().as_ref() { + unsubscribed(self.clone(), prev_track); + } + } + + super::set_track( + &self.inner, + &TrackPublication::Remote(self.clone()), + track.clone().map(Into::into), + ); + + if let Some(track) = track { + if let Some(subscribed) = self.remote.events.subscribed.lock().as_ref() { + subscribed(self.clone(), track); + } + } + + self.emit_subscription_update(old_subscription_state); + self.emit_permission_update(old_permission_state); + } + + pub(crate) fn emit_subscription_update(&self, old_subscription_state: SubscriptionStatus) { + if old_subscription_state != self.subscription_status() { + if let Some(subscription_status_changed) = self + .remote + .events + .subscription_status_changed + .lock() + .as_ref() + { + subscription_status_changed( + self.clone(), + old_subscription_state, + self.subscription_status(), + ); + } + } + } + + pub(crate) fn emit_permission_update(&self, old_permission_state: PermissionStatus) { + if old_permission_state != self.permission_status() { + if let Some(subscription_permission_changed) = + self.remote.events.permission_status_changed.lock().as_ref() + { + subscription_permission_changed( + self.clone(), + old_permission_state, + self.permission_status(), + ); + } + } + } + + pub(crate) fn update_info(&self, info: proto::TrackInfo) { + super::update_info(&self.inner, &TrackPublication::Remote(self.clone()), info); + } + + pub(crate) fn on_muted(&self, f: impl Fn(TrackPublication, Track) + Send + 'static) { + *self.inner.events.muted.lock() = Some(Box::new(f)); + } + + pub(crate) fn on_unmuted(&self, f: impl Fn(TrackPublication, Track) + Send + 'static) { + *self.inner.events.unmuted.lock() = Some(Box::new(f)); + } + + pub(crate) fn on_subscribed( + &self, + f: impl Fn(RemoteTrackPublication, RemoteTrack) + Send + 'static, + ) { + *self.remote.events.subscribed.lock() = Some(Box::new(f)); + } + + pub(crate) fn on_unsubscribed( + &self, + f: impl Fn(RemoteTrackPublication, RemoteTrack) + Send + 'static, + ) { + *self.remote.events.unsubscribed.lock() = Some(Box::new(f)); + } + + #[allow(dead_code)] + pub(crate) fn on_subscription_status_changed( + &self, + f: impl Fn(RemoteTrackPublication, SubscriptionStatus, SubscriptionStatus) + Send + 'static, + ) { + *self.remote.events.subscription_status_changed.lock() = Some(Box::new(f)); + } + + #[allow(dead_code)] + pub(crate) fn on_permission_status_changed( + &self, + f: impl Fn(RemoteTrackPublication, PermissionStatus, PermissionStatus) + Send + 'static, + ) { + *self.remote.events.permission_status_changed.lock() = Some(Box::new(f)); + } + + pub(crate) fn on_subscription_update_needed( + &self, + f: impl Fn(RemoteTrackPublication) + Send + 'static, + ) { + *self.remote.events.subscription_update_needed.lock() = Some(Box::new(f)); + } + + pub async fn set_subscribed(&self, subscribed: bool) { + let old_subscription_state = self.subscription_status(); + let old_permission_state = self.permission_status(); + + let mut info = self.remote.info.write(); info.subscribed = subscribed; if subscribed { info.allowed = true; } - let participant = self.inner.publication_inner.participant.upgrade(); - if participant.is_none() { - log::warn!("publication's participant is invalid, set_subscribed failed"); - return; - } - let participant = participant.unwrap(); - - let update_subscription = proto::UpdateSubscription { - track_sids: vec![self.sid().0], - subscribe: subscribed, - participant_tracks: vec![proto::ParticipantTracks { - participant_sid: participant.sid().0, - track_sids: vec![self.sid().0], - }], - }; - - // Engine update subscription - - if old_subscription_state != self.subscription_status() { - self.inner.publication_inner.dispatcher.dispatch( - &PublicationEvent::SubscriptionStatusChanged { - old_state: old_subscription_state, - new_state: self.subscription_status(), - }, - ) + // Request to send an update to the SFU + if let Some(subscription_update_needed) = self + .remote + .events + .subscription_update_needed + .lock() + .as_ref() + { + subscription_update_needed(self.clone()); } - if old_permission_state != self.permission_status() { - self.inner.publication_inner.dispatcher.dispatch( - &PublicationEvent::SubscriptionPermissionChanged { - old_state: old_permission_state, - new_state: self.permission_status(), - }, - ) - } + self.emit_subscription_update(old_subscription_state); + self.emit_permission_update(old_permission_state); } - #[inline] pub fn subscription_status(&self) -> SubscriptionStatus { - if !self.inner.info.read().subscribed { + if !self.is_subscribed() { return SubscriptionStatus::Unsubscribed; } @@ -106,9 +208,8 @@ impl RemoteTrackPublication { SubscriptionStatus::Subscribed } - #[inline] pub fn permission_status(&self) -> PermissionStatus { - if self.inner.info.read().allowed { + if self.is_allowed() { PermissionStatus::Allowed } else { PermissionStatus::NotAllowed @@ -116,69 +217,55 @@ impl RemoteTrackPublication { } pub fn is_subscribed(&self) -> bool { - self.inner.info.read().allowed && self.track().is_some() + self.is_allowed() && self.track().is_some() + } + + pub fn is_allowed(&self) -> bool { + self.remote.info.read().allowed } - #[inline] pub fn sid(&self) -> TrackSid { - self.inner.publication_inner.sid() + self.inner.info.read().sid.clone() } - #[inline] pub fn name(&self) -> String { - self.inner.publication_inner.name() + self.inner.info.read().name.clone() } - #[inline] pub fn kind(&self) -> TrackKind { - self.inner.publication_inner.kind() + self.inner.info.read().kind } - #[inline] pub fn source(&self) -> TrackSource { - self.inner.publication_inner.source() + self.inner.info.read().source } - #[inline] pub fn simulcasted(&self) -> bool { - self.inner.publication_inner.simulcasted() + self.inner.info.read().simulcasted } - #[inline] pub fn dimension(&self) -> TrackDimension { - self.inner.publication_inner.dimension() + self.inner.info.read().dimension.clone() } - #[inline] pub fn track(&self) -> Option { self.inner - .publication_inner - .track() + .info + .read() + .track + .clone() .map(|track| track.try_into().unwrap()) } - #[inline] pub fn mime_type(&self) -> String { - self.inner.publication_inner.mime_type() + self.inner.info.read().mime_type.clone() } - #[inline] pub fn is_muted(&self) -> bool { - self.inner.publication_inner.is_muted() + self.inner.info.read().muted } - #[inline] pub fn is_remote(&self) -> bool { true } - - #[inline] - pub(crate) fn update_track(&self, track: Option) { - self.inner.publication_inner.update_track(track); - } - - #[inline] - pub(crate) fn update_info(&self, info: proto::TrackInfo) { - self.inner.publication_inner.update_info(info); - } } diff --git a/livekit/src/room/track/audio_track.rs b/livekit/src/room/track/audio_track.rs new file mode 100644 index 0000000..70e73a6 --- /dev/null +++ b/livekit/src/room/track/audio_track.rs @@ -0,0 +1,43 @@ +use super::track_dispatch; +use crate::prelude::*; +use livekit_protocol as proto; +use livekit_protocol::enum_dispatch; +use livekit_webrtc::prelude::*; + +#[derive(Clone, Debug)] +pub enum AudioTrack { + Local(LocalAudioTrack), + Remote(RemoteAudioTrack), +} + +impl AudioTrack { + track_dispatch!([Local, Remote]); + + pub fn rtc_track(&self) -> RtcAudioTrack { + match self { + Self::Local(track) => track.rtc_track().into(), + Self::Remote(track) => track.rtc_track().into(), + } + } +} + +impl From for Track { + fn from(track: AudioTrack) -> Self { + match track { + AudioTrack::Local(track) => Self::LocalAudio(track), + AudioTrack::Remote(track) => Self::RemoteAudio(track), + } + } +} + +impl TryFrom for AudioTrack { + type Error = &'static str; + + fn try_from(track: Track) -> Result { + match track { + Track::LocalAudio(track) => Ok(Self::Local(track)), + Track::RemoteAudio(track) => Ok(Self::Remote(track)), + _ => Err("not an audio track"), + } + } +} diff --git a/livekit/src/room/track/local_audio_track.rs b/livekit/src/room/track/local_audio_track.rs index 309a46a..5beacca 100644 --- a/livekit/src/room/track/local_audio_track.rs +++ b/livekit/src/room/track/local_audio_track.rs @@ -6,7 +6,6 @@ use livekit_protocol as proto; use livekit_webrtc::prelude::*; use std::fmt::Debug; use std::sync::Arc; -use tokio::sync::mpsc; #[derive(Clone)] pub struct LocalAudioTrack { @@ -27,7 +26,7 @@ impl Debug for LocalAudioTrack { impl LocalAudioTrack { pub(crate) fn new(name: String, rtc_track: RtcAudioTrack, source: RtcAudioSource) -> Self { Self { - inner: Arc::new(TrackInner::new( + inner: Arc::new(super::new_inner( "unknown".to_string().into(), // sid name, TrackKind::Audio, @@ -37,96 +36,6 @@ impl LocalAudioTrack { } } - #[inline] - pub fn sid(&self) -> TrackSid { - self.inner.sid() - } - - #[inline] - pub fn name(&self) -> String { - self.inner.name() - } - - #[inline] - pub fn kind(&self) -> TrackKind { - self.inner.kind() - } - - #[inline] - pub fn source(&self) -> TrackSource { - self.inner.source() - } - - #[inline] - pub fn stream_state(&self) -> StreamState { - self.inner.stream_state() - } - - #[inline] - pub fn enable(&self) { - self.inner.enable() - } - - #[inline] - pub fn disable(&self) { - self.inner.disable() - } - - #[inline] - pub fn is_muted(&self) -> bool { - self.inner.is_muted() - } - - #[inline] - pub fn mute(&self) { - self.inner.set_muted(true); - } - - #[inline] - pub fn unmute(&self) { - self.inner.set_muted(false); - } - - #[inline] - pub fn rtc_track(&self) -> RtcAudioTrack { - if let MediaStreamTrack::Audio(audio) = self.inner.rtc_track() { - return audio; - } - unreachable!() - } - - #[inline] - pub fn rtc_source(&self) -> RtcAudioSource { - self.source.clone() - } - - #[inline] - pub fn register_observer(&self) -> mpsc::UnboundedReceiver { - self.inner.register_observer() - } - - #[inline] - pub fn is_remote(&self) -> bool { - false - } - - #[inline] - pub(crate) fn transceiver(&self) -> Option { - self.inner.transceiver() - } - - #[inline] - pub(crate) fn update_transceiver(&self, transceiver: Option) { - self.inner.update_transceiver(transceiver) - } - - #[inline] - pub(crate) fn update_info(&self, info: proto::TrackInfo) { - self.inner.update_info(info) - } -} - -impl LocalAudioTrack { pub fn create_audio_track(name: &str, source: RtcAudioSource) -> LocalAudioTrack { let rtc_track = match source.clone() { #[cfg(not(target_arch = "wasm32"))] @@ -141,4 +50,79 @@ impl LocalAudioTrack { }; Self::new(name.to_string(), rtc_track, source) } + + pub fn sid(&self) -> TrackSid { + self.inner.info.read().sid.clone() + } + + pub fn name(&self) -> String { + self.inner.info.read().name.clone() + } + + pub fn kind(&self) -> TrackKind { + self.inner.info.read().kind + } + + pub fn source(&self) -> TrackSource { + self.inner.info.read().source + } + + pub fn stream_state(&self) -> StreamState { + self.inner.info.read().stream_state + } + + pub fn enable(&self) { + self.inner.rtc_track.set_enabled(true); + } + + pub fn disable(&self) { + self.inner.rtc_track.set_enabled(false); + } + + pub fn is_muted(&self) -> bool { + self.inner.info.read().muted + } + + pub fn mute(&self) { + super::set_muted(&self.inner, &Track::LocalAudio(self.clone()), true); + } + + pub fn unmute(&self) { + super::set_muted(&self.inner, &Track::LocalAudio(self.clone()), false); + } + + pub fn rtc_track(&self) -> RtcAudioTrack { + if let MediaStreamTrack::Audio(audio) = self.inner.rtc_track.clone() { + return audio; + } + unreachable!(); + } + + pub fn rtc_source(&self) -> RtcAudioSource { + self.source.clone() + } + + pub fn is_remote(&self) -> bool { + false + } + + pub fn on_muted(&self, f: impl Fn(Track) + Send + 'static) { + *self.inner.events.muted.lock() = Some(Box::new(f)); + } + + pub fn on_unmuted(&self, f: impl Fn(Track) + Send + 'static) { + *self.inner.events.unmuted.lock() = Some(Box::new(f)); + } + + pub(crate) fn transceiver(&self) -> Option { + self.inner.info.read().transceiver.clone() + } + + pub(crate) fn set_transceiver(&self, transceiver: Option) { + self.inner.info.write().transceiver = transceiver; + } + + pub(crate) fn update_info(&self, info: proto::TrackInfo) { + super::update_info(&self.inner, &Track::LocalAudio(self.clone()), info); + } } diff --git a/livekit/src/room/track/local_track.rs b/livekit/src/room/track/local_track.rs index 9aef391..6e5bb43 100644 --- a/livekit/src/room/track/local_track.rs +++ b/livekit/src/room/track/local_track.rs @@ -1,12 +1,8 @@ -use super::TrackInner; -use super::{track_dispatch, LocalAudioTrack, LocalVideoTrack}; +use super::track_dispatch; use crate::prelude::*; -use crate::track::TrackEvent; use livekit_protocol as proto; use livekit_protocol::enum_dispatch; use livekit_webrtc::prelude::*; -use std::sync::Arc; -use tokio::sync::mpsc; #[derive(Clone, Debug)] pub enum LocalTrack { @@ -23,7 +19,6 @@ impl LocalTrack { pub fn unmute(self: &Self) -> (); ); - #[inline] pub fn rtc_track(&self) -> MediaStreamTrack { match self { Self::Audio(track) => track.rtc_track().into(), @@ -31,3 +26,24 @@ impl LocalTrack { } } } + +impl From for Track { + fn from(track: LocalTrack) -> Self { + match track { + LocalTrack::Audio(track) => Self::LocalAudio(track), + LocalTrack::Video(track) => Self::LocalVideo(track), + } + } +} + +impl TryFrom for LocalTrack { + type Error = &'static str; + + fn try_from(track: Track) -> Result { + match track { + Track::LocalAudio(track) => Ok(Self::Audio(track)), + Track::LocalVideo(track) => Ok(Self::Video(track)), + _ => Err("not a local track"), + } + } +} diff --git a/livekit/src/room/track/local_video_track.rs b/livekit/src/room/track/local_video_track.rs index a2fded7..b556c8c 100644 --- a/livekit/src/room/track/local_video_track.rs +++ b/livekit/src/room/track/local_video_track.rs @@ -5,7 +5,6 @@ use livekit_protocol as proto; use livekit_webrtc::prelude::*; use std::fmt::Debug; use std::sync::Arc; -use tokio::sync::mpsc; #[derive(Clone)] pub struct LocalVideoTrack { @@ -26,7 +25,7 @@ impl Debug for LocalVideoTrack { impl LocalVideoTrack { pub fn new(name: String, rtc_track: RtcVideoTrack, source: RtcVideoSource) -> Self { Self { - inner: Arc::new(TrackInner::new( + inner: Arc::new(super::new_inner( "unknown".to_string().into(), // sid name, TrackKind::Video, @@ -36,96 +35,6 @@ impl LocalVideoTrack { } } - #[inline] - pub fn sid(&self) -> TrackSid { - self.inner.sid() - } - - #[inline] - pub fn name(&self) -> String { - self.inner.name() - } - - #[inline] - pub fn kind(&self) -> TrackKind { - self.inner.kind() - } - - #[inline] - pub fn source(&self) -> TrackSource { - self.inner.source() - } - - #[inline] - pub fn stream_state(&self) -> StreamState { - self.inner.stream_state() - } - - #[inline] - pub fn enable(&self) { - self.inner.enable() - } - - #[inline] - pub fn disable(&self) { - self.inner.disable() - } - - #[inline] - pub fn is_muted(&self) -> bool { - self.inner.is_muted() - } - - #[inline] - pub fn mute(&self) { - self.inner.set_muted(true); - } - - #[inline] - pub fn unmute(&self) { - self.inner.set_muted(false); - } - - #[inline] - pub fn rtc_track(&self) -> RtcVideoTrack { - if let MediaStreamTrack::Video(video) = self.inner.rtc_track() { - return video; - } - unreachable!() - } - - #[inline] - pub fn rtc_source(&self) -> RtcVideoSource { - self.source.clone() - } - - #[inline] - pub fn register_observer(&self) -> mpsc::UnboundedReceiver { - self.inner.register_observer() - } - - #[inline] - pub fn is_remote(&self) -> bool { - false - } - - #[inline] - pub(crate) fn transceiver(&self) -> Option { - self.inner.transceiver() - } - - #[inline] - pub(crate) fn update_transceiver(&self, transceiver: Option) { - self.inner.update_transceiver(transceiver) - } - - #[inline] - pub(crate) fn update_info(&self, info: proto::TrackInfo) { - self.inner.update_info(info) - } -} - -impl LocalVideoTrack { pub fn create_video_track(name: &str, source: RtcVideoSource) -> LocalVideoTrack { let rtc_track = match source.clone() { #[cfg(not(target_arch = "wasm32"))] @@ -141,4 +50,79 @@ impl LocalVideoTrack { Self::new(name.to_string(), rtc_track, source) } + + pub fn sid(&self) -> TrackSid { + self.inner.info.read().sid.clone() + } + + pub fn name(&self) -> String { + self.inner.info.read().name.clone() + } + + pub fn kind(&self) -> TrackKind { + self.inner.info.read().kind + } + + pub fn source(&self) -> TrackSource { + self.inner.info.read().source + } + + pub fn stream_state(&self) -> StreamState { + self.inner.info.read().stream_state + } + + pub fn enable(&self) { + self.inner.rtc_track.set_enabled(true); + } + + pub fn disable(&self) { + self.inner.rtc_track.set_enabled(false); + } + + pub fn is_muted(&self) -> bool { + self.inner.info.read().muted + } + + pub fn mute(&self) { + super::set_muted(&self.inner, &Track::LocalVideo(self.clone()), true); + } + + pub fn unmute(&self) { + super::set_muted(&self.inner, &Track::LocalVideo(self.clone()), false); + } + + pub fn rtc_track(&self) -> RtcVideoTrack { + if let MediaStreamTrack::Video(video) = self.inner.rtc_track.clone() { + return video; + } + unreachable!(); + } + + pub fn is_remote(&self) -> bool { + false + } + + pub fn rtc_source(&self) -> RtcVideoSource { + self.source.clone() + } + + pub fn on_muted(&self, f: impl Fn(Track) + Send + 'static) { + *self.inner.events.muted.lock() = Some(Box::new(f)); + } + + pub fn on_unmuted(&self, f: impl Fn(Track) + Send + 'static) { + *self.inner.events.unmuted.lock() = Some(Box::new(f)); + } + + pub(crate) fn transceiver(&self) -> Option { + self.inner.info.read().transceiver.clone() + } + + pub(crate) fn set_transceiver(&self, transceiver: Option) { + self.inner.info.write().transceiver = transceiver; + } + + pub(crate) fn update_info(&self, info: proto::TrackInfo) { + super::update_info(&self.inner, &&Track::LocalVideo(self.clone()), info); + } } diff --git a/livekit/src/room/track/mod.rs b/livekit/src/room/track/mod.rs index adb0529..56373e2 100644 --- a/livekit/src/room/track/mod.rs +++ b/livekit/src/room/track/mod.rs @@ -1,25 +1,29 @@ use crate::prelude::*; use livekit_protocol as proto; use livekit_protocol::enum_dispatch; -use livekit_protocol::observer::Dispatcher; use livekit_webrtc::prelude::*; -use parking_lot::RwLock; +use parking_lot::{Mutex, RwLock}; +use std::fmt::Debug; +use std::sync::Arc; use thiserror::Error; -use tokio::sync::mpsc; +mod audio_track; mod local_audio_track; mod local_track; mod local_video_track; mod remote_audio_track; mod remote_track; mod remote_video_track; +mod video_track; +pub use audio_track::*; pub use local_audio_track::*; pub use local_track::*; pub use local_video_track::*; pub use remote_audio_track::*; pub use remote_track::*; pub use remote_video_track::*; +pub use video_track::*; #[derive(Error, Debug, Clone)] pub enum TrackError { @@ -48,35 +52,9 @@ pub enum TrackSource { ScreenshareAudio, } -#[derive(Debug, Clone)] -pub enum TrackEvent { - Muted, - Unmuted, -} - -#[derive(Clone, Copy, Debug)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct TrackDimension(pub u32, pub u32); -#[derive(Clone, Debug)] -pub enum Track { - LocalAudio(LocalAudioTrack), - LocalVideo(LocalVideoTrack), - RemoteAudio(RemoteAudioTrack), - RemoteVideo(RemoteVideoTrack), -} - -#[derive(Clone, Debug)] -pub enum VideoTrack { - Local(LocalVideoTrack), - Remote(RemoteVideoTrack), -} - -#[derive(Clone, Debug)] -pub enum AudioTrack { - Local(LocalAudioTrack), - Remote(RemoteAudioTrack), -} - macro_rules! track_dispatch { ([$($variant:ident),+]) => { enum_dispatch!( @@ -90,21 +68,27 @@ macro_rules! track_dispatch { pub fn disable(self: &Self) -> (); pub fn is_muted(self: &Self) -> bool; pub fn is_remote(self: &Self) -> bool; - pub fn register_observer(self: &Self) -> mpsc::UnboundedReceiver; + pub fn on_muted(self: &Self, on_mute: impl Fn(Track) + Send + 'static) -> (); + pub fn on_unmuted(self: &Self, on_unmute: impl Fn(Track) + Send + 'static) -> (); pub(crate) fn transceiver(self: &Self) -> Option; - pub(crate) fn update_transceiver(self: &Self, transceiver: Option) -> (); + pub(crate) fn set_transceiver(self: &Self, transceiver: Option) -> (); pub(crate) fn update_info(self: &Self, info: proto::TrackInfo) -> (); ); }; } -pub(crate) use track_dispatch; +#[derive(Clone, Debug)] +pub enum Track { + LocalAudio(LocalAudioTrack), + LocalVideo(LocalVideoTrack), + RemoteAudio(RemoteAudioTrack), + RemoteVideo(RemoteVideoTrack), +} impl Track { track_dispatch!([LocalAudio, LocalVideo, RemoteAudio, RemoteVideo]); - #[inline] pub fn rtc_track(&self) -> MediaStreamTrack { match self { Self::LocalAudio(track) => track.rtc_track().into(), @@ -115,232 +99,83 @@ impl Track { } } -impl VideoTrack { - track_dispatch!([Local, Remote]); +pub(super) use track_dispatch; - #[inline] - pub fn rtc_track(&self) -> RtcVideoTrack { - match self { - Self::Local(track) => track.rtc_track(), - Self::Remote(track) => track.rtc_track(), - } - } -} - -impl AudioTrack { - track_dispatch!([Local, Remote]); - - #[inline] - pub fn rtc_track(&self) -> RtcAudioTrack { - match self { - Self::Local(track) => track.rtc_track().into(), - Self::Remote(track) => track.rtc_track().into(), - } - } +#[derive(Default)] +struct TrackEvents { + pub muted: Mutex>>, + pub unmuted: Mutex>>, } #[derive(Debug)] struct TrackInfo { + pub sid: TrackSid, + pub name: String, + pub kind: TrackKind, + pub source: TrackSource, + pub stream_state: StreamState, + pub muted: bool, + pub transceiver: Option, +} + +pub(super) struct TrackInner { + info: RwLock, + rtc_track: MediaStreamTrack, + events: TrackEvents, +} + +pub(super) fn new_inner( sid: TrackSid, name: String, kind: TrackKind, - source: TrackSource, - stream_state: StreamState, - muted: bool, - transceiver: Option, -} - -#[derive(Debug)] -pub(crate) struct TrackInner { - info: RwLock, rtc_track: MediaStreamTrack, - dispatcher: Dispatcher, -} - -impl TrackInner { - pub fn new(sid: TrackSid, name: String, kind: TrackKind, rtc_track: MediaStreamTrack) -> Self { - Self { - info: RwLock::new(TrackInfo { - sid, - name, - kind, - source: TrackSource::Unknown, - stream_state: StreamState::Active, - muted: false, - transceiver: None, - }), - rtc_track, - dispatcher: Default::default(), - } - } - - pub fn sid(&self) -> TrackSid { - self.info.read().sid.clone() - } - - pub fn name(&self) -> String { - self.info.read().name.clone() - } - - pub fn kind(&self) -> TrackKind { - self.info.read().kind - } - - pub fn source(&self) -> TrackSource { - self.info.read().source - } - - pub fn stream_state(&self) -> StreamState { - self.info.read().stream_state - } - - pub fn is_muted(&self) -> bool { - self.info.read().muted - } - - pub fn enable(&self) { - self.rtc_track.set_enabled(true); - } - - pub fn disable(&self) { - self.rtc_track.set_enabled(false); - } - - pub fn rtc_track(&self) -> MediaStreamTrack { - self.rtc_track.clone() - } - - pub fn register_observer(&self) -> mpsc::UnboundedReceiver { - self.dispatcher.register() - } - - pub fn transceiver(&self) -> Option { - self.info.read().transceiver.clone() - } - - pub fn update_transceiver(&self, transceiver: Option) { - self.info.write().transceiver = transceiver; - } - - pub fn set_muted(&self, muted: bool) { - log::debug!("set_muted: {} {}", self.sid(), muted); - if self.is_muted() == muted { - return; - } - - if muted { - self.disable(); - } else { - self.enable(); - } - - self.dispatcher.dispatch(if muted { - &TrackEvent::Muted - } else { - &TrackEvent::Unmuted - }); - } - - pub fn update_info(&self, new_info: proto::TrackInfo) { - let mut info = self.info.write(); - info.name = new_info.name; - info.sid = new_info.sid.into(); - info.kind = - TrackKind::try_from(proto::TrackType::from_i32(new_info.r#type).unwrap()).unwrap(); - info.source = TrackSource::from(proto::TrackSource::from_i32(new_info.source).unwrap()); - // Muted and StreamState are not handled separately (events) +) -> TrackInner { + TrackInner { + info: RwLock::new(TrackInfo { + sid, + name, + kind, + source: TrackSource::Unknown, + stream_state: StreamState::Active, + muted: false, + transceiver: None, + }), + rtc_track, + events: Default::default(), } } -impl From for Track { - fn from(track: RemoteTrack) -> Self { - match track { - RemoteTrack::Audio(track) => Self::RemoteAudio(track), - RemoteTrack::Video(track) => Self::RemoteVideo(track), +pub(super) fn set_muted(inner: &Arc, track: &Track, muted: bool) { + let info = inner.info.read(); + log::debug!("set_muted: {} {}", info.sid, muted); + if info.muted == muted { + return; + } + drop(info); + + if muted { + inner.rtc_track.set_enabled(false); + } else { + inner.rtc_track.set_enabled(true); + } + + inner.info.write().muted = muted; + + if muted { + if let Some(on_mute) = inner.events.muted.lock().as_ref() { + on_mute(track.clone()); + } + } else { + if let Some(on_unmute) = inner.events.unmuted.lock().as_ref() { + on_unmute(track.clone()); } } } -impl From for Track { - fn from(track: LocalTrack) -> Self { - match track { - LocalTrack::Audio(track) => Self::LocalAudio(track), - LocalTrack::Video(track) => Self::LocalVideo(track), - } - } -} - -impl From for Track { - fn from(track: VideoTrack) -> Self { - match track { - VideoTrack::Local(track) => Self::LocalVideo(track), - VideoTrack::Remote(track) => Self::RemoteVideo(track), - } - } -} - -impl From for Track { - fn from(track: AudioTrack) -> Self { - match track { - AudioTrack::Local(track) => Self::LocalAudio(track), - AudioTrack::Remote(track) => Self::RemoteAudio(track), - } - } -} - -impl TryFrom for RemoteTrack { - type Error = &'static str; - - fn try_from(track: Track) -> Result { - match track { - Track::RemoteAudio(track) => Ok(Self::Audio(track)), - Track::RemoteVideo(track) => Ok(Self::Video(track)), - _ => Err("not a remote track"), - } - } -} - -impl TryFrom for LocalTrack { - type Error = &'static str; - - fn try_from(track: Track) -> Result { - match track { - Track::LocalAudio(track) => Ok(Self::Audio(track)), - Track::LocalVideo(track) => Ok(Self::Video(track)), - _ => Err("not a local track"), - } - } -} - -impl TryFrom for VideoTrack { - type Error = &'static str; - - fn try_from(track: Track) -> Result { - match track { - Track::LocalVideo(track) => Ok(Self::Local(track)), - Track::RemoteVideo(track) => Ok(Self::Remote(track)), - _ => Err("not a video track"), - } - } -} - -impl TryFrom for AudioTrack { - type Error = &'static str; - - fn try_from(track: Track) -> Result { - match track { - Track::LocalAudio(track) => Ok(Self::Local(track)), - Track::RemoteAudio(track) => Ok(Self::Remote(track)), - _ => Err("not an audio track"), - } - } -} - -impl From for MediaType { - fn from(kind: TrackKind) -> Self { - match kind { - TrackKind::Audio => Self::Audio, - TrackKind::Video => Self::Video, - } - } +pub(super) fn update_info(inner: &Arc, _track: &Track, new_info: proto::TrackInfo) { + let mut info = inner.info.write(); + info.name = new_info.name; + info.sid = new_info.sid.into(); + info.kind = TrackKind::try_from(proto::TrackType::from_i32(new_info.r#type).unwrap()).unwrap(); + info.source = TrackSource::from(proto::TrackSource::from_i32(new_info.source).unwrap()); } diff --git a/livekit/src/room/track/remote_audio_track.rs b/livekit/src/room/track/remote_audio_track.rs index 1504311..ab431ac 100644 --- a/livekit/src/room/track/remote_audio_track.rs +++ b/livekit/src/room/track/remote_audio_track.rs @@ -1,15 +1,13 @@ -use super::remote_track; -use super::TrackInner; +use super::{remote_track, TrackInner}; use crate::prelude::*; use livekit_protocol as proto; use livekit_webrtc::prelude::*; use std::fmt::Debug; use std::sync::Arc; -use tokio::sync::mpsc; #[derive(Clone)] pub struct RemoteAudioTrack { - pub(crate) inner: Arc, + inner: Arc, } impl Debug for RemoteAudioTrack { @@ -25,7 +23,7 @@ impl Debug for RemoteAudioTrack { impl RemoteAudioTrack { pub(crate) fn new(sid: TrackSid, name: String, rtc_track: RtcAudioTrack) -> Self { Self { - inner: Arc::new(TrackInner::new( + inner: Arc::new(super::new_inner( sid, name, TrackKind::Audio, @@ -34,78 +32,68 @@ impl RemoteAudioTrack { } } - #[inline] pub fn sid(&self) -> TrackSid { - self.inner.sid() + self.inner.info.read().sid.clone() } - #[inline] pub fn name(&self) -> String { - self.inner.name() + self.inner.info.read().name.clone() } - #[inline] pub fn kind(&self) -> TrackKind { - self.inner.kind() + self.inner.info.read().kind } - #[inline] pub fn source(&self) -> TrackSource { - self.inner.source() + self.inner.info.read().source } - #[inline] pub fn stream_state(&self) -> StreamState { - self.inner.stream_state() + self.inner.info.read().stream_state } - #[inline] pub fn enable(&self) { - self.inner.enable() + self.inner.rtc_track.set_enabled(true); } - #[inline] pub fn disable(&self) { - self.inner.disable() + self.inner.rtc_track.set_enabled(false); } - #[inline] pub fn is_muted(&self) -> bool { - self.inner.is_muted() + self.inner.info.read().muted } - #[inline] pub fn rtc_track(&self) -> RtcAudioTrack { - if let MediaStreamTrack::Audio(audio) = self.inner.rtc_track() { + if let MediaStreamTrack::Audio(audio) = self.inner.rtc_track.clone() { return audio; } - unreachable!() + unreachable!(); } - #[inline] - pub fn register_observer(&self) -> mpsc::UnboundedReceiver { - self.inner.register_observer() - } - - #[inline] pub fn is_remote(&self) -> bool { true } + pub fn on_muted(&self, f: impl Fn(Track) + Send + 'static) { + *self.inner.events.muted.lock() = Some(Box::new(f)); + } + + pub fn on_unmuted(&self, f: impl Fn(Track) + Send + 'static) { + *self.inner.events.unmuted.lock() = Some(Box::new(f)); + } + #[allow(dead_code)] - #[inline] pub(crate) fn transceiver(&self) -> Option { - self.inner.transceiver() + self.inner.info.read().transceiver.clone() } - #[inline] #[allow(dead_code)] - pub(crate) fn update_transceiver(&self, transceiver: Option) { - self.inner.update_transceiver(transceiver) + pub(crate) fn set_transceiver(&self, transceiver: Option) { + self.inner.info.write().transceiver = transceiver; } - #[inline] pub(crate) fn update_info(&self, info: proto::TrackInfo) { - remote_track::update_info(&self.inner, info); + remote_track::update_info(&self.inner, &Track::RemoteAudio(self.clone()), info); } } diff --git a/livekit/src/room/track/remote_track.rs b/livekit/src/room/track/remote_track.rs index f105881..11bca60 100644 --- a/livekit/src/room/track/remote_track.rs +++ b/livekit/src/room/track/remote_track.rs @@ -1,13 +1,10 @@ use super::track_dispatch; use super::TrackInner; -use super::{RemoteAudioTrack, RemoteVideoTrack}; use crate::prelude::*; -use crate::track::TrackEvent; use livekit_protocol as proto; use livekit_protocol::enum_dispatch; use livekit_webrtc::prelude::*; use std::sync::Arc; -use tokio::sync::mpsc; #[derive(Clone, Debug)] pub enum RemoteTrack { @@ -27,7 +24,28 @@ impl RemoteTrack { } } -pub(crate) fn update_info(track: &Arc, new_info: proto::TrackInfo) { - track.update_info(new_info.clone()); - track.set_muted(new_info.muted); +pub(super) fn update_info(inner: &Arc, track: &Track, new_info: proto::TrackInfo) { + super::update_info(inner, track, new_info.clone()); + super::set_muted(inner, track, new_info.muted); +} + +impl From for Track { + fn from(track: RemoteTrack) -> Self { + match track { + RemoteTrack::Audio(track) => Self::RemoteAudio(track), + RemoteTrack::Video(track) => Self::RemoteVideo(track), + } + } +} + +impl TryFrom for RemoteTrack { + type Error = &'static str; + + fn try_from(track: Track) -> Result { + match track { + Track::RemoteAudio(track) => Ok(Self::Audio(track)), + Track::RemoteVideo(track) => Ok(Self::Video(track)), + _ => Err("not a local track"), + } + } } diff --git a/livekit/src/room/track/remote_video_track.rs b/livekit/src/room/track/remote_video_track.rs index 825043b..8e345b0 100644 --- a/livekit/src/room/track/remote_video_track.rs +++ b/livekit/src/room/track/remote_video_track.rs @@ -1,14 +1,14 @@ -use super::{remote_track, TrackInner}; +use super::remote_track; +use super::TrackInner; use crate::prelude::*; use livekit_protocol as proto; use livekit_webrtc::prelude::*; use std::fmt::Debug; use std::sync::Arc; -use tokio::sync::mpsc; #[derive(Clone)] pub struct RemoteVideoTrack { - pub(crate) inner: Arc, + inner: Arc, } impl Debug for RemoteVideoTrack { @@ -24,7 +24,7 @@ impl Debug for RemoteVideoTrack { impl RemoteVideoTrack { pub(crate) fn new(sid: TrackSid, name: String, rtc_track: RtcVideoTrack) -> Self { Self { - inner: Arc::new(TrackInner::new( + inner: Arc::new(super::new_inner( sid, name, TrackKind::Video, @@ -33,78 +33,68 @@ impl RemoteVideoTrack { } } - #[inline] pub fn sid(&self) -> TrackSid { - self.inner.sid() + self.inner.info.read().sid.clone() } - #[inline] pub fn name(&self) -> String { - self.inner.name() + self.inner.info.read().name.clone() } - #[inline] pub fn kind(&self) -> TrackKind { - self.inner.kind() + self.inner.info.read().kind } - #[inline] pub fn source(&self) -> TrackSource { - self.inner.source() + self.inner.info.read().source } - #[inline] pub fn stream_state(&self) -> StreamState { - self.inner.stream_state() + self.inner.info.read().stream_state } - #[inline] pub fn enable(&self) { - self.inner.enable() + self.inner.rtc_track.set_enabled(true); } - #[inline] pub fn disable(&self) { - self.inner.disable() + self.inner.rtc_track.set_enabled(false); } - #[inline] pub fn is_muted(&self) -> bool { - self.inner.is_muted() + self.inner.info.read().muted } - #[inline] pub fn rtc_track(&self) -> RtcVideoTrack { - if let MediaStreamTrack::Video(video) = self.inner.rtc_track() { + if let MediaStreamTrack::Video(video) = self.inner.rtc_track.clone() { return video; } - unreachable!() + unreachable!(); } - #[inline] - pub fn register_observer(&self) -> mpsc::UnboundedReceiver { - self.inner.register_observer() - } - - #[inline] pub fn is_remote(&self) -> bool { true } + pub fn on_muted(&self, f: impl Fn(Track) + Send + 'static) { + *self.inner.events.muted.lock() = Some(Box::new(f)); + } + + pub fn on_unmuted(&self, f: impl Fn(Track) + Send + 'static) { + *self.inner.events.unmuted.lock() = Some(Box::new(f)); + } + #[allow(dead_code)] - #[inline] pub(crate) fn transceiver(&self) -> Option { - self.inner.transceiver() + self.inner.info.read().transceiver.clone() } #[allow(dead_code)] - #[inline] - pub(crate) fn update_transceiver(&self, transceiver: Option) { - self.inner.update_transceiver(transceiver) + pub(crate) fn set_transceiver(&self, transceiver: Option) { + self.inner.info.write().transceiver = transceiver; } - #[inline] pub(crate) fn update_info(&self, info: proto::TrackInfo) { - remote_track::update_info(&self.inner, info); + remote_track::update_info(&self.inner, &Track::RemoteVideo(self.clone()), info); } } diff --git a/livekit/src/room/track/video_track.rs b/livekit/src/room/track/video_track.rs new file mode 100644 index 0000000..1d3ac7f --- /dev/null +++ b/livekit/src/room/track/video_track.rs @@ -0,0 +1,44 @@ +use super::track_dispatch; +use crate::prelude::*; +use livekit_protocol as proto; +use livekit_protocol::enum_dispatch; +use livekit_webrtc::prelude::*; + +#[derive(Clone, Debug)] +pub enum VideoTrack { + Local(LocalVideoTrack), + Remote(RemoteVideoTrack), +} + +impl VideoTrack { + track_dispatch!([Local, Remote]); + + #[inline] + pub fn rtc_track(&self) -> RtcVideoTrack { + match self { + Self::Local(track) => track.rtc_track(), + Self::Remote(track) => track.rtc_track(), + } + } +} + +impl From for Track { + fn from(track: VideoTrack) -> Self { + match track { + VideoTrack::Local(track) => Self::LocalVideo(track), + VideoTrack::Remote(track) => Self::RemoteVideo(track), + } + } +} + +impl TryFrom for VideoTrack { + type Error = &'static str; + + fn try_from(track: Track) -> Result { + match track { + Track::LocalVideo(track) => Ok(Self::Local(track)), + Track::RemoteVideo(track) => Ok(Self::Remote(track)), + _ => Err("not a video track"), + } + } +} diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index 3db651e..56d17e5 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -654,7 +654,10 @@ impl SessionInner { if track.kind() == TrackKind::Video { let capabilities = LkRuntime::instance() .pc_factory() - .get_rtp_sender_capabilities(track.kind().into()); + .get_rtp_sender_capabilities(match track.kind() { + TrackKind::Video => MediaType::Video, + TrackKind::Audio => MediaType::Audio, + }); let mut matched = Vec::new(); let mut partial_matched = Vec::new(); diff --git a/webrtc-sys/src/webrtc.cpp b/webrtc-sys/src/webrtc.cpp index 92182b1..2c9b54d 100644 --- a/webrtc-sys/src/webrtc.cpp +++ b/webrtc-sys/src/webrtc.cpp @@ -42,7 +42,6 @@ static webrtc::Mutex g_mutex{}; static uint32_t g_release_counter(0); RtcRuntime::RtcRuntime() { - rtc::LogMessage::LogToDebug(rtc::LS_INFO); RTC_LOG(LS_VERBOSE) << "RtcRuntime()"; {