From 4465afad0c5292cb3e786b5db9f216ef86d2f38b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Wed, 14 Dec 2022 23:38:20 +0100 Subject: [PATCH] reconnect WIP --- crates/livekit-core/src/events.rs | 187 ++++---- crates/livekit-core/src/lib.rs | 2 +- crates/livekit-core/src/room/mod.rs | 172 ++++--- .../src/room/participant/local_participant.rs | 32 +- .../livekit-core/src/room/participant/mod.rs | 20 +- .../room/participant/remote_participant.rs | 258 +++++----- .../livekit-core/src/room/publication/mod.rs | 7 +- crates/livekit-core/src/room/track/events.rs | 1 - .../src/room/track/local_audio_track.rs | 1 + .../src/room/track/local_video_track.rs | 1 + crates/livekit-core/src/room/track/mod.rs | 5 +- .../src/room/track/remote_audio_track.rs | 1 + .../src/room/track/remote_track.rs | 2 +- .../src/room/track/remote_video_track.rs | 1 + crates/livekit-core/src/rtc_engine/mod.rs | 288 +++++++++--- .../src/rtc_engine/pc_transport.rs | 18 +- crates/livekit-core/src/signal_client/mod.rs | 71 ++- .../libwebrtc-sys/src/webrtc.cpp | 2 +- crates/livekit-webrtc/src/media_stream.rs | 31 +- examples/Cargo.lock | 439 ++++++++++-------- examples/simple_room/Cargo.toml | 1 - examples/simple_room/src/app.rs | 134 ++++-- examples/simple_room/src/events.rs | 9 +- examples/simple_room/src/main.rs | 9 +- examples/simple_room/src/video_renderer.rs | 4 + 25 files changed, 1039 insertions(+), 657 deletions(-) delete mode 100644 crates/livekit-core/src/room/track/events.rs diff --git a/crates/livekit-core/src/events.rs b/crates/livekit-core/src/events.rs index c85506f..7d56e0c 100644 --- a/crates/livekit-core/src/events.rs +++ b/crates/livekit-core/src/events.rs @@ -1,4 +1,13 @@ +use crate::room::id::TrackSid; +use crate::room::participant::remote_participant::RemoteParticipant; +use crate::room::publication::RemoteTrackPublication; +use crate::room::track::remote_track::RemoteTrackHandle; +use crate::room::{ConnectionState, RoomHandle}; +use futures::future::Future; use futures_util::future::BoxFuture; +use parking_lot::Mutex; +use std::fmt::Debug; +use std::sync::Arc; use thiserror::Error; type EventHandler = Box BoxFuture<'static, ()> + Send + Sync>; @@ -21,120 +30,94 @@ pub enum TrackError { TrackNotFound(String), } -pub mod room { - use super::{EventHandler, TrackError}; - use crate::room::id::TrackSid; - use crate::room::participant::remote_participant::RemoteParticipant; - use crate::room::publication::RemoteTrackPublication; - use crate::room::track::remote_track::RemoteTrackHandle; - use crate::room::RoomHandle; - use futures::future::Future; - use parking_lot::Mutex; - use std::sync::Arc; +#[derive(Clone, Debug)] +pub struct ParticipantConnectedEvent { + pub room_handle: RoomHandle, + pub participant: Arc, +} - #[derive(Clone)] - pub struct ParticipantConnectedEvent { - pub room_handle: RoomHandle, - pub participant: Arc, - } +#[derive(Clone, Debug)] +pub struct ParticipantDisconnectedEvent { + pub room_handle: RoomHandle, + pub participant: Arc, +} - #[derive(Clone)] - pub struct ParticipantDisconnectedEvent { - pub room_handle: RoomHandle, - pub participant: Arc, - } +#[derive(Clone, Debug)] +pub struct TrackSubscribedEvent { + pub room_handle: RoomHandle, + pub track: RemoteTrackHandle, + pub publication: RemoteTrackPublication, + pub participant: Arc, +} - #[derive(Clone)] - pub struct TrackSubscribedEvent { - pub room_handle: RoomHandle, - pub track: RemoteTrackHandle, - pub publication: RemoteTrackPublication, - pub participant: Arc, - } +#[derive(Clone, Debug)] +pub struct TrackPublishedEvent { + pub room_handle: RoomHandle, + pub publication: RemoteTrackPublication, + pub participant: Arc, +} - #[derive(Clone)] - pub struct TrackPublishedEvent { - pub room_handle: RoomHandle, - pub publication: RemoteTrackPublication, - pub participant: Arc, - } +#[derive(Clone, Debug)] +pub struct TrackSubscriptionFailedEvent { + pub room_handle: RoomHandle, + pub error: TrackError, + pub sid: TrackSid, + pub participant: Arc, +} - #[derive(Clone)] - pub struct TrackSubscriptionFailedEvent { - pub room_handle: RoomHandle, - pub error: TrackError, - pub sid: TrackSid, - pub participant: Arc, - } +#[derive(Clone, Debug)] +pub struct ConnectionStateChangedEvent { + pub room_handle: RoomHandle, + pub state: ConnectionState, +} - pub(crate) type OnParticipantConnectedHandler = EventHandler; - pub(crate) type OnParticipantDisconnectedHandler = EventHandler; - pub(crate) type OnTrackSubscribedEventHandler = EventHandler; - pub(crate) type OnTrackPublishedEventHandler = EventHandler; - pub(crate) type OnTrackSubscriptionFailedHandler = EventHandler; +pub(crate) type OnParticipantConnectedHandler = EventHandler; +pub(crate) type OnParticipantDisconnectedHandler = EventHandler; +pub(crate) type OnTrackSubscribedHandler = EventHandler; +pub(crate) type OnTrackPublishedHandler = EventHandler; +pub(crate) type OnTrackSubscriptionFailedHandler = EventHandler; +pub(crate) type OnConnectionStateChangedHandler = EventHandler; - #[derive(Default)] - pub struct RoomEvents { - pub(crate) on_participant_connected: Mutex>, - pub(crate) on_participant_disconnected: Mutex>, - pub(crate) on_track_subscribed: Mutex>, - pub(crate) on_track_published: Mutex>, - pub(crate) on_track_subscription_failed: Mutex>, - } +#[derive(Default)] +pub struct RoomEvents { + pub(crate) on_participant_connected: Mutex>, + pub(crate) on_participant_disconnected: Mutex>, + pub(crate) on_track_subscribed: Mutex>, + pub(crate) on_track_published: Mutex>, + pub(crate) on_track_subscription_failed: Mutex>, + pub(crate) on_connection_state_changed: Mutex>, +} - impl RoomEvents { - event_setter!(on_participant_connected, ParticipantConnectedEvent); - event_setter!(on_participant_disconnected, ParticipantDisconnectedEvent); - event_setter!(on_track_subscribed, TrackSubscribedEvent); - event_setter!(on_track_published, TrackPublishedEvent); - event_setter!(on_track_subscription_failed, TrackSubscriptionFailedEvent); +impl Debug for RoomEvents { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "RoomEvents") } } -pub mod participant { - use super::{EventHandler, TrackError}; - use crate::room::id::TrackSid; - use crate::room::participant::remote_participant::RemoteParticipant; - use crate::room::publication::RemoteTrackPublication; - use crate::room::track::remote_track::RemoteTrackHandle; - use futures::future::Future; - use parking_lot::Mutex; - use std::sync::Arc; +impl RoomEvents { + event_setter!(on_participant_connected, ParticipantConnectedEvent); + event_setter!(on_participant_disconnected, ParticipantDisconnectedEvent); + event_setter!(on_track_subscribed, TrackSubscribedEvent); + event_setter!(on_track_published, TrackPublishedEvent); + event_setter!(on_track_subscription_failed, TrackSubscriptionFailedEvent); + event_setter!(on_connection_state_changed, ConnectionStateChangedEvent); +} - #[derive(Clone)] - pub struct TrackPublishedEvent { - pub publication: RemoteTrackPublication, - pub participant: Arc, - } +#[derive(Default)] +pub struct ParticipantEvents { + pub(crate) on_track_published: Mutex>, + pub(crate) on_track_subscribed: Mutex>, + pub(crate) on_track_subscription_failed: Mutex>, +} - #[derive(Clone)] - pub struct TrackSubscribedEvent { - pub track: RemoteTrackHandle, - pub publication: RemoteTrackPublication, - pub participant: Arc, - } - - #[derive(Clone)] - pub struct TrackSubscriptionFailedEvent { - pub sid: TrackSid, - pub error: TrackError, - pub participant: Arc, - } - - pub(crate) type TrackPublishedHandler = EventHandler; - pub(crate) type TrackSubscribedHandler = EventHandler; - pub(crate) type TrackSubscriptionFailedHandler = EventHandler; - - #[derive(Default)] - pub struct ParticipantEvents { - pub(crate) on_track_published: Mutex>, - pub(crate) on_track_subscribed: Mutex>, - pub(crate) on_track_subscription_failed: Mutex>, - } - - impl ParticipantEvents { - event_setter!(on_track_published, TrackPublishedEvent); - event_setter!(on_track_subscribed, TrackSubscribedEvent); - event_setter!(on_track_subscription_failed, TrackSubscriptionFailedEvent); +impl Debug for ParticipantEvents { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "ParticipantEvents") } } + +impl ParticipantEvents { + event_setter!(on_track_published, TrackPublishedEvent); + event_setter!(on_track_subscribed, TrackSubscribedEvent); + event_setter!(on_track_subscription_failed, TrackSubscriptionFailedEvent); +} diff --git a/crates/livekit-core/src/lib.rs b/crates/livekit-core/src/lib.rs index cbdb92a..0dd45e4 100644 --- a/crates/livekit-core/src/lib.rs +++ b/crates/livekit-core/src/lib.rs @@ -4,8 +4,8 @@ pub mod proto { include!(concat!(env!("OUT_DIR"), "/livekit.rs")); } -mod events; mod rtc_engine; mod signal_client; +pub mod events; pub mod room; diff --git a/crates/livekit-core/src/room/mod.rs b/crates/livekit-core/src/room/mod.rs index f9305ea..6c6648c 100644 --- a/crates/livekit-core/src/room/mod.rs +++ b/crates/livekit-core/src/room/mod.rs @@ -1,21 +1,21 @@ -use parking_lot::lock_api::RwLockUpgradableReadGuard; use parking_lot::{Mutex, RwLock}; use std::collections::HashMap; -use std::sync::atomic::AtomicU8; +use std::sync::atomic::{AtomicU8, Ordering}; use std::sync::Arc; -use self::id::ParticipantSid; +use self::id::{ParticipantIdentity, ParticipantSid}; use self::participant::local_participant::LocalParticipant; use self::participant::remote_participant::RemoteParticipant; use self::participant::ParticipantInternalTrait; use self::participant::ParticipantTrait; -use crate::events::room::{ - ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents, TrackSubscribedEvent, +use crate::events::{ + ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents, TrackPublishedEvent, + TrackSubscribedEvent, }; use crate::proto; use crate::proto::participant_info; use thiserror::Error; -use tracing::{debug, error}; +use tracing::{debug, error, instrument, trace_span, Level}; use crate::rtc_engine::{EngineError, EngineEvent, EngineEvents, RTCEngine}; use crate::signal_client::SignalOptions; @@ -27,15 +27,15 @@ pub mod track; #[derive(Error, Debug)] pub enum RoomError { - #[error("internal RTCEngine failure")] + #[error("engine : {0}")] Engine(#[from] EngineError), - #[error("internal Room failure")] + #[error("room failure: {0}")] Internal(String), } -type RoomResult = Result; +pub type RoomResult = Result; -#[derive(Debug)] +#[derive(Debug, Clone, Copy, Eq, PartialEq)] pub enum ConnectionState { Disconnected, Connecting, @@ -43,6 +43,7 @@ pub enum ConnectionState { Reconnecting, } +#[derive(Debug)] struct RoomInner { state: AtomicU8, // ConnectionState sid: Mutex, @@ -52,6 +53,7 @@ struct RoomInner { local_participant: Arc, } +#[derive(Debug)] pub struct Room { inner: Option>, events: Arc, @@ -65,14 +67,19 @@ impl Room { } } + #[instrument(level = Level::DEBUG)] pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> { let (rtc_engine, engine_events) = RTCEngine::connect(url, token, SignalOptions::default()).await?; let rtc_engine = Arc::new(rtc_engine); let join_response = rtc_engine.join_response(); + let pi = join_response.participant.unwrap().clone(); let local_participant = Arc::new(LocalParticipant::new( rtc_engine.clone(), - join_response.participant.unwrap().clone(), + pi.sid.into(), + pi.identity.into(), + pi.name, + pi.metadata, )); let room_info = join_response.room.unwrap(); let inner = Arc::new(RoomInner { @@ -84,14 +91,25 @@ impl Room { local_participant, }); - self.inner = Some(inner.clone()); - - // Add already connected participants for pi in join_response.other_participants { - let p = Self::create_participant(inner.clone(), self.events.clone(), pi.clone()); - p.update_info(pi).await; + let participant = { + let pi = pi.clone(); + Self::create_participant( + inner.clone(), + self.events.clone(), + pi.sid.into(), + pi.identity.into(), + pi.name, + pi.metadata, + ) + }; + participant.update_info(pi.clone()); + participant + .update_tracks(RoomHandle::from(inner.clone()), pi.tracks) + .await; } + self.inner = Some(inner.clone()); tokio::spawn(Self::room_task(inner, self.events.clone(), engine_events)); Ok(()) @@ -121,6 +139,7 @@ impl Room { } } + #[instrument(level = Level::DEBUG, skip(room_inner, room_events))] async fn handle_event( room_inner: Arc, room_events: Arc, @@ -155,10 +174,18 @@ impl Room { Self::get_participant(room_inner.clone(), &participant_sid.to_string().into()); if let Some(remote_participant) = remote_participant { - remote_participant.add_subscribed_media_track( - track_sid.to_string().into(), - rtp_receiver.track(), - ); + tokio::spawn({ + let track_sid = track_sid.to_owned().into(); + async move { + remote_participant + .add_subscribed_media_track( + RoomHandle::from(room_inner), + track_sid, + rtp_receiver.track(), + ) + .await; + } + }); } else { // The server should send participant updates before sending a new offer // So this should not happen. @@ -173,6 +200,7 @@ impl Room { Ok(()) } + #[instrument(level = Level::DEBUG, skip(room_inner, room_events))] async fn handle_participant_update( room_inner: Arc, room_events: Arc, @@ -182,7 +210,7 @@ impl Room { if pi.sid == room_inner.local_participant.sid() || pi.identity == room_inner.local_participant.identity() { - room_inner.local_participant.clone().update_info(pi).await; + room_inner.local_participant.clone().update_info(pi); continue; } @@ -199,12 +227,24 @@ impl Room { ) } else { // Participant is already connected, update the informations - remote_participant.update_info(pi).await; + remote_participant.update_info(pi.clone()); + remote_participant + .update_tracks(RoomHandle::from(room_inner.clone()), pi.tracks) + .await; } } else { // Create a new participant and call OnConnect event - let remote_participant = - Self::create_participant(room_inner.clone(), room_events.clone(), pi); + let remote_participant = { + let pi = pi.clone(); + Self::create_participant( + room_inner.clone(), + room_events.clone(), + pi.sid.into(), + pi.identity.into(), + pi.name, + pi.metadata, + ) + }; let mut handler = room_events.on_participant_connected.lock(); if let Some(cb) = handler.as_mut() { cb(ParticipantConnectedEvent { @@ -212,10 +252,16 @@ impl Room { participant: remote_participant.clone(), }); } + + remote_participant.update_info(pi.clone()); + remote_participant + .update_tracks(RoomHandle::from(room_inner.clone()), pi.tracks) + .await; } } } + #[instrument(level = Level::DEBUG, skip(room_inner, room_events))] fn handle_participant_disconnect( room_inner: Arc, room_events: Arc, @@ -247,42 +293,64 @@ impl Room { fn create_participant( room_inner: Arc, room_events: Arc, - pi: proto::ParticipantInfo, + sid: ParticipantSid, + identity: ParticipantIdentity, + name: String, + metadata: String, ) -> Arc { - let p = Arc::new(RemoteParticipant::new(pi.clone())); + let p = Arc::new(RemoteParticipant::new( + sid.clone(), + identity, + name, + metadata, + )); + + macro_rules! forward_event { + ($type:ident, when_connected) => { + p.internal_events().$type({ + let room_events = room_events.clone(); + let room_inner = room_inner.clone(); + move |event| { + let room_events = room_events.clone(); + let room_inner = room_inner.clone(); + async move { + if room_inner.state.load(Ordering::SeqCst) + == ConnectionState::Connected as u8 + { + if let Some(cb) = room_events.$type.lock().as_mut() { + cb(event).await; + } + } + } + } + }) + }; + ($type:ident) => { + p.internal_events().$type({ + let room_events = room_events.clone(); + move |event| { + let room_events = room_events.clone(); + async move { + if let Some(cb) = room_events.$type.lock().as_mut() { + cb(event).await; + } + } + } + }) + }; + } // Forward participantevents to room events - p.internal_events().on_track_subscribed({ - let room_events = room_events.clone(); - let room_inner = room_inner.clone(); + forward_event!(on_track_published, when_connected); + forward_event!(on_track_subscribed); + forward_event!(on_track_subscription_failed); - move |event| { - let room_events = room_events.clone(); - let room_inner = room_inner.clone(); - - async move { - if let Some(cb) = room_events.clone().on_track_subscribed.lock().as_mut() { - cb(TrackSubscribedEvent { - room_handle: RoomHandle::from(room_inner.clone()), - track: event.track, - participant: event.participant, - publication: event.publication, - }) - .await; - } - } - } - }); - - room_inner - .participants - .write() - .insert(pi.sid.into(), p.clone()); + room_inner.participants.write().insert(sid, p.clone()); p } } -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct RoomHandle { inner: Arc, } diff --git a/crates/livekit-core/src/room/participant/local_participant.rs b/crates/livekit-core/src/room/participant/local_participant.rs index 7927e21..05ce33d 100644 --- a/crates/livekit-core/src/room/participant/local_participant.rs +++ b/crates/livekit-core/src/room/participant/local_participant.rs @@ -1,22 +1,28 @@ +use std::sync::Weak; + use crate::proto::{data_packet, DataPacket, UserPacket}; -use crate::room::participant::{impl_participant_trait, ParticipantShared, ParticipantInternalTrait}; -use crate::room::RoomError; +use crate::room::participant::{ + impl_participant_trait, ParticipantInternalTrait, ParticipantShared, +}; +use crate::room::{RoomError, RoomInner}; use crate::rtc_engine::RTCEngine; +#[derive(Debug)] pub struct LocalParticipant { shared: ParticipantShared, rtc_engine: Arc, } impl LocalParticipant { - pub(crate) fn new(rtc_engine: Arc, info: ParticipantInfo) -> Self { + pub(crate) fn new( + rtc_engine: Arc, + sid: ParticipantSid, + identity: ParticipantIdentity, + name: String, + metadata: String, + ) -> Self { Self { - shared: ParticipantShared::new( - info.sid.into(), - info.identity.into(), - info.name, - info.metadata, - ), + shared: ParticipantShared::new(sid, identity, name, metadata), rtc_engine, } } @@ -40,16 +46,16 @@ impl LocalParticipant { .await .map_err(Into::into) } - - pub(crate) async fn update_info(self: Arc, info: ParticipantInfo) { - self.shared.update_info(info); - } } impl ParticipantInternalTrait for LocalParticipant { fn internal_events(&self) -> Arc { self.shared.internal_events.clone() } + + fn update_info(&self, info: ParticipantInfo) { + self.shared.update_info(info); + } } impl_participant_trait!(LocalParticipant); diff --git a/crates/livekit-core/src/room/participant/mod.rs b/crates/livekit-core/src/room/participant/mod.rs index 49b6f0d..a243691 100644 --- a/crates/livekit-core/src/room/participant/mod.rs +++ b/crates/livekit-core/src/room/participant/mod.rs @@ -1,10 +1,9 @@ -use crate::events::participant::ParticipantEvents; +use crate::events::ParticipantEvents; use crate::proto::ParticipantInfo; use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid}; use crate::room::participant::local_participant::LocalParticipant; use crate::room::participant::remote_participant::RemoteParticipant; use crate::room::publication::{TrackPublication, TrackPublicationTrait}; -use futures_util::future::BoxFuture; use livekit_utils::enum_dispatch; use parking_lot::{Mutex, RwLock}; use std::collections::HashMap; @@ -13,8 +12,7 @@ use std::sync::Arc; pub mod local_participant; pub mod remote_participant; -type OnTrackSubscribed = Box BoxFuture<'static, ()> + Send + Sync>; - +#[derive(Debug)] pub(super) struct ParticipantShared { pub(super) events: Arc, pub(super) internal_events: Arc, @@ -57,6 +55,7 @@ impl ParticipantShared { pub(crate) trait ParticipantInternalTrait { fn internal_events(&self) -> Arc; + fn update_info(&self, info: ParticipantInfo); } pub trait ParticipantTrait { @@ -73,20 +72,11 @@ pub enum ParticipantHandle { Remote(Arc), } -impl ParticipantHandle { - // TODO(theomonnom): Add async support to wrap_variants ... - pub(crate) async fn update_info(&self, info: ParticipantInfo) { - match self { - Self::Local(inner) => inner.clone().update_info(info).await, - Self::Remote(inner) => inner.clone().update_info(info).await, - } - } -} - impl ParticipantInternalTrait for ParticipantHandle { enum_dispatch!( [Local, Remote] fnc!(internal_events, &Self, [], Arc); + fnc!(update_info, &Self, [info: ParticipantInfo], ()); ); } @@ -103,7 +93,7 @@ impl ParticipantTrait for ParticipantHandle { macro_rules! impl_participant_trait { ($x:ty) => { - use crate::events::participant::ParticipantEvents; + use crate::events::ParticipantEvents; use crate::proto::ParticipantInfo; use crate::room::id::{ParticipantIdentity, ParticipantSid}; use std::sync::Arc; diff --git a/crates/livekit-core/src/room/participant/remote_participant.rs b/crates/livekit-core/src/room/participant/remote_participant.rs index 12efbae..81653c4 100644 --- a/crates/livekit-core/src/room/participant/remote_participant.rs +++ b/crates/livekit-core/src/room/participant/remote_participant.rs @@ -1,7 +1,7 @@ -use crate::events::participant::{ - TrackPublishedEvent, TrackSubscribedEvent, TrackSubscriptionFailedEvent, +use crate::events::{ + TrackError, TrackPublishedEvent, TrackSubscribedEvent, TrackSubscriptionFailedEvent, }; -use crate::events::TrackError; +use crate::proto::TrackInfo; use crate::room::id::TrackSid; use crate::room::participant::{ impl_participant_trait, ParticipantInternalTrait, ParticipantShared, @@ -12,141 +12,35 @@ use crate::room::publication::{ use crate::room::track::remote_audio_track::RemoteAudioTrack; use crate::room::track::remote_track::RemoteTrackHandle; use crate::room::track::remote_video_track::RemoteVideoTrack; -use crate::room::track::{TrackKind, TrackTrait, TrackHandle}; +use crate::room::track::{TrackKind, TrackTrait}; +use crate::room::{RoomHandle, RoomInner}; use livekit_webrtc::media_stream::MediaStreamTrackHandle; use std::collections::HashSet; use std::time::Duration; use tokio::time::{sleep, timeout}; -use tracing::{info, error}; +use tracing::{debug, debug_span, error, instrument, Instrument, Level}; use super::ParticipantTrait; const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5); +#[derive(Debug)] pub struct RemoteParticipant { shared: ParticipantShared, } impl RemoteParticipant { - pub(crate) fn new(info: ParticipantInfo) -> Self { + pub(crate) fn new( + sid: ParticipantSid, + identity: ParticipantIdentity, + name: String, + metadata: String, + ) -> Self { Self { - shared: ParticipantShared::new( - info.sid.into(), - info.identity.into(), - info.name, - info.metadata, - ), + shared: ParticipantShared::new(sid, identity, name, metadata), } } - pub(crate) fn add_subscribed_media_track( - self: Arc, - sid: TrackSid, - media_track: MediaStreamTrackHandle, - ) { - tokio::spawn(async move { - let wait_publication = { - let participant = self.clone(); - let sid = sid.clone(); - async move { - loop { - let publication = participant.get_track_publication(&sid); - if let Some(publication) = publication { - return publication; - } - - sleep(Duration::from_millis(50)).await; - } - } - }; - - if let Ok(remote_publication) = timeout(ADD_TRACK_TIMEOUT, wait_publication).await { - let track = match remote_publication.kind() { - TrackKind::Audio => { - if let MediaStreamTrackHandle::Audio(rtc_track) = media_track { - let audio_track = RemoteAudioTrack::new( - remote_publication.sid().into(), - remote_publication.name(), - rtc_track, - ); - RemoteTrackHandle::Audio(Arc::new(audio_track)) - } else { - unreachable!(); - } - } - TrackKind::Video => { - if let MediaStreamTrackHandle::Video(rtc_track) = media_track { - let video_track = RemoteVideoTrack::new( - remote_publication.sid().into(), - remote_publication.name(), - rtc_track, - ); - RemoteTrackHandle::Video(Arc::new(video_track)) - } else { - unreachable!() - } - } - _ => unreachable!(), - }; - - info!("starting track: {:?}", sid); - - remote_publication.update_track(Some(track.clone().into())); - self.shared - .add_track_publication(TrackPublication::Remote(remote_publication.clone())); - track.start(); - - let event = TrackSubscribedEvent { - track, - publication: remote_publication, - participant: self.clone(), - }; - - if let Some(cb) = self - .shared - .internal_events - .on_track_subscribed - .lock() - .as_mut() - { - cb(event.clone()).await; - } - - if let Some(cb) = self.shared.events.on_track_subscribed.lock().as_mut() { - cb(event).await; - } - } else { - error!("could not find published track with sid: {:?}", sid); - - let event = TrackSubscriptionFailedEvent { - sid: sid.clone(), - error: TrackError::TrackNotFound(sid.clone().to_string()), - participant: self.clone(), - }; - - if let Some(cb) = self - .shared - .internal_events - .on_track_subscription_failed - .lock() - .as_mut() - { - cb(event.clone()).await; - } - - if let Some(cb) = self - .shared - .events - .on_track_subscription_failed - .lock() - .as_mut() - { - cb(event).await; - } - } - }); - } - fn get_track_publication(&self, sid: &TrackSid) -> Option { self.shared.tracks.read().get(sid).map(|track| { if let TrackPublication::Remote(remote) = track { @@ -157,12 +51,125 @@ impl RemoteParticipant { }) } - pub(crate) async fn update_info(self: Arc, info: ParticipantInfo) { - self.shared.update_info(info.clone()); + #[instrument(level = Level::DEBUG, skip(room_handle))] + pub(crate) async fn add_subscribed_media_track( + self: Arc, + room_handle: RoomHandle, + sid: TrackSid, + media_track: MediaStreamTrackHandle, + ) { + let wait_publication = { + let participant = self.clone(); + let sid = sid.clone(); + async move { + loop { + let publication = participant.get_track_publication(&sid); + if let Some(publication) = publication { + return publication; + } + sleep(Duration::from_millis(50)).await; + } + } + }; + + if let Ok(remote_publication) = timeout(ADD_TRACK_TIMEOUT, wait_publication).await { + let track = match remote_publication.kind() { + TrackKind::Audio => { + if let MediaStreamTrackHandle::Audio(rtc_track) = media_track { + let audio_track = RemoteAudioTrack::new( + remote_publication.sid().into(), + remote_publication.name(), + rtc_track, + ); + RemoteTrackHandle::Audio(Arc::new(audio_track)) + } else { + unreachable!(); + } + } + TrackKind::Video => { + if let MediaStreamTrackHandle::Video(rtc_track) = media_track { + let video_track = RemoteVideoTrack::new( + remote_publication.sid().into(), + remote_publication.name(), + rtc_track, + ); + RemoteTrackHandle::Video(Arc::new(video_track)) + } else { + unreachable!() + } + } + _ => unreachable!(), + }; + + debug!("starting track: {:?}", sid); + + remote_publication.update_track(Some(track.clone().into())); + self.shared + .add_track_publication(TrackPublication::Remote(remote_publication.clone())); + track.start(); + + let event = TrackSubscribedEvent { + room_handle, + track, + publication: remote_publication, + participant: self.clone(), + }; + + if let Some(cb) = self + .shared + .internal_events + .on_track_subscribed + .lock() + .as_mut() + { + cb(event.clone()).await; + } + + if let Some(cb) = self.shared.events.on_track_subscribed.lock().as_mut() { + cb(event).await; + } + } else { + error!("could not find published track with sid: {:?}", sid); + + let event = TrackSubscriptionFailedEvent { + room_handle, + sid: sid.clone(), + error: TrackError::TrackNotFound(sid.clone().to_string()), + participant: self.clone(), + }; + + if let Some(cb) = self + .shared + .internal_events + .on_track_subscription_failed + .lock() + .as_mut() + { + cb(event.clone()).await; + } + + if let Some(cb) = self + .shared + .events + .on_track_subscription_failed + .lock() + .as_mut() + { + cb(event).await; + } + } + } + + #[instrument(level = Level::DEBUG, skip(room_handle))] + pub(crate) async fn update_tracks( + self: Arc, + room_handle: RoomHandle, + tracks: Vec, + ) { let mut valid_tracks = HashSet::::new(); - for track in info.tracks { + for track in tracks { if let Some(publication) = self.get_track_publication(&track.sid.clone().into()) { publication.update_info(track.clone()); } else { @@ -172,6 +179,7 @@ impl RemoteParticipant { // This is a new track, fire publish events let event = TrackPublishedEvent { + room_handle: room_handle.clone(), participant: self.clone(), publication: publication.clone(), }; @@ -200,6 +208,10 @@ impl ParticipantInternalTrait for RemoteParticipant { fn internal_events(&self) -> Arc { self.shared.internal_events.clone() } + + fn update_info(&self, info: ParticipantInfo) { + self.shared.update_info(info) + } } impl_participant_trait!(RemoteParticipant); diff --git a/crates/livekit-core/src/room/publication/mod.rs b/crates/livekit-core/src/room/publication/mod.rs index 547e825..7f7864d 100644 --- a/crates/livekit-core/src/room/publication/mod.rs +++ b/crates/livekit-core/src/room/publication/mod.rs @@ -25,6 +25,7 @@ pub trait TrackPublicationTrait { fn simulcasted(&self) -> bool; } +#[derive(Debug)] pub(super) struct TrackPublicationShared { pub(super) track: Mutex>, pub(super) name: Mutex, @@ -75,7 +76,7 @@ impl TrackPublicationShared { } } -#[derive(Clone)] +#[derive(Clone, Debug)] pub enum TrackPublication { Local(LocalTrackPublication), Remote(RemoteTrackPublication), @@ -146,7 +147,7 @@ macro_rules! impl_publication_trait { }; } -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct LocalTrackPublication { shared: Arc, } @@ -161,7 +162,7 @@ impl LocalTrackPublication { } } -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct RemoteTrackPublication { shared: Arc, } diff --git a/crates/livekit-core/src/room/track/events.rs b/crates/livekit-core/src/room/track/events.rs deleted file mode 100644 index 86d8cad..0000000 --- a/crates/livekit-core/src/room/track/events.rs +++ /dev/null @@ -1 +0,0 @@ -pub struct TrackEvents {} diff --git a/crates/livekit-core/src/room/track/local_audio_track.rs b/crates/livekit-core/src/room/track/local_audio_track.rs index c194c34..057796a 100644 --- a/crates/livekit-core/src/room/track/local_audio_track.rs +++ b/crates/livekit-core/src/room/track/local_audio_track.rs @@ -1,5 +1,6 @@ use crate::room::track::{impl_track_trait, TrackShared}; +#[derive(Debug)] pub struct LocalAudioTrack { shared: TrackShared, } diff --git a/crates/livekit-core/src/room/track/local_video_track.rs b/crates/livekit-core/src/room/track/local_video_track.rs index cb3f0d1..c46b155 100644 --- a/crates/livekit-core/src/room/track/local_video_track.rs +++ b/crates/livekit-core/src/room/track/local_video_track.rs @@ -1,5 +1,6 @@ use crate::room::track::{impl_track_trait, TrackShared}; +#[derive(Debug)] pub struct LocalVideoTrack { shared: TrackShared, } diff --git a/crates/livekit-core/src/room/track/mod.rs b/crates/livekit-core/src/room/track/mod.rs index ad9e218..28d4a96 100644 --- a/crates/livekit-core/src/room/track/mod.rs +++ b/crates/livekit-core/src/room/track/mod.rs @@ -11,7 +11,6 @@ use std::sync::atomic::AtomicU8; use std::sync::Arc; pub mod audio_track; -pub mod events; pub mod local_audio_track; pub mod local_track; pub mod local_video_track; @@ -97,6 +96,7 @@ impl From for TrackSource { } } +#[derive(Clone, Copy, Debug)] pub struct TrackDimension(pub u32, pub u32); pub trait TrackTrait { @@ -108,6 +108,7 @@ pub trait TrackTrait { fn stop(&self); } +#[derive(Debug)] pub(super) struct TrackShared { pub(super) sid: Mutex, pub(super) name: Mutex, @@ -141,7 +142,7 @@ impl TrackShared { } } -#[derive(Clone)] +#[derive(Clone, Debug)] pub enum TrackHandle { LocalVideo(Arc), LocalAudio(Arc), diff --git a/crates/livekit-core/src/room/track/remote_audio_track.rs b/crates/livekit-core/src/room/track/remote_audio_track.rs index 933f307..1ce12c6 100644 --- a/crates/livekit-core/src/room/track/remote_audio_track.rs +++ b/crates/livekit-core/src/room/track/remote_audio_track.rs @@ -2,6 +2,7 @@ use crate::room::track::{impl_track_trait, TrackShared}; use livekit_webrtc::media_stream::{AudioTrack, MediaStreamTrackHandle}; use std::sync::Arc; +#[derive(Debug)] pub struct RemoteAudioTrack { shared: TrackShared, } diff --git a/crates/livekit-core/src/room/track/remote_track.rs b/crates/livekit-core/src/room/track/remote_track.rs index 81e53ea..c4badca 100644 --- a/crates/livekit-core/src/room/track/remote_track.rs +++ b/crates/livekit-core/src/room/track/remote_track.rs @@ -9,7 +9,7 @@ use livekit_utils::enum_dispatch; use super::TrackTrait; -#[derive(Clone)] +#[derive(Clone, Debug)] pub enum RemoteTrackHandle { Audio(Arc), Video(Arc), diff --git a/crates/livekit-core/src/room/track/remote_video_track.rs b/crates/livekit-core/src/room/track/remote_video_track.rs index 57caf67..b3fc13f 100644 --- a/crates/livekit-core/src/room/track/remote_video_track.rs +++ b/crates/livekit-core/src/room/track/remote_video_track.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use crate::room::track::{impl_track_trait, TrackShared}; +#[derive(Debug)] pub struct RemoteVideoTrack { shared: TrackShared, } diff --git a/crates/livekit-core/src/rtc_engine/mod.rs b/crates/livekit-core/src/rtc_engine/mod.rs index 4fbb08a..95dba5d 100644 --- a/crates/livekit-core/src/rtc_engine/mod.rs +++ b/crates/livekit-core/src/rtc_engine/mod.rs @@ -1,4 +1,5 @@ use parking_lot::Mutex; +use std::error; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::{Arc, Weak}; use std::time::Duration; @@ -10,7 +11,7 @@ use prost::Message; use serde::{Deserialize, Serialize}; use thiserror::Error; use tokio::time::sleep; -use tracing::{debug, error, info, trace}; +use tracing::{debug, error, info, trace, warn}; use crate::{proto, signal_client}; use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState}; @@ -46,6 +47,10 @@ pub(crate) type EngineEmitter = mpsc::Sender; pub(crate) type EngineEvents = mpsc::Receiver; pub(crate) type EngineResult = Result; +// TODO(theomonnom): Smarter retry intervals +pub(crate) const RECONNECT_ATTEMPTS: u32 = 10; +pub(crate) const RECONNECT_INTERVAL: Duration = Duration::from_millis(300); + pub(crate) const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); pub(crate) const LOSSY_DC_LABEL: &str = "_lossy"; pub(crate) const RELIABLE_DC_LABEL: &str = "_reliable"; @@ -69,7 +74,7 @@ struct IceCandidateJSON { #[derive(Error, Debug)] pub enum EngineError { - #[error("signal failure")] + #[error("signal failure: {0}")] Signal(#[from] SignalError), #[error("internal webrtc failure")] Rtc(#[from] RTCError), @@ -94,13 +99,25 @@ pub(crate) enum EngineEvent { rtp_receiver: RtpReceiver, streams: Vec, }, + Connected, + Resuming, + Resumed, + SignalResumed, + Restarting, + Restarted, } #[derive(Debug)] struct EngineInner { - has_published: AtomicBool, + // Join infornation + url: String, + token: Mutex, // The token is refreshed periodically + options: Mutex, join_response: Mutex, + + has_published: AtomicBool, pc_state: AtomicU8, // Casted to PCState enum + reconnecting: AtomicBool, publisher_pc: AsyncMutex, subscriber_pc: AsyncMutex, @@ -109,12 +126,13 @@ struct EngineInner { // Used to send data to other participants ( The SFU forward the messages ) lossy_dc: Mutex, reliable_dc: Mutex, - // Subscriber data channels // These fields are never used, we just keep a strong reference to them, // so we can receive data from other participants sub_reliable_dc: Mutex>, sub_lossy_dc: Mutex>, + + closed: AtomicBool, } #[derive(Debug)] @@ -126,7 +144,72 @@ pub struct RTCEngine { lk_runtime: Arc, // Keep a reference while we're using the RTCEngine } +impl EngineInner { + async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> { + if !self.join_response.lock().subscriber_primary { + return Ok(()); + } + + let publisher = &self.publisher_pc; + { + let mut publisher = publisher.lock().await; + if !publisher.is_connected() + && publisher.peer_connection().ice_connection_state() + != IceConnectionState::IceConnectionChecking + { + let _ = self.negotiate_publisher().await; + } + } + + let dc = self.data_channel(kind); + if dc.lock().state() == DataState::Open { + return Ok(()); + } + + // Wait until the PeerConnection is connected + let wait_connected = async move { + while publisher.lock().await.is_connected() && dc.lock().state() == DataState::Open { + sleep(Duration::from_millis(50)).await; + } + }; + + tokio::select! { + _ = wait_connected => Ok(()), + _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => { + let err = EngineError::Connection("could not establish publisher connection: timeout".to_string()); + error!(error = ?err); + Err(err) + } + } + } + + async fn negotiate_publisher(&self) -> EngineResult<()> { + self.has_published.store(true, Ordering::SeqCst); + if let Err(err) = self.publisher_pc.lock().await.negotiate().await { + error!("failed to negotiate the publisher: {:?}", err); + Err(err)? + } else { + Ok(()) + } + } + + fn data_channel(&self, kind: data_packet::Kind) -> &Mutex { + if kind == data_packet::Kind::Reliable { + &self.reliable_dc + } else { + &self.lossy_dc + } + } +} + impl RTCEngine { + pub fn new() -> Self { + + Self { + + } + } + #[tracing::instrument(skip(url, token))] pub(crate) async fn connect( url: &str, @@ -172,16 +255,16 @@ impl RTCEngine { emitter.clone(), )); + if !join_response.subscriber_primary { + engine_inner.negotiate_publisher().await?; + } + let rtc_engine = Self { signal_client, engine_inner, lk_runtime, }; - if !join_response.subscriber_primary { - rtc_engine.negotiate_publisher().await?; - } - Ok((rtc_engine, events)) } @@ -191,8 +274,9 @@ impl RTCEngine { data: &DataPacket, kind: data_packet::Kind, ) -> Result<(), EngineError> { - self.ensure_publisher_connected(kind).await?; - self.data_channel(kind) + self.engine_inner.ensure_publisher_connected(kind).await?; + self.engine_inner + .data_channel(kind) .lock() .send(&data.encode_to_vec(), true) .map_err(Into::into) @@ -244,7 +328,11 @@ impl RTCEngine { } } SignalEvent::Close => { - // Try reconnect if this isn't expected + Self::handle_disconnected( + signal_client.clone(), + engine_inner.clone(), + emitter.clone(), + ); } } } @@ -279,23 +367,23 @@ impl RTCEngine { }); } RTCEvent::ConnectionChange { state, target } => { - // Reconnect if we've been disconnected unexpectedly - trace!("Connection change, {:?} {:?}", state, target); + trace!("connection change, {:?} {:?}", state, target); let subscriber_primary = engine_inner.join_response.lock().subscriber_primary; let is_primary = subscriber_primary && target == SignalTarget::Subscriber; - if is_primary && state == PeerConnectionState::Disconnected { + if is_primary && state == PeerConnectionState::Connected { let old_state = engine_inner .pc_state .swap(PCState::Connected as u8, Ordering::SeqCst); if old_state == PCState::New as u8 { - // TODO(theomonnom) Handle disconnect + let _ = emitter.send(EngineEvent::Connected).await; // First time connected } } else if state == PeerConnectionState::Failed { engine_inner .pc_state .store(PCState::Disconnected as u8, Ordering::SeqCst); - // TODO(theomonnom) Handle disconnect + + Self::handle_disconnected(signal_client, engine_inner, emitter); } } RTCEvent::DataChannel { @@ -449,30 +537,131 @@ impl RTCEngine { Ok(()) } - async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> { - if !self.join_response().subscriber_primary { - return Ok(()); - } - - let publisher = &self.engine_inner.publisher_pc; + async fn handle_disconnected( + signal_client: Arc, + engine_inner: Arc, + emitter: EngineEmitter, + ) { + if engine_inner.closed.load(Ordering::SeqCst) + || engine_inner.reconnecting.load(Ordering::SeqCst) { - let mut publisher = publisher.lock().await; - if !publisher.is_connected() - && publisher.peer_connection().ice_connection_state() - != IceConnectionState::IceConnectionChecking - { - let _ = self.negotiate_publisher().await; + return; + } + + engine_inner.reconnecting.store(true, Ordering::SeqCst); + warn!("RTCEngine disconnected unexpectedly, reconnecting..."); + + let mut full_reconnect = false; + for i in 0..RECONNECT_ATTEMPTS { + if full_reconnect { + if i == 0 { + let _ = emitter.send(EngineEvent::Restarting).await; + } + + info!("restarting connection... attempt: {}", i); + if let Err(err) = Self::try_restart_connection( + signal_client.clone(), + engine_inner.clone(), + emitter.clone(), + ) + .await + { + error!("restarting connection failed: {}", err); + } else { + return; + } + } else { + if i == 0 { + let _ = emitter.send(EngineEvent::Resuming).await; + } + + info!("resuming connection... attempt: {}", i); + if let Err(err) = Self::try_resume_connection( + signal_client.clone(), + engine_inner.clone(), + emitter.clone(), + ) + .await + { + error!("resuming connection failed: {}", err); + if let EngineError::Signal(_) = err { + full_reconnect = true; + } + } else { + return; + } } + + tokio::time::sleep(RECONNECT_INTERVAL).await; + } + error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS); + engine_inner.reconnecting.store(false, Ordering::SeqCst); + + // TODO DISCONNECT + } + + async fn try_restart_connection( + signal_client: Arc, + engine_inner: Arc, + emitter: EngineEmitter, + ) -> EngineResult<()> { + Ok(()) + } + + async fn try_resume_connection( + signal_client: Arc, + engine_inner: Arc, + emitter: EngineEmitter, + ) -> EngineResult<()> { + let mut options = engine_inner.options.lock().clone(); + options.sid = engine_inner + .join_response + .lock() + .participant + .as_ref() + .unwrap() + .sid + .clone(); + + signal_client + .reconnect( + &engine_inner.url, + &engine_inner.token.lock().clone(), + options, + ) + .await?; + + let _ = emitter.send(EngineEvent::SignalResumed).await; + + engine_inner + .subscriber_pc + .lock() + .await + .prepare_ice_restart(); + + if engine_inner.has_published.load(Ordering::SeqCst) { + engine_inner + .publisher_pc + .lock() + .await + .create_and_send_offer(RTCOfferAnswerOptions { + ice_restart: true, + ..Default::default() + }) + .await?; } - let dc = self.data_channel(kind); - if dc.lock().state() == DataState::Open { - return Ok(()); - } + Self::wait_pc_connection(engine_inner).await?; + signal_client.flush_queue().await; - // Wait until the PeerConnection is connected + let _ = emitter.send(EngineEvent::Resumed); + + Ok(()) + } + + async fn wait_pc_connection(engine_inner: Arc) -> EngineResult<()> { let wait_connected = async move { - while publisher.lock().await.is_connected() && dc.lock().state() == DataState::Open { + while engine_inner.pc_state.load(Ordering::SeqCst) != PCState::Connected as u8 { sleep(Duration::from_millis(50)).await; } }; @@ -480,30 +669,14 @@ impl RTCEngine { tokio::select! { _ = wait_connected => Ok(()), _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => { - let err = EngineError::Connection("could not establish publisher connection: timeout".to_string()); - error!(error = ?err); + let err = EngineError::Connection("wait_pc_connection timed out".to_string()); Err(err) } } } - async fn negotiate_publisher(&self) -> EngineResult<()> { - self.engine_inner - .has_published - .store(true, Ordering::SeqCst); - if let Err(err) = self - .engine_inner - .publisher_pc - .lock() - .await - .negotiate() - .await - { - error!("failed to negotiate the publisher: {:?}", err); - Err(err)? - } else { - Ok(()) - } + fn close(&self) { + // TODO } fn configure_engine( @@ -617,16 +790,9 @@ impl RTCEngine { reliable_dc: Mutex::new(reliable_dc), sub_lossy_dc: Mutex::new(None), sub_reliable_dc: Mutex::new(None), + closed: AtomicBool::new(false), }, events, )) } - - fn data_channel(&self, kind: data_packet::Kind) -> &Mutex { - if kind == data_packet::Kind::Reliable { - &self.engine_inner.reliable_dc - } else { - &self.engine_inner.lossy_dc - } - } } diff --git a/crates/livekit-core/src/rtc_engine/pc_transport.rs b/crates/livekit-core/src/rtc_engine/pc_transport.rs index 8d9676a..19c44a9 100644 --- a/crates/livekit-core/src/rtc_engine/pc_transport.rs +++ b/crates/livekit-core/src/rtc_engine/pc_transport.rs @@ -19,12 +19,12 @@ pub type OnOfferHandler = Box< + Sync, >; -pub struct PCTransport { +pub(crate) struct PCTransport { peer_connection: PeerConnection, pending_candidates: Vec, on_offer_handler: Option, - restarting_ice: bool, renegotiate: bool, + restarting_ice: bool, } impl Debug for PCTransport { @@ -58,7 +58,11 @@ impl PCTransport { self.on_offer_handler = Some(handler); } - #[tracing::instrument] + pub fn prepare_ice_restart(&mut self) { + self.restarting_ice = true; + } + + #[tracing::instrument(level = Level::DEBUG)] pub async fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) -> Result<(), RTCError> { if self.peer_connection.remote_description().is_none() { self.pending_candidates.push(ice_candidate); @@ -71,7 +75,7 @@ impl PCTransport { Ok(()) } - #[tracing::instrument] + #[tracing::instrument(level = Level::DEBUG)] pub async fn set_remote_description( &mut self, remote_description: SessionDescription, @@ -94,15 +98,15 @@ impl PCTransport { Ok(()) } - #[tracing::instrument] + #[tracing::instrument(level = Level::DEBUG)] pub async fn negotiate(&mut self) -> Result<(), RTCError> { // TODO(theomonnom) Debounce here with NEGOTIATION_FREQUENCY self.create_and_send_offer(RTCOfferAnswerOptions::default()) .await } - #[tracing::instrument] - async fn create_and_send_offer( + #[tracing::instrument(level = Level::DEBUG)] + pub async fn create_and_send_offer( &mut self, options: RTCOfferAnswerOptions, ) -> Result<(), RTCError> { diff --git a/crates/livekit-core/src/signal_client/mod.rs b/crates/livekit-core/src/signal_client/mod.rs index 2958de4..4567a25 100644 --- a/crates/livekit-core/src/signal_client/mod.rs +++ b/crates/livekit-core/src/signal_client/mod.rs @@ -1,15 +1,18 @@ use std::fmt::Debug; +use std::sync::RwLockWriteGuard; use std::time::Duration; use livekit_webrtc::peer_connection_factory::{ ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration, }; +use parking_lot::RwLock; use thiserror::Error; use tokio::sync::mpsc; use tokio_tungstenite::tungstenite::Error as WsError; use crate::proto::{signal_request, signal_response, JoinResponse}; use crate::signal_client::signal_stream::SignalStream; +use tracing::{instrument, Level}; mod signal_stream; @@ -21,7 +24,7 @@ pub const JOIN_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); #[derive(Error, Debug)] pub enum SignalError { - #[error("websocket failure")] + #[error("ws failure: {0}")] WsError(#[from] WsError), #[error("failed to parse the url")] UrlParse(#[from] url::ParseError), @@ -39,12 +42,12 @@ pub(crate) enum SignalEvent { Close, } -#[derive(Debug)] +#[derive(Debug, Clone)] pub(crate) struct SignalOptions { - reconnect: bool, - auto_subscribe: bool, - sid: String, - adaptive_stream: bool, + pub(crate) reconnect: bool, + pub(crate) sid: String, + pub auto_subscribe: bool, + pub adaptive_stream: bool, } impl Default for SignalOptions { @@ -60,32 +63,59 @@ impl Default for SignalOptions { #[derive(Debug)] pub struct SignalClient { - stream: SignalStream, + stream: RwLock>, emitter: SignalEmitter, } impl SignalClient { + pub fn new() -> (Self, SignalEvents) { + let (emitter, events) = mpsc::channel(8); + ( + Self { + stream: Default::default(), + emitter, + }, + events, + ) + } + + #[instrument(level = Level::DEBUG, skip(url, token, options))] pub(crate) async fn connect( + &self, url: &str, token: &str, options: SignalOptions, - ) -> SignalResult<(Self, SignalEvents)> { - let (emitter, events) = mpsc::channel(8); - let stream = SignalStream::connect(url, token, options, emitter.clone()).await?; - - // TODO(theomonnom) Retry initial connection - - Ok((Self { stream, emitter }, events)) + ) -> SignalResult<()> { + let stream = SignalStream::connect(url, token, options, self.emitter.clone()).await?; + *self.stream.write() = Some(stream); + Ok(()) } - pub async fn send(&self, signal: signal_request::Message) { - if let Err(_) = self.stream.send(signal).await { - // TODO(theomonnom) Queue message ( Ignore on full reconnect ) + #[instrument(level = Level::DEBUG)] + pub async fn close(&self) { + if let Some(stream) = self.stream.write().take() { + stream.close().await; } } - pub async fn reconnect(&self) { - // TODO(theomonnom) Close & recreate SignalStream, also send the queue if needed + #[instrument(level = Level::DEBUG)] + pub async fn send(&self, signal: signal_request::Message) { + if let Some(stream) = self.stream.read().as_ref() { + if stream.send(signal).await.is_ok() { + return; + } + } + + // TODO(theomonnom): enqueue message + } + + pub async fn clear_queue(&self) { + // TODO(theomonnom): impl + } + + #[instrument(level = Level::DEBUG)] + pub async fn flush_queue(&self) { + // TODO(theomonnom): impl } } @@ -115,8 +145,9 @@ pub mod utils { use tokio::sync::mpsc; use tokio::time::timeout; use tokio_tungstenite::tungstenite::Error as WsError; - use tracing::{event, Level}; + use tracing::{event, instrument, Level}; + #[instrument(level = Level::DEBUG, skip(receiver))] pub(crate) async fn next_join_response( receiver: &mut mpsc::Receiver, ) -> SignalResult { diff --git a/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp b/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp index 0afc9d9..a4cdfde 100644 --- a/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp +++ b/crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp @@ -43,4 +43,4 @@ rtc::Thread* RTCRuntime::signaling_thread() const { std::shared_ptr create_rtc_runtime() { return std::make_shared(); } -} // namespace livekit \ No newline at end of file +} // namespace livekit diff --git a/crates/livekit-webrtc/src/media_stream.rs b/crates/livekit-webrtc/src/media_stream.rs index 565c92b..0ad7a7f 100644 --- a/crates/livekit-webrtc/src/media_stream.rs +++ b/crates/livekit-webrtc/src/media_stream.rs @@ -1,6 +1,7 @@ use cxx::UniquePtr; use libwebrtc_sys::media_stream as sys_ms; use libwebrtc_sys::MEDIA_TYPE_VIDEO; +use livekit_utils::enum_dispatch; use std::fmt::{Debug, Formatter}; use std::pin::Pin; use std::sync::{Arc, Mutex}; @@ -25,17 +26,6 @@ pub enum MediaStreamTrackHandle { Video(Arc), } -macro_rules! shared_getter { - ($x:ident, $ret:ty) => { - fn $x(&self) -> $ret { - match self { - Self::Video(inner) => inner.$x(), - Self::Audio(inner) => inner.$x(), - } - } - }; -} - impl MediaStreamTrackHandle { pub(crate) fn new(cxx_handle: UniquePtr) -> Self { unsafe { @@ -66,17 +56,14 @@ impl Debug for MediaStreamTrackHandle { } impl MediaStreamTrackTrait for MediaStreamTrackHandle { - shared_getter!(kind, String); - shared_getter!(id, String); - shared_getter!(enabled, bool); - shared_getter!(state, TrackState); - - fn set_enabled(&self, enabled: bool) -> bool { - match self { - Self::Video(inner) => inner.set_enabled(enabled), - Self::Audio(inner) => inner.set_enabled(enabled), - } - } + enum_dispatch!( + [Audio, Video] + fnc!(kind, &Self, [], String); + fnc!(id, &Self, [], String); + fnc!(enabled, &Self, [], bool); + fnc!(state, &Self, [], TrackState); + fnc!(set_enabled, &Self, [enabled: bool], bool); + ); } pub struct AudioTrack { diff --git a/examples/Cargo.lock b/examples/Cargo.lock index ce8a9ce..5553bfc 100644 --- a/examples/Cargo.lock +++ b/examples/Cargo.lock @@ -48,9 +48,9 @@ dependencies = [ [[package]] name = "aho-corasick" -version = "0.7.19" +version = "0.7.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b4f55bd91a0978cbfd91c457a164bab8b4001c833b7f323132c0a4e1922dd44e" +checksum = "cc936419f96fa211c1b9166887b38e5e40b19958e5b895be7c1f93adec7071ac" dependencies = [ "memchr", ] @@ -64,20 +64,11 @@ dependencies = [ "libc", ] -[[package]] -name = "ansi_term" -version = "0.12.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d52a9bb7ec0cf484c551830a7ce27bd20d67eac647e1befb56b0be4ee39a55d2" -dependencies = [ - "winapi", -] - [[package]] name = "anyhow" -version = "1.0.65" +version = "1.0.66" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "98161a4e3e2184da77bb14f02184cdd111e83bbbcc9979dfee3c44b9a85f5602" +checksum = "216261ddc8289130e551ddcd5ce8a064710c0d064a4d2895c67151c92b5443f6" [[package]] name = "arboard" @@ -138,9 +129,9 @@ checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa" [[package]] name = "base64" -version = "0.13.0" +version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "904dfeac50f3cdaba28fc6f57fdcddb75f49ed61346676a78c4ffe55877802fd" +checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8" [[package]] name = "bit-set" @@ -212,18 +203,18 @@ checksum = "14c189c53d098945499cdfa7ecc63567cf3886b3332b312a5b4585d8d3a6a610" [[package]] name = "bytes" -version = "1.2.1" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ec8a7b6a70fde80372154c65702f00a0f56f3e1c36abbc6c440484be248856db" +checksum = "dfb24e866b15a1af2a1b663f10c6b6b8f397a84aadb828f12e5b289ec23a3a3c" [[package]] name = "calloop" -version = "0.10.3" +version = "0.10.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5bcf530afb40e45e14440701e5e996d7fd139e84a912a4d83a8d6a0fb3e58663" +checksum = "19457a0da465234abd76134a5c2a910c14bd3c5558463e4396ab9a37a328e465" dependencies = [ "log", - "nix 0.25.0", + "nix 0.25.1", "slotmap", "thiserror", "vec_map", @@ -231,9 +222,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.0.73" +version = "1.0.77" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2fff2a6927b3bb87f9595d67196a70493f627687a71d87a0d692242c33f58c11" +checksum = "e9f73505338f7d905b19d18738976aae232eb46b8efc15554ffc56deb5d9ebe4" [[package]] name = "cesu8" @@ -436,9 +427,9 @@ checksum = "b365fabc795046672053e29c954733ec3b05e4be654ab130fe8f1f94d7051f35" [[package]] name = "cxx" -version = "1.0.78" +version = "1.0.83" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19f39818dcfc97d45b03953c1292efc4e80954e1583c4aa770bac1383e2310a4" +checksum = "bdf07d07d6531bfcdbe9b8b739b104610c6508dcc4d63b410585faf338241daf" dependencies = [ "cc", "cxxbridge-flags", @@ -448,9 +439,9 @@ dependencies = [ [[package]] name = "cxx-build" -version = "1.0.78" +version = "1.0.83" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3e580d70777c116df50c390d1211993f62d40302881e54d4b79727acb83d0199" +checksum = "d2eb5b96ecdc99f72657332953d4d9c50135af1bac34277801cc3937906ebd39" dependencies = [ "cc", "codespan-reporting", @@ -463,15 +454,15 @@ dependencies = [ [[package]] name = "cxxbridge-flags" -version = "1.0.78" +version = "1.0.83" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "56a46460b88d1cec95112c8c363f0e2c39afdb237f60583b0b36343bf627ea9c" +checksum = "ac040a39517fd1674e0f32177648334b0f4074625b5588a64519804ba0553b12" [[package]] name = "cxxbridge-macro" -version = "1.0.78" +version = "1.0.83" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "747b608fecf06b0d72d440f27acc99288207324b793be2c17991839f3d4995ea" +checksum = "1362b0ddcfc4eb0a1f57b68bd77dd99f0e826958a96abd0ae9bd092e114ffed6" dependencies = [ "proc-macro2", "quote", @@ -526,9 +517,9 @@ dependencies = [ [[package]] name = "digest" -version = "0.10.5" +version = "0.10.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "adfbc57365a37acbd2ebf2b64d7e69bb766e2fea813521ed536f5d0520dcf86c" +checksum = "8168378f4e5023e7218c89c891c0fd8ecdb5e5e4f18cb78f38cf245dd021e76f" dependencies = [ "block-buffer", "crypto-common", @@ -569,10 +560,18 @@ dependencies = [ "wio", ] +[[package]] +name = "ecolor" +version = "0.20.0" +source = "git+https://github.com/emilk/egui#e7471f1191081ecb514fe129bb8618938b67e796" +dependencies = [ + "bytemuck", +] + [[package]] name = "egui" -version = "0.19.0" -source = "git+https://github.com/emilk/egui#0336816faf9f361e59ca77f99c19fcfa2bf7c993" +version = "0.20.0" +source = "git+https://github.com/emilk/egui#e7471f1191081ecb514fe129bb8618938b67e796" dependencies = [ "ahash 0.8.2", "epaint", @@ -582,8 +581,8 @@ dependencies = [ [[package]] name = "egui-wgpu" -version = "0.19.0" -source = "git+https://github.com/emilk/egui#0336816faf9f361e59ca77f99c19fcfa2bf7c993" +version = "0.20.0" +source = "git+https://github.com/emilk/egui#e7471f1191081ecb514fe129bb8618938b67e796" dependencies = [ "bytemuck", "egui", @@ -596,8 +595,8 @@ dependencies = [ [[package]] name = "egui-winit" -version = "0.19.0" -source = "git+https://github.com/emilk/egui#0336816faf9f361e59ca77f99c19fcfa2bf7c993" +version = "0.20.0" +source = "git+https://github.com/emilk/egui#e7471f1191081ecb514fe129bb8618938b67e796" dependencies = [ "arboard", "egui", @@ -608,26 +607,6 @@ dependencies = [ "winit", ] -[[package]] -name = "egui_demo_lib" -version = "0.19.0" -source = "git+https://github.com/emilk/egui#0336816faf9f361e59ca77f99c19fcfa2bf7c993" -dependencies = [ - "egui", - "egui_extras", - "enum-map", - "tracing", - "unicode_names2", -] - -[[package]] -name = "egui_extras" -version = "0.19.0" -source = "git+https://github.com/emilk/egui#0336816faf9f361e59ca77f99c19fcfa2bf7c993" -dependencies = [ - "egui", -] - [[package]] name = "either" version = "1.8.0" @@ -636,42 +615,22 @@ checksum = "90e5c1c8368803113bf0c9584fc495a58b86dc8a29edbf8fe877d21d9507e797" [[package]] name = "emath" -version = "0.19.0" -source = "git+https://github.com/emilk/egui#0336816faf9f361e59ca77f99c19fcfa2bf7c993" +version = "0.20.0" +source = "git+https://github.com/emilk/egui#e7471f1191081ecb514fe129bb8618938b67e796" dependencies = [ "bytemuck", ] -[[package]] -name = "enum-map" -version = "2.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f5a56d54c8dd9b3ad34752ed197a4eb2a6601bc010808eb097a04a58ae4c43e1" -dependencies = [ - "enum-map-derive", - "serde", -] - -[[package]] -name = "enum-map-derive" -version = "0.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9045e2676cd5af83c3b167d917b0a5c90a4d8e266e2683d6631b235c457fc27" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "epaint" -version = "0.19.0" -source = "git+https://github.com/emilk/egui#0336816faf9f361e59ca77f99c19fcfa2bf7c993" +version = "0.20.0" +source = "git+https://github.com/emilk/egui#e7471f1191081ecb514fe129bb8618938b67e796" dependencies = [ "ab_glyph", "ahash 0.8.2", "atomic_refcell", "bytemuck", + "ecolor", "emath", "nohash-hasher", "parking_lot", @@ -803,9 +762,9 @@ dependencies = [ [[package]] name = "futures" -version = "0.3.24" +version = "0.3.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f21eda599937fba36daeb58a22e8f5cee2d14c4a17b5b7739c7c8e5e3b8230c" +checksum = "38390104763dc37a5145a53c29c63c1290b5d316d6086ec32c293f6736051bb0" dependencies = [ "futures-channel", "futures-core", @@ -818,9 +777,9 @@ dependencies = [ [[package]] name = "futures-channel" -version = "0.3.24" +version = "0.3.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30bdd20c28fadd505d0fd6712cdfcb0d4b5648baf45faef7f852afb2399bb050" +checksum = "52ba265a92256105f45b719605a571ffe2d1f0fea3807304b522c1d778f79eed" dependencies = [ "futures-core", "futures-sink", @@ -828,15 +787,15 @@ dependencies = [ [[package]] name = "futures-core" -version = "0.3.24" +version = "0.3.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e5aa3de05362c3fb88de6531e6296e85cde7739cccad4b9dfeeb7f6ebce56bf" +checksum = "04909a7a7e4633ae6c4a9ab280aeb86da1236243a77b694a49eacd659a4bd3ac" [[package]] name = "futures-executor" -version = "0.3.24" +version = "0.3.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ff63c23854bee61b6e9cd331d523909f238fc7636290b96826e9cfa5faa00ab" +checksum = "7acc85df6714c176ab5edf386123fafe217be88c0840ec11f199441134a074e2" dependencies = [ "futures-core", "futures-task", @@ -845,15 +804,15 @@ dependencies = [ [[package]] name = "futures-io" -version = "0.3.24" +version = "0.3.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbf4d2a7a308fd4578637c0b17c7e1c7ba127b8f6ba00b29f717e9655d85eb68" +checksum = "00f5fb52a06bdcadeb54e8d3671f8888a39697dcb0b81b23b55174030427f4eb" [[package]] name = "futures-macro" -version = "0.3.24" +version = "0.3.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42cd15d1c7456c04dbdf7e88bcd69760d74f3a798d6444e16974b505b0e62f17" +checksum = "bdfb8ce053d86b91919aad980c220b1fb8401a9394410e1c289ed7e66b61835d" dependencies = [ "proc-macro2", "quote", @@ -862,21 +821,21 @@ dependencies = [ [[package]] name = "futures-sink" -version = "0.3.24" +version = "0.3.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "21b20ba5a92e727ba30e72834706623d94ac93a725410b6a6b6fbc1b07f7ba56" +checksum = "39c15cf1a4aa79df40f1bb462fb39676d0ad9e366c2a33b590d7c66f4f81fcf9" [[package]] name = "futures-task" -version = "0.3.24" +version = "0.3.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6508c467c73851293f390476d4491cf4d227dbabcd4170f3bb6044959b294f1" +checksum = "2ffb393ac5d9a6eaa9d3fdf37ae2776656b706e200c8e16b1bdb227f5198e6ea" [[package]] name = "futures-util" -version = "0.3.24" +version = "0.3.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "44fb6cb1be61cc1d2e43b262516aafcf63b241cffdb1d3fa115f91d9c7b09c90" +checksum = "197676987abd2f9cadff84926f410af1c183608d36641465df73ae8211dc65d6" dependencies = [ "futures-channel", "futures-core", @@ -921,9 +880,9 @@ dependencies = [ [[package]] name = "getrandom" -version = "0.2.7" +version = "0.2.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4eb1a864a501629691edf6c15a593b7a51eebaa1e8468e9ddc623de7c9b58ec6" +checksum = "c05aeb6a22b8f62540c194aac980f2115af067bfe15a0734d7277a768d396b31" dependencies = [ "cfg-if", "libc", @@ -1052,9 +1011,9 @@ dependencies = [ [[package]] name = "indexmap" -version = "1.9.1" +version = "1.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "10a35a97730320ffe8e2d410b5d3b69279b98d2c14bdb8b70ea89ecf7888d41e" +checksum = "1885e79c1fc4b10f0e172c475f458b7f7b93061064d98c3293e98c5ba0c8b399" dependencies = [ "autocfg", "hashbrown", @@ -1083,9 +1042,9 @@ dependencies = [ [[package]] name = "itoa" -version = "1.0.3" +version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6c8af84674fe1f223a982c933a0ee1086ac4d4052aa0fb8060c12c6ad838e754" +checksum = "4217ad341ebadf8d8e724e264f13e593e0648f5b3e94b3896a5df283be015ecc" [[package]] name = "jni" @@ -1135,9 +1094,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" [[package]] name = "libc" -version = "0.2.134" +version = "0.2.138" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "329c933548736bc49fd575ee68c89e8be4d260064184389a5b77517cddd99ffb" +checksum = "db6d7e329c562c5dfab7a46a2afabc8b987ab9a4834c9d1ca04dc54c1546cef8" [[package]] name = "libloading" @@ -1298,14 +1257,14 @@ dependencies = [ [[package]] name = "mio" -version = "0.8.4" +version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "57ee1c23c7c63b0c9250c339ffdc69255f110b298b901b9f6c82547b7b87caaf" +checksum = "e5d732bc30207a6423068df043e3d02e0735b155ad7ce1a6f76fe2baa5b158de" dependencies = [ "libc", "log", "wasi", - "windows-sys", + "windows-sys 0.42.0", ] [[package]] @@ -1336,9 +1295,9 @@ dependencies = [ [[package]] name = "native-tls" -version = "0.2.10" +version = "0.2.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fd7e2f3618557f980e0b17e8856252eee3c97fa12c54dff0ca290fb6266ca4a9" +checksum = "07226173c32f2926027b63cce4bcd8076c3552846cbe7925f3aaffeac0a3b92e" dependencies = [ "lazy_static", "libc", @@ -1412,9 +1371,9 @@ dependencies = [ [[package]] name = "nix" -version = "0.24.2" +version = "0.24.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "195cdbc1741b8134346d515b3a56a1c94b0912758009cfd53f99ea0f57b065fc" +checksum = "fa52e972a9a719cecb6864fb88568781eb706bac2cd1d4f04a648542dbf78069" dependencies = [ "bitflags", "cfg-if", @@ -1424,9 +1383,9 @@ dependencies = [ [[package]] name = "nix" -version = "0.25.0" +version = "0.25.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e322c04a9e3440c327fca7b6c8a63e6890a32fa2ad689db972425f07e0d22abb" +checksum = "f346ff70e7dbfd675fe90590b92d59ef2de15a8779ae305ebcbfd3f0caf59be4" dependencies = [ "autocfg", "bitflags", @@ -1451,6 +1410,16 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "nu-ansi-term" +version = "0.46.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84" +dependencies = [ + "overload", + "winapi", +] + [[package]] name = "num-traits" version = "0.2.15" @@ -1462,9 +1431,9 @@ dependencies = [ [[package]] name = "num_cpus" -version = "1.13.1" +version = "1.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19e64526ebdee182341572e50e9ad03965aa510cd94427a4549448f285e957a1" +checksum = "f6058e64324c71e02bc2b150e4f3bc8286db6c83092132ffa3f6b1eab0f9def5" dependencies = [ "hermit-abi", "libc", @@ -1532,15 +1501,15 @@ dependencies = [ [[package]] name = "once_cell" -version = "1.15.0" +version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e82dad04139b71a90c080c8463fe0dc7902db5192d939bd0950f074d014339e1" +checksum = "86f0b0d4bf799edbc74508c1e8bf170ff5f41238e5f8225603ca7caaae2b7860" [[package]] name = "openssl" -version = "0.10.42" +version = "0.10.44" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "12fc0523e3bd51a692c8850d075d74dc062ccf251c0110668cbd921917118a13" +checksum = "29d971fd5722fec23977260f6e81aa67d2f22cadbdc2aa049f1022d9a3be1566" dependencies = [ "bitflags", "cfg-if", @@ -1570,9 +1539,9 @@ checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" [[package]] name = "openssl-sys" -version = "0.9.76" +version = "0.9.79" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5230151e44c0f05157effb743e8d517472843121cf9243e8b81393edb5acd9ce" +checksum = "5454462c0eced1e97f2ec09036abc8da362e66802f66fd20f86854d9d8cbcbc4" dependencies = [ "autocfg", "cc", @@ -1581,6 +1550,12 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "overload" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39" + [[package]] name = "owned_ttf_parser" version = "0.17.1" @@ -1602,15 +1577,15 @@ dependencies = [ [[package]] name = "parking_lot_core" -version = "0.9.3" +version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09a279cbf25cb0757810394fbc1e359949b59e348145c643a939a525692e6929" +checksum = "7ff9f3fef3968a3ec5945535ed654cb38ff72d7495a25619e2247fb15a2ed9ba" dependencies = [ "cfg-if", "libc", "redox_syscall", "smallvec", - "windows-sys", + "windows-sys 0.42.0", ] [[package]] @@ -1643,9 +1618,9 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" [[package]] name = "pkg-config" -version = "0.3.25" +version = "0.3.26" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1df8c4ec4b0627e53bdf214615ad287367e482558cf84b109250b37464dc03ae" +checksum = "6ac9a59f73473f1b8d852421e59e64809f025994837ef743615c6d0c5b305160" [[package]] name = "png" @@ -1667,9 +1642,19 @@ checksum = "5da3b0203fd7ee5720aa0b5e790b591aa5d3f41c3ed2c34a3a393382198af2f7" [[package]] name = "ppv-lite86" -version = "0.2.16" +version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb9f9e6e233e5c4a35559a617bf40a4ec447db2e84c20b55a6f83167b7e57872" +checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de" + +[[package]] +name = "prettyplease" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c142c0e46b57171fe0c528bee8c5b7569e80f0c17e377cd0e30ea57dbc11bb51" +dependencies = [ + "proc-macro2", + "syn", +] [[package]] name = "proc-macro-crate" @@ -1684,9 +1669,9 @@ dependencies = [ [[package]] name = "proc-macro2" -version = "1.0.46" +version = "1.0.47" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94e2ef8dbfc347b10c094890f778ee2e36ca9bb4262e86dc99cd217e35f3470b" +checksum = "5ea3d908b0e36316caf9e9e2c4625cdde190a7e6f440d794667ed17a1855e725" dependencies = [ "unicode-ident", ] @@ -1699,9 +1684,9 @@ checksum = "74605f360ce573babfe43964cbe520294dcb081afbf8c108fc6e23036b4da2df" [[package]] name = "prost" -version = "0.11.0" +version = "0.11.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "399c3c31cdec40583bb68f0b18403400d01ec4289c383aa047560439952c4dd7" +checksum = "c0b18e655c21ff5ac2084a5ad0611e827b3f92badf79f4910b5a5c58f4d87ff0" dependencies = [ "bytes", "prost-derive", @@ -1709,9 +1694,9 @@ dependencies = [ [[package]] name = "prost-build" -version = "0.11.1" +version = "0.11.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f835c582e6bd972ba8347313300219fed5bfa52caf175298d860b61ff6069bb" +checksum = "276470f7f281b0ed53d2ae42dd52b4a8d08853a3c70e7fe95882acbb98a6ae94" dependencies = [ "bytes", "heck", @@ -1720,18 +1705,20 @@ dependencies = [ "log", "multimap", "petgraph", + "prettyplease", "prost", "prost-types", "regex", + "syn", "tempfile", "which", ] [[package]] name = "prost-derive" -version = "0.11.0" +version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7345d5f0e08c0536d7ac7229952590239e77abf0a0100a1b1d890add6ea96364" +checksum = "164ae68b6587001ca506d3bf7f1000bfa248d0e1217b618108fba4ec1d0cc306" dependencies = [ "anyhow", "itertools", @@ -1742,9 +1729,9 @@ dependencies = [ [[package]] name = "prost-types" -version = "0.11.1" +version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4dfaa718ad76a44b3415e6c4d53b17c8f99160dcb3a99b10470fce8ad43f6e3e" +checksum = "747761bc3dc48f9a34553bf65605cf6cb6288ba219f3450b4275dbd81539551a" dependencies = [ "bytes", "prost", @@ -1824,9 +1811,9 @@ dependencies = [ [[package]] name = "regex" -version = "1.6.0" +version = "1.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c4eb3267174b8c6c2f654116623910a0fef09c4753f8dd83db29c48a0df988b" +checksum = "e076559ef8e241f2ae3479e36f97bd5741c0330689e217ad51ce2c76808b868a" dependencies = [ "aho-corasick", "memchr", @@ -1835,9 +1822,9 @@ dependencies = [ [[package]] name = "regex-syntax" -version = "0.6.27" +version = "0.6.28" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a3f87b73ce11b1619a3c6332f45341e0047173771e8b8b73f87bfeefb7b56244" +checksum = "456c603be3e8d448b072f410900c09faf164fbce2d480456f50eea6e25f9c848" [[package]] name = "remove_dir_all" @@ -1891,7 +1878,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "88d6731146462ea25d9244b2ed5fd1d716d25c52e4d54aa4fb0f3c4e9854dbe2" dependencies = [ "lazy_static", - "windows-sys", + "windows-sys 0.36.1", ] [[package]] @@ -1949,18 +1936,18 @@ dependencies = [ [[package]] name = "serde" -version = "1.0.145" +version = "1.0.149" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "728eb6351430bccb993660dfffc5a72f91ccc1295abaa8ce19b27ebe4f75568b" +checksum = "256b9932320c590e707b94576e3cc1f7c9024d0ee6612dfbcf1cb106cbe8e055" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.145" +version = "1.0.149" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "81fa1584d3d1bcacd84c277a0dfe21f5b0f6accf4a23d04d4c6d61f1af522b4c" +checksum = "b4eae9b04cbffdfd550eb462ed33bc6a1b68c935127d008b27444d08380f94e4" dependencies = [ "proc-macro2", "quote", @@ -1969,9 +1956,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.85" +version = "1.0.89" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e55a28e3aaef9d5ce0506d0a14dbba8054ddc7e499ef522dd8b26859ec9d4a44" +checksum = "020ff22c755c2ed3f8cf162dbb41a7268d934702f3ed3631656ea597e08fc3db" dependencies = [ "itoa", "ryu", @@ -2001,9 +1988,9 @@ dependencies = [ [[package]] name = "sha-1" -version = "0.10.0" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "028f48d513f9678cda28f6e4064755b3fbb2af6acd672f2c209b62323f7aea0f" +checksum = "f5058ada175748e33390e40e872bd0fe59a19f265d0158daa551c5a88a76009c" dependencies = [ "cfg-if", "cpufeatures", @@ -2035,7 +2022,6 @@ dependencies = [ "egui", "egui-wgpu", "egui-winit", - "egui_demo_lib", "futures", "livekit", "parking_lot", @@ -2066,9 +2052,9 @@ dependencies = [ [[package]] name = "smallvec" -version = "1.9.0" +version = "1.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2fd0db749597d91ff862fd1d55ea87f7855a744a8425a64695b6fca237d1dad1" +checksum = "a507befe795404456341dfab10cef66ead4c041f62b8b11bbb92bffe5d0953e0" [[package]] name = "smithay-client-toolkit" @@ -2082,7 +2068,7 @@ dependencies = [ "lazy_static", "log", "memmap2", - "nix 0.24.2", + "nix 0.24.3", "pkg-config", "wayland-client", "wayland-cursor", @@ -2139,9 +2125,9 @@ checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623" [[package]] name = "syn" -version = "1.0.101" +version = "1.0.105" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e90cde112c4b9690b8cbe810cba9ddd8bc1d7472e2cae317b69e9438c1cba7d2" +checksum = "60b9b43d45702de4c839cb9b51d9f529c5dd26a4aff255b42b1ebc03e88ee908" dependencies = [ "proc-macro2", "quote", @@ -2242,9 +2228,9 @@ checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c" [[package]] name = "tokio" -version = "1.21.2" +version = "1.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9e03c497dc955702ba729190dc4aac6f2a0ce97f913e5b1b5912fc5039d9099" +checksum = "eab6d665857cc6ca78d6e80303a02cea7a7851e85dfbd77cbdc09bd129f1ef46" dependencies = [ "autocfg", "bytes", @@ -2257,14 +2243,14 @@ dependencies = [ "signal-hook-registry", "socket2", "tokio-macros", - "winapi", + "windows-sys 0.42.0", ] [[package]] name = "tokio-macros" -version = "1.8.0" +version = "1.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9724f9a975fb987ef7a3cd9be0350edcbe130698af5b8f7a631e23d42d052484" +checksum = "d266c00fde287f55d3f1c3e96c500c362a2b8c695076ec180f27918820bc6df8" dependencies = [ "proc-macro2", "quote", @@ -2306,9 +2292,9 @@ dependencies = [ [[package]] name = "tracing" -version = "0.1.36" +version = "0.1.37" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2fce9567bd60a67d08a16488756721ba392f24f29006402881e43b19aac64307" +checksum = "8ce8c33a8d48bd45d624a6e523445fd21ec13d3653cd51f681abf67418f54eb8" dependencies = [ "cfg-if", "pin-project-lite", @@ -2318,9 +2304,9 @@ dependencies = [ [[package]] name = "tracing-attributes" -version = "0.1.22" +version = "0.1.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11c75893af559bc8e10716548bdef5cb2b983f8e637db9d0e15126b61b484ee2" +checksum = "4017f8f45139870ca7e672686113917c71c7a6e02d4924eda67186083c03081a" dependencies = [ "proc-macro2", "quote", @@ -2329,9 +2315,9 @@ dependencies = [ [[package]] name = "tracing-core" -version = "0.1.29" +version = "0.1.30" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5aeea4303076558a00714b823f9ad67d58a3bbda1df83d8827d21193156e22f7" +checksum = "24eb03ba0eab1fd845050058ce5e616558e8f8d8fca633e6b163fe25c797213a" dependencies = [ "once_cell", "valuable", @@ -2350,11 +2336,11 @@ dependencies = [ [[package]] name = "tracing-subscriber" -version = "0.3.15" +version = "0.3.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "60db860322da191b40952ad9affe65ea23e7dd6a5c442c2c42865810c6ab8e6b" +checksum = "a6176eae26dd70d0c919749377897b54a9276bd7061339665dd68777926b5a70" dependencies = [ - "ansi_term", + "nu-ansi-term", "sharded-slab", "smallvec", "thread_local", @@ -2399,9 +2385,9 @@ dependencies = [ [[package]] name = "typenum" -version = "1.15.0" +version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dcf81ac59edc17cc8697ff311e8f5ef2d99fcbd9817b34cec66f90b6c3dfd987" +checksum = "497961ef93d974e23eb6f433eb5fe1b7930b659f06d12dec6fc44a8f554c0bba" [[package]] name = "unicode-bidi" @@ -2411,9 +2397,9 @@ checksum = "099b7128301d285f79ddd55b9a83d5e6b9e97c92e0ea0daebee7263e932de992" [[package]] name = "unicode-ident" -version = "1.0.4" +version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dcc811dc4066ac62f84f11307873c4850cb653bfa9b1719cee2bd2204a4bc5dd" +checksum = "6ceab39d59e4c9499d4e5a8ee0e2735b891bb7308ac83dfb4e80cad195c9f6f3" [[package]] name = "unicode-normalization" @@ -2436,12 +2422,6 @@ version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f962df74c8c05a667b5ee8bcf162993134c104e96440b663c8daa176dc772d8c" -[[package]] -name = "unicode_names2" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "029df4cc8238cefc911704ff8fa210853a0f3bce2694d8f51181dd41ee0f3301" - [[package]] name = "url" version = "2.3.1" @@ -2575,7 +2555,7 @@ dependencies = [ "bitflags", "downcast-rs", "libc", - "nix 0.24.2", + "nix 0.24.3", "scoped-tls", "wayland-commons", "wayland-scanner", @@ -2588,7 +2568,7 @@ version = "0.29.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8691f134d584a33a6606d9d717b95c4fa20065605f798a3f350d78dced02a902" dependencies = [ - "nix 0.24.2", + "nix 0.24.3", "once_cell", "smallvec", "wayland-sys", @@ -2600,7 +2580,7 @@ version = "0.29.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6865c6b66f13d6257bef1cd40cbfe8ef2f150fb8ebbdb1e8e873455931377661" dependencies = [ - "nix 0.24.2", + "nix 0.24.3", "wayland-client", "xcursor", ] @@ -2667,9 +2647,9 @@ dependencies = [ [[package]] name = "wgpu" -version = "0.14.0" +version = "0.14.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c2272b17bffc8a0c7d53897435da7c1db587c87d3a14e8dae9cdb8d1d210fc0f" +checksum = "81f643110d228fd62a60c5ed2ab56c4d5b3704520bd50561174ec4ec74932937" dependencies = [ "arrayvec 0.7.2", "js-sys", @@ -2689,9 +2669,9 @@ dependencies = [ [[package]] name = "wgpu-core" -version = "0.14.0" +version = "0.14.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "73d14cad393054caf992ee02b7da6a372245d39a484f7461c1f44f6f6359bd28" +checksum = "6000d1284ef8eec6076fd5544a73125fd7eb9b635f18dceeb829d826f41724ca" dependencies = [ "arrayvec 0.7.2", "bit-vec", @@ -2822,43 +2802,100 @@ version = "0.36.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ea04155a16a59f9eab786fe12a4a450e75cdb175f9e0d80da1e17db09f55b8d2" dependencies = [ - "windows_aarch64_msvc", - "windows_i686_gnu", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_msvc", + "windows_aarch64_msvc 0.36.1", + "windows_i686_gnu 0.36.1", + "windows_i686_msvc 0.36.1", + "windows_x86_64_gnu 0.36.1", + "windows_x86_64_msvc 0.36.1", ] +[[package]] +name = "windows-sys" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a3e1820f08b8513f676f7ab6c1f99ff312fb97b553d30ff4dd86f9f15728aa7" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc 0.42.0", + "windows_i686_gnu 0.42.0", + "windows_i686_msvc 0.42.0", + "windows_x86_64_gnu 0.42.0", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc 0.42.0", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d2aa71f6f0cbe00ae5167d90ef3cfe66527d6f613ca78ac8024c3ccab9a19e" + [[package]] name = "windows_aarch64_msvc" version = "0.36.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9bb8c3fd39ade2d67e9874ac4f3db21f0d710bee00fe7cab16949ec184eeaa47" +[[package]] +name = "windows_aarch64_msvc" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd0f252f5a35cac83d6311b2e795981f5ee6e67eb1f9a7f64eb4500fbc4dcdb4" + [[package]] name = "windows_i686_gnu" version = "0.36.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "180e6ccf01daf4c426b846dfc66db1fc518f074baa793aa7d9b9aaeffad6a3b6" +[[package]] +name = "windows_i686_gnu" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fbeae19f6716841636c28d695375df17562ca208b2b7d0dc47635a50ae6c5de7" + [[package]] name = "windows_i686_msvc" version = "0.36.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e2e7917148b2812d1eeafaeb22a97e4813dfa60a3f8f78ebe204bcc88f12f024" +[[package]] +name = "windows_i686_msvc" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84c12f65daa39dd2babe6e442988fc329d6243fdce47d7d2d155b8d874862246" + [[package]] name = "windows_x86_64_gnu" version = "0.36.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4dcd171b8776c41b97521e5da127a2d86ad280114807d0b2ab1e462bc764d9e1" +[[package]] +name = "windows_x86_64_gnu" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf7b1b21b5362cbc318f686150e5bcea75ecedc74dd157d874d754a2ca44b0ed" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09d525d2ba30eeb3297665bd434a54297e4170c7f1a44cad4ef58095b4cd2028" + [[package]] name = "windows_x86_64_msvc" version = "0.36.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c811ca4a8c853ef420abd8592ba53ddbbac90410fab6903b3e79972a631f7680" +[[package]] +name = "windows_x86_64_msvc" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40009d85759725a34da6d89a94e63d7bdc50a862acf0dbc7c8e488f1edcb6f5" + [[package]] name = "winit" version = "0.27.5" @@ -2888,7 +2925,7 @@ dependencies = [ "wayland-client", "wayland-protocols", "web-sys", - "windows-sys", + "windows-sys 0.36.1", "x11-dl", ] @@ -2919,7 +2956,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "592b4883219f345e712b3209c62654ebda0bb50887f330cbd018d0f654bfd507" dependencies = [ "gethostname", - "nix 0.24.2", + "nix 0.24.3", "winapi", "winapi-wsapoll", "x11rb-protocol", @@ -2931,7 +2968,7 @@ version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "56b245751c0ac9db0e006dc812031482784e434630205a93c73cfefcaabeac67" dependencies = [ - "nix 0.24.2", + "nix 0.24.3", ] [[package]] diff --git a/examples/simple_room/Cargo.toml b/examples/simple_room/Cargo.toml index a5438f2..6d59c39 100644 --- a/examples/simple_room/Cargo.toml +++ b/examples/simple_room/Cargo.toml @@ -16,4 +16,3 @@ parking_lot = "0.12.1" 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_demo_lib = { git = "https://github.com/emilk/egui" } diff --git a/examples/simple_room/src/app.rs b/examples/simple_room/src/app.rs index 5ba5395..0855249 100644 --- a/examples/simple_room/src/app.rs +++ b/examples/simple_room/src/app.rs @@ -1,7 +1,8 @@ -use crate::events::DemoEvent; -use crate::video_grid::VideoGrid; +use crate::events::UiCmd; use crate::video_renderer::VideoRenderer; +use crate::{events::AsyncCmd, video_grid::VideoGrid}; use egui_wgpu::WgpuConfiguration; +use livekit::room::track::remote_track::RemoteTrackHandle; use parking_lot::Mutex; use std::sync::{ atomic::{AtomicBool, Ordering}, @@ -9,10 +10,11 @@ use std::sync::{ }; use tokio::sync::mpsc; -use livekit::room::Room; +use livekit::room::{ConnectionState, Room, RoomError}; -const URL: &str = "ws://localhost:7880"; -const TOKEN : &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY0NzMsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NjQ4MDY0NzMsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.BgVdBnq3XFD3_BQHoe1azqjifYysubgFl6Qlzu9IQGI"; +// Useful default constants for developing +const DEFAULT_URL: &str = "ws://localhost:7880"; +const DEFAULT_TOKEN : &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY0NzMsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NjQ4MDY0NzMsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.BgVdBnq3XFD3_BQHoe1azqjifYysubgFl6Qlzu9IQGI"; // eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY3MzAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ3ZWIiLCJuYmYiOjE2NjQ4MDY3MzAsInN1YiI6IndlYiIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.VbDoULjX1CVGZu2sPy3SvWYlVZUBXxQVPmdB9BnmlN4 @@ -30,16 +32,19 @@ struct AppState { struct App { state: Arc, - renderers: Vec, + video_renderers: Vec, egui_context: egui::Context, egui_state: egui_winit::State, egui_painter: egui_wgpu::winit::Painter, window: winit::window::Window, - event_tx: mpsc::UnboundedSender, + cmd_tx: mpsc::UnboundedSender, + cmd_rx: mpsc::UnboundedReceiver, // UI State lk_url: String, lk_token: String, + connection_failure: Option, + room_state: ConnectionState, } pub fn run(rt: tokio::runtime::Runtime) { @@ -58,8 +63,8 @@ pub fn run(rt: tokio::runtime::Runtime) { egui_painter.set_window(Some(&window)); } - let (event_tx, mut event_rx) = mpsc::unbounded_channel::(); - let (event_tx, mut event_rx) = mpsc::unbounded_channel::(); + let (async_cmd_tx, mut async_cmd_rx) = mpsc::unbounded_channel::(); + let (ui_cmd_tx, ui_cmd_rx) = mpsc::unbounded_channel::(); let state = Arc::new(AppState { room: Mutex::new(Room::new()), @@ -68,25 +73,44 @@ pub fn run(rt: tokio::runtime::Runtime) { let mut app = App { state: state.clone(), - renderers: Vec::default(), + video_renderers: Vec::default(), egui_context, egui_state, egui_painter, window, - event_tx, - lk_url: "ws://localhost:8080/".to_owned(), - lk_token: "your token".to_owned(), + cmd_tx: async_cmd_tx, + cmd_rx: ui_cmd_rx, + lk_url: DEFAULT_URL.to_owned(), + lk_token: DEFAULT_TOKEN.to_owned(), + connection_failure: None, + room_state: ConnectionState::Connected, }; // Async event loop tokio::spawn(async move { - while let Some(event) = event_rx.recv().await { + { + let events = state.room.lock().events(); + events.on_track_subscribed({ + let ui_cmd_tx = ui_cmd_tx.clone(); + move |event| { + let ui_cmd_tx = ui_cmd_tx.clone(); + async move { + ui_cmd_tx.send(UiCmd::TrackSubscribed { event }).unwrap(); + } + } + }); + } + while let Some(event) = async_cmd_rx.recv().await { match event { - DemoEvent::RoomConnect { url, token } => { + AsyncCmd::RoomConnect { url, token } => { state.connecting.store(true, Ordering::SeqCst); let mut room = state.room.lock(); - room.connect(&url, &token).await.unwrap(); + ui_cmd_tx + .send(UiCmd::ConnectResult { + result: room.connect(&url, &token).await, + }) + .unwrap(); state.connecting.store(false, Ordering::SeqCst); } @@ -105,6 +129,33 @@ pub fn run(rt: tokio::runtime::Runtime) { impl App { fn update(&mut self, event: Event<'_, T>, control_flow: &mut ControlFlow) { + if let Ok(cmd) = self.cmd_rx.try_recv() { + match cmd { + UiCmd::ConnectResult { result } => { + if let Err(err) = result { + self.connection_failure = Some(err.to_string()); + } else { + self.connection_failure = None + } + } + UiCmd::TrackSubscribed { event } => { + match event.track { + RemoteTrackHandle::Video(video_track) => { + // Create a new VideoRenderer + let video_renderer = VideoRenderer::new( + self.egui_painter.render_state().clone().unwrap(), + video_track.rtc_track(), + ); + self.video_renderers.push(video_renderer); + } + RemoteTrackHandle::Audio(_) => { + // The demo doesn't support Audio rendering at the moment. + } + }; + } + } + } + match event { Event::WindowEvent { window_id, event } => { if let Some(flow) = self.on_window_event(window_id, event) { @@ -157,16 +208,16 @@ impl App { if ui.button("Logs").clicked() {} if ui.button("Profiler").clicked() {} if ui.button("WebRTC Stats").clicked() {} + if ui.button("Events").clicked() {} }); ui.menu_button("Simulate", |ui| {}); }); }); egui::SidePanel::right("room_panel") - .default_width(128.0) + .default_width(256.0) .show(ui.ctx(), |ui| { ui.heading("Livekit - Connect to a room"); - ui.separator(); ui.horizontal(|ui| { @@ -182,9 +233,11 @@ impl App { ui.horizontal(|ui| { let connecting = self.state.connecting.load(Ordering::SeqCst); ui.set_enabled(!connecting); + if ui.button("Connect").clicked() { - self.event_tx - .send(DemoEvent::RoomConnect { + self.connection_failure = None; + self.cmd_tx + .send(AsyncCmd::RoomConnect { url: self.lk_url.clone(), token: self.lk_token.clone(), }) @@ -196,7 +249,11 @@ impl App { } }); - ui.allocate_space(ui.available_size()); + if let Some(err) = &self.connection_failure { + ui.colored_label(egui::Color32::RED, err); + } + + ui.separator(); }); egui::CentralPanel::default().show(ui.ctx(), |ui| { @@ -204,14 +261,33 @@ impl App { VideoGrid::new("default_grid") .max_columns(6) .show(ui, |ui| { - for _ in 0..20 { - ui.video_frame(|ui| { - egui::Frame::none() - .fill(egui::Color32::DARK_GRAY) - .show(ui, |ui| { - ui.allocate_space(ui.available_size()); - }); - }); + if self.room_state == ConnectionState::Disconnected { + for _ in 0..20 { + ui.video_frame(|ui| { + egui::Frame::none().fill(egui::Color32::DARK_GRAY).show( + ui, + |ui| { + ui.allocate_space(ui.available_size()); + }, + ); + }); + } + } else { + for video_renderer in &self.video_renderers { + ui.video_frame(|ui| { + if let Some(tex) = video_renderer.texture_id() { + ui.painter().image( + tex, + ui.available_rect_before_wrap(), + egui::Rect::from_min_max( + egui::pos2(0.0, 0.0), + egui::pos2(1.0, 1.0), + ), + egui::Color32::WHITE, + ); + } + }); + } } }); }); diff --git a/examples/simple_room/src/events.rs b/examples/simple_room/src/events.rs index 29d785e..92c8d8d 100644 --- a/examples/simple_room/src/events.rs +++ b/examples/simple_room/src/events.rs @@ -1,3 +1,5 @@ +use livekit::events::TrackSubscribedEvent; + #[derive(Debug)] pub enum AsyncCmd { RoomConnect { url: String, token: String }, @@ -5,5 +7,10 @@ pub enum AsyncCmd { #[derive(Debug)] pub enum UiCmd { - ConnectResult, + ConnectResult { + result: livekit::room::RoomResult<()>, + }, + TrackSubscribed { + event: TrackSubscribedEvent, + }, } diff --git a/examples/simple_room/src/main.rs b/examples/simple_room/src/main.rs index cebde08..0d97987 100644 --- a/examples/simple_room/src/main.rs +++ b/examples/simple_room/src/main.rs @@ -1,9 +1,16 @@ +use tracing_subscriber::prelude::*; +mod app; mod events; mod video_grid; mod video_renderer; -mod app; fn main() { + let fmt_layer = tracing_subscriber::fmt::Layer::default(); + + tracing_subscriber::registry() + .with(fmt_layer) + .init(); + let rt = tokio::runtime::Builder::new_multi_thread() .enable_all() .build() diff --git a/examples/simple_room/src/video_renderer.rs b/examples/simple_room/src/video_renderer.rs index 6ae4371..0f0da9e 100644 --- a/examples/simple_room/src/video_renderer.rs +++ b/examples/simple_room/src/video_renderer.rs @@ -3,6 +3,7 @@ use livekit::webrtc::video_frame_buffer::PlanarYuv8Buffer; use livekit::webrtc::video_frame_buffer::PlanarYuvBuffer; use livekit::webrtc::video_frame_buffer::VideoFrameBufferTrait; use livekit::webrtc::yuv_helper; +use tracing::debug_span; use std::convert::TryInto; use std::num::NonZeroU32; use std::{ @@ -101,6 +102,9 @@ impl VideoRenderer { let internal = internal.clone(); Box::new(move |_frame, buffer| { + let span = debug_span!("texture_upload"); + let _enter = span.enter(); + let mut internal = internal.lock().unwrap(); let buffer = buffer.to_i420();