diff --git a/crates/livekit-core/src/events.rs b/crates/livekit-core/src/events.rs deleted file mode 100644 index cca6613..0000000 --- a/crates/livekit-core/src/events.rs +++ /dev/null @@ -1,123 +0,0 @@ -use crate::room::id::TrackSid; -use crate::room::participant::remote_participant::RemoteParticipant; -use crate::room::publication::RemoteTrackPublication; -use crate::room::room_session::{ConnectionState, RoomSession}; -use crate::room::track::remote_track::RemoteTrackHandle; -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>; - -macro_rules! event_setter { - ($fnc:ident, $event:ty) => { - pub fn $fnc(&self, mut callback: F) - where - F: FnMut($event) -> Fut + Send + Sync + 'static, - Fut: Future + Send + 'static, - { - *self.$fnc.lock() = Some(Box::new(move |event| Box::pin(callback(event)))); - } - }; -} - -#[derive(Error, Debug, Clone)] -pub enum TrackError { - #[error("could not find published track with sid: {0}")] - TrackNotFound(String), -} - -#[derive(Clone, Debug)] -pub struct ParticipantConnectedEvent { - pub room_session: RoomSession, - pub participant: Arc, -} - -#[derive(Clone, Debug)] -pub struct ParticipantDisconnectedEvent { - pub room_session: RoomSession, - pub participant: Arc, -} - -#[derive(Clone, Debug)] -pub struct TrackSubscribedEvent { - pub room_session: RoomSession, - pub track: RemoteTrackHandle, - pub publication: RemoteTrackPublication, - pub participant: Arc, -} - -#[derive(Clone, Debug)] -pub struct TrackPublishedEvent { - pub room_session: RoomSession, - pub publication: RemoteTrackPublication, - pub participant: Arc, -} - -#[derive(Clone, Debug)] -pub struct TrackSubscriptionFailedEvent { - pub room_session: RoomSession, - pub error: TrackError, - pub sid: TrackSid, - pub participant: Arc, -} - -#[derive(Clone, Debug)] -pub struct ConnectionStateChangedEvent { - pub room_session: RoomSession, - pub state: ConnectionState, -} - -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>, - pub(crate) on_connection_state_changed: Mutex>, -} - -impl Debug for RoomEvents { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "RoomEvents") - } -} - -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(Default)] -pub struct ParticipantEvents { - pub(crate) on_track_published: Mutex>, - pub(crate) on_track_subscribed: Mutex>, - pub(crate) on_track_subscription_failed: Mutex>, -} - -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 0dd45e4..cb1ca08 100644 --- a/crates/livekit-core/src/lib.rs +++ b/crates/livekit-core/src/lib.rs @@ -7,5 +7,4 @@ pub mod proto { 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 cf7da7e..aec30ce 100644 --- a/crates/livekit-core/src/room/mod.rs +++ b/crates/livekit-core/src/room/mod.rs @@ -1,25 +1,13 @@ -use parking_lot::{Mutex, RwLock}; -use std::collections::HashMap; -use std::sync::atomic::{AtomicU8, Ordering}; +use self::room_session::{ConnectionState, RoomSession, SessionHandle}; +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::rtc_engine::EngineError; +use std::fmt::Debug; use std::sync::Arc; - -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 self::room_session::{RoomInternal, RoomSession}; -use crate::events::{ - ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents, TrackPublishedEvent, - TrackSubscribedEvent, -}; -use crate::proto; -use crate::proto::participant_info; use thiserror::Error; -use tracing::{debug, error, instrument, trace_span, Level}; - -use crate::rtc_engine::{EngineError, EngineEvent, EngineEvents, EngineResult, RTCEngine}; -use crate::signal_client::SignalOptions; +use tokio::sync::mpsc; pub use crate::rtc_engine::SimulateScenario; @@ -29,6 +17,10 @@ pub mod publication; pub mod room_session; pub mod track; +pub type RoomEvents = mpsc::UnboundedReceiver; +pub type RoomEmitter = mpsc::UnboundedSender; +pub type RoomResult = Result; + #[derive(Error, Debug)] pub enum RoomError { #[error("engine : {0}")] @@ -37,34 +29,54 @@ pub enum RoomError { Internal(String), } -pub type RoomResult = Result; +#[derive(Error, Debug, Clone)] +pub enum TrackError { + #[error("could not find published track with sid: {0}")] + TrackNotFound(String), +} -#[derive(Debug, Default)] +#[derive(Clone, Debug)] +pub enum RoomEvent { + ParticipantConnected(Arc), + ParticipantDisconnected(Arc), + TrackSubscribed { + track: RemoteTrackHandle, + publication: RemoteTrackPublication, + participant: Arc, + }, + TrackPublished { + publication: RemoteTrackPublication, + participant: Arc, + }, + TrackSubscriptionFailed { + error: TrackError, + sid: TrackSid, + participant: Arc, + }, + ConnectionStateChanged(ConnectionState), + Connected, + Disconnected, + Reconnecting, + Reconnected, +} + +#[derive(Debug)] pub struct Room { - internal: Option, - events: Arc, // Keep the same RoomEvents across sessions + handle: SessionHandle, } impl Room { - #[instrument(level = Level::DEBUG)] - pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> { - let internal = RoomInternal::connect(self.events.clone(), url, token).await?; - self.internal = Some(internal); - Ok(()) + pub async fn connect(url: &str, token: &str) -> RoomResult<(Self, RoomEvents)> { + let (emitter, events) = mpsc::unbounded_channel(); + let handle = SessionHandle::connect(emitter, url, token).await?; + Ok((Self { handle }, events)) } - #[instrument(level = Level::DEBUG)] - pub async fn close(&mut self) { - if let Some(internal) = self.internal.take() { - internal.close().await; - } + pub async fn close(self) { + self.handle.close().await; } - pub fn events(&self) -> Arc { - self.events.clone() - } - - pub fn session(&self) -> Option { - self.internal.as_ref().map(RoomInternal::session) + pub fn session(&self) -> RoomSession { + self.handle.session() } } diff --git a/crates/livekit-core/src/room/participant/local_participant.rs b/crates/livekit-core/src/room/participant/local_participant.rs index 8d2227f..754b165 100644 --- a/crates/livekit-core/src/room/participant/local_participant.rs +++ b/crates/livekit-core/src/room/participant/local_participant.rs @@ -1,10 +1,8 @@ -use std::sync::Weak; - use crate::proto::{data_packet, DataPacket, UserPacket}; use crate::room::participant::{ - impl_participant_trait, ParticipantInternalTrait, ParticipantShared, + impl_participant_trait, ParticipantInternalTrait, ParticipantShared, ParticipantTrait, }; -use crate::room::RoomError; +use crate::room::{RoomError, RoomEmitter}; use crate::rtc_engine::RTCEngine; #[derive(Debug)] @@ -20,9 +18,10 @@ impl LocalParticipant { identity: ParticipantIdentity, name: String, metadata: String, + room_emitter: RoomEmitter, ) -> Self { Self { - shared: ParticipantShared::new(sid, identity, name, metadata), + shared: ParticipantShared::new(sid, identity, name, metadata, room_emitter), rtc_engine, } } @@ -35,7 +34,7 @@ impl LocalParticipant { let data = DataPacket { kind: kind as i32, value: Some(data_packet::Value::User(UserPacket { - participant_sid: "".to_string(), /*self.sid().to_owned()*/ + participant_sid: self.sid().to_string(), payload: data.to_vec(), destination_sids: vec![], })), @@ -49,10 +48,6 @@ impl LocalParticipant { } 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); } diff --git a/crates/livekit-core/src/room/participant/mod.rs b/crates/livekit-core/src/room/participant/mod.rs index a243691..52781c8 100644 --- a/crates/livekit-core/src/room/participant/mod.rs +++ b/crates/livekit-core/src/room/participant/mod.rs @@ -1,4 +1,3 @@ -use crate::events::ParticipantEvents; use crate::proto::ParticipantInfo; use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid}; use crate::room::participant::local_participant::LocalParticipant; @@ -14,13 +13,12 @@ pub mod remote_participant; #[derive(Debug)] pub(super) struct ParticipantShared { - pub(super) events: Arc, - pub(super) internal_events: Arc, pub(super) sid: Mutex, pub(super) identity: Mutex, pub(super) name: Mutex, pub(super) metadata: Mutex, pub(super) tracks: RwLock>, + pub(super) room_emitter: RoomEmitter, } impl ParticipantShared { @@ -29,15 +27,15 @@ impl ParticipantShared { identity: ParticipantIdentity, name: String, metadata: String, + room_emitter: RoomEmitter, ) -> Self { Self { - events: Default::default(), - internal_events: Default::default(), sid: Mutex::new(sid), identity: Mutex::new(identity), name: Mutex::new(name), metadata: Mutex::new(metadata), tracks: Default::default(), + room_emitter } } @@ -54,12 +52,10 @@ impl ParticipantShared { } pub(crate) trait ParticipantInternalTrait { - fn internal_events(&self) -> Arc; fn update_info(&self, info: ParticipantInfo); } pub trait ParticipantTrait { - fn events(&self) -> Arc; fn sid(&self) -> ParticipantSid; fn identity(&self) -> ParticipantIdentity; fn name(&self) -> String; @@ -75,7 +71,6 @@ pub enum ParticipantHandle { impl ParticipantInternalTrait for ParticipantHandle { enum_dispatch!( [Local, Remote] - fnc!(internal_events, &Self, [], Arc); fnc!(update_info, &Self, [info: ParticipantInfo], ()); ); } @@ -83,7 +78,6 @@ impl ParticipantInternalTrait for ParticipantHandle { impl ParticipantTrait for ParticipantHandle { enum_dispatch!( [Local, Remote] - fnc!(events, &Self, [], Arc); fnc!(sid, &Self, [], ParticipantSid); fnc!(identity, &Self, [], ParticipantIdentity); fnc!(name, &Self, [], String); @@ -93,16 +87,11 @@ impl ParticipantTrait for ParticipantHandle { macro_rules! impl_participant_trait { ($x:ty) => { - use crate::events::ParticipantEvents; use crate::proto::ParticipantInfo; use crate::room::id::{ParticipantIdentity, ParticipantSid}; use std::sync::Arc; impl crate::room::participant::ParticipantTrait for $x { - fn events(&self) -> Arc { - self.shared.events.clone() - } - fn sid(&self) -> ParticipantSid { self.shared.sid.lock().clone() } @@ -123,3 +112,5 @@ macro_rules! impl_participant_trait { } pub(super) use impl_participant_trait; + +use super::RoomEmitter; diff --git a/crates/livekit-core/src/room/participant/remote_participant.rs b/crates/livekit-core/src/room/participant/remote_participant.rs index baddddb..a6caf25 100644 --- a/crates/livekit-core/src/room/participant/remote_participant.rs +++ b/crates/livekit-core/src/room/participant/remote_participant.rs @@ -1,6 +1,3 @@ -use crate::events::{ - TrackError, TrackPublishedEvent, TrackSubscribedEvent, TrackSubscriptionFailedEvent, -}; use crate::proto::TrackInfo; use crate::room::id::TrackSid; use crate::room::participant::{ @@ -14,6 +11,7 @@ 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}; +use crate::room::{RoomEmitter, RoomEvent, TrackError}; use livekit_webrtc::media_stream::MediaStreamTrackHandle; use std::collections::HashSet; use std::time::Duration; @@ -35,9 +33,10 @@ impl RemoteParticipant { identity: ParticipantIdentity, name: String, metadata: String, + room_emitter: RoomEmitter, ) -> Self { Self { - shared: ParticipantShared::new(sid, identity, name, metadata), + shared: ParticipantShared::new(sid, identity, name, metadata, room_emitter), } } @@ -109,55 +108,21 @@ impl RemoteParticipant { .add_track_publication(TrackPublication::Remote(remote_publication.clone())); track.start(); - let event = TrackSubscribedEvent { - room_session, - track, + self.shared.room_emitter.send(RoomEvent::TrackSubscribed { + track: 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_session, - 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; - } + self.shared + .room_emitter + .send(RoomEvent::TrackSubscriptionFailed { + sid: sid.clone(), + error: TrackError::TrackNotFound(sid.clone().to_string()), + participant: self.clone(), + }); } } @@ -178,25 +143,10 @@ impl RemoteParticipant { .add_track_publication(TrackPublication::Remote(publication.clone())); // This is a new track, fire publish events - let event = TrackPublishedEvent { - room_session: room_session.clone(), - participant: self.clone(), + self.shared.room_emitter.send(RoomEvent::TrackPublished { publication: publication.clone(), - }; - - if let Some(cb) = self - .shared - .internal_events - .on_track_published - .lock() - .as_mut() - { - cb(event.clone()).await; - } - - if let Some(cb) = self.shared.events.on_track_published.lock().as_mut() { - cb(event).await; - } + participant: self.clone(), + }); } valid_tracks.insert(track.sid.into()); @@ -205,10 +155,6 @@ impl RemoteParticipant { } 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) } diff --git a/crates/livekit-core/src/room/room_session.rs b/crates/livekit-core/src/room/room_session.rs index 360f786..5b72afc 100644 --- a/crates/livekit-core/src/room/room_session.rs +++ b/crates/livekit-core/src/room/room_session.rs @@ -1,26 +1,22 @@ +use super::id::{ParticipantIdentity, ParticipantSid}; +use super::participant::local_participant::LocalParticipant; +use super::participant::remote_participant::RemoteParticipant; +use super::participant::{ParticipantInternalTrait, ParticipantTrait}; +use super::{RoomEmitter, RoomError, RoomEvent, RoomResult, SimulateScenario}; +use crate::proto::{self, participant_info}; +use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RTCEngine}; +use crate::signal_client::SignalOptions; use parking_lot::{Mutex, RwLock}; use std::collections::HashMap; use std::sync::atomic::{AtomicU8, Ordering}; use std::sync::Arc; use tokio::sync::oneshot; use tokio::task::JoinHandle; - -use crate::events::{ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents}; -use crate::proto::{self, participant_info}; -use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RTCEngine}; -use crate::signal_client::SignalOptions; - -use super::id::{ParticipantIdentity, ParticipantSid}; -use super::participant::local_participant::LocalParticipant; -use super::participant::remote_participant::RemoteParticipant; -use super::participant::{ParticipantInternalTrait, ParticipantTrait}; -use super::{RoomError, RoomResult, SimulateScenario}; use tracing::{error, instrument, Level}; -#[derive(Debug, Clone, Eq, PartialEq)] +#[derive(Debug, Clone, Copy, Eq, PartialEq)] pub enum ConnectionState { Disconnected, - Connecting, Connected, Reconnecting, } @@ -31,14 +27,14 @@ impl TryFrom for ConnectionState { fn try_from(value: u8) -> Result { match value { 0 => Ok(ConnectionState::Disconnected), - 1 => Ok(ConnectionState::Connecting), - 2 => Ok(ConnectionState::Connected), - 3 => Ok(ConnectionState::Reconnecting), + 1 => Ok(ConnectionState::Connected), + 2 => Ok(ConnectionState::Reconnecting), _ => Err("invalid ConnectionState"), } } } +/// Internal representation of a RoomSession #[derive(Debug)] struct SessionInner { state: AtomicU8, // ConnectionState @@ -47,7 +43,14 @@ struct SessionInner { participants: RwLock>>, rtc_engine: Arc, local_participant: Arc, - room_events: Arc, + room_emitter: RoomEmitter, +} + +#[derive(Debug)] +pub(crate) struct SessionHandle { + session: RoomSession, + session_task: JoinHandle<()>, + close_emitter: oneshot::Sender<()>, } /// RoomSession represents a connection to a room. @@ -57,16 +60,8 @@ pub struct RoomSession { inner: Arc, } -/// Responsible for creating and closing the room session. -#[derive(Debug)] -pub struct RoomInternal { - inner: Arc, - session_task: JoinHandle<()>, - close_emitter: oneshot::Sender<()>, -} - -impl RoomInternal { - pub async fn connect(room_events: Arc, url: &str, token: &str) -> RoomResult { +impl SessionHandle { + pub async fn connect(room_emitter: RoomEmitter, url: &str, token: &str) -> RoomResult { let (rtc_engine, engine_events) = RTCEngine::new(); let rtc_engine = Arc::new(rtc_engine); rtc_engine @@ -81,16 +76,17 @@ impl RoomInternal { pi.identity.into(), pi.name, pi.metadata, + room_emitter.clone(), )); let room_info = join_response.room.unwrap(); let inner = Arc::new(SessionInner { - state: AtomicU8::new(ConnectionState::Connecting as u8), + state: AtomicU8::new(ConnectionState::Disconnected as u8), sid: Mutex::new(room_info.sid), name: Mutex::new(room_info.name), participants: Default::default(), rtc_engine, local_participant, - room_events, + room_emitter, }); for pi in join_response.other_participants { @@ -107,8 +103,12 @@ impl RoomInternal { let (close_emitter, close_receiver) = oneshot::channel(); let session_task = tokio::spawn(inner.clone().room_task(engine_events, close_receiver)); + inner + .update_connection_state(ConnectionState::Connected) + .await; + let session = Self { - inner, + session: RoomSession::from(inner), session_task, close_emitter, }; @@ -116,17 +116,13 @@ impl RoomInternal { } pub async fn close(self) { - self.inner.close(); + self.session.inner.close().await; let _ = self.close_emitter.send(()); - self.session_task.await; + let _ = self.session_task.await; } pub fn session(&self) -> RoomSession { - RoomSession::from(self.inner.clone()) - } - - pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> { - self.inner.rtc_engine.simulate_scenario(scenario).await + self.session.clone() } } @@ -161,6 +157,7 @@ impl RoomSession { } impl SessionInner { + #[instrument(level = Level::DEBUG)] async fn room_task( self: Arc, mut engine_events: EngineEvents, @@ -168,7 +165,7 @@ impl SessionInner { ) { loop { tokio::select! { - res = engine_events.recv() => { + res = engine_events.recv() => { if let Some(event) = res { if let Err(err) = self.on_engine_event(event).await { error!("failed to handle engine event: {:?}", err); @@ -240,6 +237,7 @@ impl SessionInner { Ok(()) } + #[instrument(level = Level::DEBUG)] async fn close(&self) { self.rtc_engine.close().await; } @@ -248,6 +246,21 @@ impl SessionInner { self.participants.read().get(sid).cloned() } + /// Change the connection state and emit an event + /// Does nothing if the state is already the same + #[instrument(level = Level::DEBUG)] + async fn update_connection_state(self: &Arc, state: ConnectionState) { + let old_state = self.state.load(Ordering::Acquire); + if old_state == state as u8 { + return; + } + + self.state.store(state as u8, Ordering::Release); + let _ = self + .room_emitter + .send(RoomEvent::ConnectionStateChanged(state)); + } + /// Update the participants inside a Room. /// It'll create, update or remove a participant /// It also update the participant tracks. @@ -280,13 +293,10 @@ impl SessionInner { let pi = pi.clone(); self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata) }; - let mut handler = self.room_events.on_participant_connected.lock(); - if let Some(cb) = handler.as_mut() { - cb(ParticipantConnectedEvent { - room_session: RoomSession::from(self.clone()), - participant: remote_participant.clone(), - }); - } + + let _ = self + .room_emitter + .send(RoomEvent::ParticipantConnected(remote_participant.clone())); remote_participant.update_info(pi.clone()); remote_participant @@ -296,21 +306,20 @@ impl SessionInner { } } + /// A participant has disconnected + /// Cleanup the participant and emit an event #[instrument(level = Level::DEBUG)] fn handle_participant_disconnect(self: &Arc, remote_participant: Arc) { self.participants.write().remove(&remote_participant.sid()); // TODO(theomonnom): Unpublish all tracks - - let mut handler = self.room_events.on_participant_disconnected.lock(); - if let Some(cb) = handler.as_mut() { - cb(ParticipantDisconnectedEvent { - room_session: RoomSession::from(self.clone()), - participant: remote_participant.clone(), - }); - } + let _ = self.room_emitter.send(RoomEvent::ParticipantDisconnected( + remote_participant.clone(), + )); } + /// Create a new participant + /// Also add it to the participants list fn create_participant( self: &Arc, sid: ParticipantSid, @@ -323,46 +332,9 @@ impl SessionInner { identity, name, metadata, + self.room_emitter.clone(), )); - macro_rules! forward_event { - ($type:ident, when_connected) => { - p.internal_events().$type({ - let room_internal = self.clone(); - move |event| { - let room_internal = room_internal.clone(); - async move { - if room_internal.state.load(Ordering::SeqCst) - == ConnectionState::Connected as u8 - { - if let Some(cb) = room_internal.room_events.$type.lock().as_mut() { - cb(event).await; - } - } - } - } - }) - }; - ($type:ident) => { - p.internal_events().$type({ - let room_internal = self.clone(); - move |event| { - let room_internal = room_internal.clone(); - async move { - if let Some(cb) = room_internal.room_events.$type.lock().as_mut() { - cb(event).await; - } - } - } - }) - }; - } - - // Forward participantevents to room events - forward_event!(on_track_published, when_connected); - forward_event!(on_track_subscribed); - forward_event!(on_track_subscription_failed); - self.participants.write().insert(sid, p.clone()); p }