use dispatcher for room events (#30)

examples/simple_room/src/:w

@@ -1,15 +0,0 @@
-use livekit::prelude::*;
-use livekit::{RoomResult, SimulateScenario};
-
-#[derive(Debug)]
-pub enum AsyncCmd {
-    RoomConnect { url: String, token: String },
-    RoomDisconnect,
-    SimulateScenario { scenario: SimulateScenario },
-}
-
-#[derive(Debug)]
-pub enum UiCmd {
-    ConnectResult { result: RoomResult<()> },
-    RoomEvent { event: RoomEvent },
-}

examples/simple_room/src/app.rs

@@ -150,7 +150,7 @@ pub fn run(rt: tokio::runtime::Runtime) {
 
 async fn room_task(
     _app_state: Arc<AppState>,
-    mut room_events: RoomEvents,
+    mut room_events: mpsc::UnboundedReceiver<RoomEvent>,
     mut close_rx: oneshot::Receiver<()>,
     ui_cmd_tx: mpsc::UnboundedSender<UiCmd>,
 ) {

livekit/src/prelude.rs

@@ -2,7 +2,7 @@ pub use crate::participant::{
     LocalParticipant, Participant, ParticipantEvent, ParticipantTrait, RemoteParticipant,
 };
 
-pub use crate::{ConnectionState, Room, RoomError, RoomEvent, RoomEvents, RoomSession};
+pub use crate::{ConnectionState, Room, RoomError, RoomEvent, RoomSession};
 
 pub use crate::publication::{
     LocalTrackPublication, RemoteTrackPublication, TrackPublication, TrackPublicationTrait,

livekit/src/room/mod.rs

@@ -1,5 +1,5 @@
-use crate::prelude::*;
 use crate::participant::ConnectionQuality;
+use crate::prelude::*;
 use crate::proto;
 use crate::rtc_engine::EngineError;
 use std::fmt::Debug;
@@ -17,8 +17,6 @@ pub mod track;
 
 pub use room_session::*;
 
-pub type RoomEvents = mpsc::UnboundedReceiver<RoomEvent>;
-pub type RoomEmitter = mpsc::UnboundedSender<RoomEvent>;
 pub type RoomResult<T> = Result<T, RoomError>;
 
 #[derive(Error, Debug)]
@@ -89,9 +87,12 @@ pub struct Room {
 }
 
 impl Room {
-    pub async fn connect(url: &str, token: &str) -> RoomResult<(Self, RoomEvents)> {
+    pub async fn connect(
-        let (emitter, events) = mpsc::unbounded_channel();
+        url: &str,
-        let handle = SessionHandle::connect(emitter, url, token).await?;
+        token: &str,
+    ) -> RoomResult<(Self, mpsc::UnboundedReceiver<RoomEvent>)> {
+        let handle = SessionHandle::connect(url, token).await?;
+        let events = handle.subscribe();
         Ok((Self { handle }, events))
     }
 
@@ -99,6 +100,11 @@ impl Room {
         self.handle.close().await;
     }
 
+    /// Allow multiple subscribers/observers to receive events
+    pub fn subscribe(&self) -> mpsc::UnboundedReceiver<RoomEvent> {
+        self.handle.subscribe()
+    }
+
     pub fn session(&self) -> RoomSession {
         self.handle.session()
     }

livekit/src/room/room_session.rs

@@ -3,7 +3,8 @@ use crate::prelude::*;
 use crate::proto;
 use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RTCEngine};
 use crate::signal_client::SignalOptions;
-use crate::{RoomEmitter, RoomError, RoomEvent, RoomResult, SimulateScenario};
+use crate::{RoomError, RoomEvent, RoomResult, SimulateScenario};
+use livekit_utils::observer::Dispatcher;
 use parking_lot::{Mutex, RwLock};
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicU8, Ordering};
@@ -43,7 +44,7 @@ struct SessionInner {
     active_speakers: RwLock<Vec<Participant>>,
     rtc_engine: Arc<RTCEngine>,
     local_participant: Arc<LocalParticipant>,
-    room_emitter: RoomEmitter,
+    dispatcher: Mutex<Dispatcher<RoomEvent>>,
 }
 
 #[derive(Debug)]
@@ -61,7 +62,7 @@ pub struct RoomSession {
 }
 
 impl SessionHandle {
-    pub async fn connect(room_emitter: RoomEmitter, url: &str, token: &str) -> RoomResult<Self> {
+    pub async fn connect(url: &str, token: &str) -> RoomResult<Self> {
         let (rtc_engine, engine_events) = RTCEngine::new();
         let rtc_engine = Arc::new(rtc_engine);
         rtc_engine
@@ -88,7 +89,7 @@ impl SessionHandle {
             active_speakers: Default::default(),
             rtc_engine,
             local_participant,
-            room_emitter,
+            dispatcher: Default::default(),
         });
 
         for pi in join_response.other_participants {
@@ -118,6 +119,10 @@ impl SessionHandle {
         let _ = self.session_task.await;
     }
 
+    pub fn subscribe(&self) -> mpsc::UnboundedReceiver<RoomEvent> {
+        self.session.inner.dispatcher.lock().register()
+    }
+
     pub fn session(&self) -> RoomSession {
         self.session.clone()
     }
@@ -215,30 +220,36 @@ impl SessionInner {
         if let Participant::Remote(remote_participant) = participant {
             match event {
                 ParticipantEvent::TrackPublished { publication } => {
-                    let _ = self.room_emitter.send(RoomEvent::TrackPublished {
+                    self.dispatcher.lock().dispatch(&RoomEvent::TrackPublished {
                         participant: remote_participant.clone(),
                         publication,
                     });
                 }
                 ParticipantEvent::TrackUnpublished { publication } => {
-                    let _ = self.room_emitter.send(RoomEvent::TrackUnpublished {
+                    self.dispatcher
-                        participant: remote_participant.clone(),
+                        .lock()
-                        publication,
+                        .dispatch(&RoomEvent::TrackUnpublished {
-                    });
+                            participant: remote_participant.clone(),
+                            publication,
+                        });
                 }
                 ParticipantEvent::TrackSubscribed { track, publication } => {
-                    let _ = self.room_emitter.send(RoomEvent::TrackSubscribed {
+                    self.dispatcher
-                        participant: remote_participant.clone(),
+                        .lock()
-                        track,
+                        .dispatch(&RoomEvent::TrackSubscribed {
-                        publication,
+                            participant: remote_participant.clone(),
-                    });
+                            track,
+                            publication,
+                        });
                 }
                 ParticipantEvent::TrackUnsubscribed { track, publication } => {
-                    let _ = self.room_emitter.send(RoomEvent::TrackUnsubscribed {
+                    self.dispatcher
-                        participant: remote_participant.clone(),
+                        .lock()
-                        track,
+                        .dispatch(&RoomEvent::TrackUnsubscribed {
-                        publication,
+                            participant: remote_participant.clone(),
-                    });
+                            track,
+                            publication,
+                        });
                 }
                 _ => {}
             };
@@ -286,12 +297,12 @@ impl SessionInner {
             }
             EngineEvent::Resuming => {
                 if self.update_connection_state(ConnectionState::Reconnecting) {
-                    let _ = self.room_emitter.send(RoomEvent::Reconnecting);
+                    self.dispatcher.lock().dispatch(&RoomEvent::Reconnecting);
                 }
             }
             EngineEvent::Resumed => {
                 self.update_connection_state(ConnectionState::Connected);
-                let _ = self.room_emitter.send(RoomEvent::Reconnected);
+                self.dispatcher.lock().dispatch(&RoomEvent::Reconnected);
 
                 // TODO(theomonnom): Update subscriptions settings
                 // TODO(theomonnom): Send sync state
@@ -306,7 +317,7 @@ impl SessionInner {
             } => {
                 let payload = Arc::new(payload);
                 if let Some(participant) = self.get_participant(&participant_sid.into()) {
-                    let _ = self.room_emitter.send(RoomEvent::DataReceived {
+                    self.dispatcher.lock().dispatch(&RoomEvent::DataReceived {
                         payload: payload.clone(),
                         kind,
                         participant: participant.clone(),
@@ -339,9 +350,9 @@ impl SessionInner {
         }
 
         self.state.store(state as u8, Ordering::Release);
-        let _ = self
+        self.dispatcher
-            .room_emitter
+            .lock()
-            .send(RoomEvent::ConnectionStateChanged(state));
+            .dispatch(&RoomEvent::ConnectionStateChanged(state));
         return true;
     }
 
@@ -380,8 +391,9 @@ impl SessionInner {
                 };
 
                 let _ = self
-                    .room_emitter
+                    .dispatcher
-                    .send(RoomEvent::ParticipantConnected(remote_participant.clone()));
+                    .lock()
+                    .dispatch(&RoomEvent::ParticipantConnected(remote_participant.clone()));
 
                 remote_participant.update_info(pi.clone(), true);
             }
@@ -419,8 +431,9 @@ impl SessionInner {
         *self.active_speakers.write() = speakers.clone();
 
         let _ = self
-            .room_emitter
+            .dispatcher
-            .send(RoomEvent::ActiveSpeakersChanged { speakers });
+            .lock()
+            .dispatch(&RoomEvent::ActiveSpeakersChanged { speakers });
     }
 
     /// Handle a connection quality update
@@ -446,10 +459,12 @@ impl SessionInner {
                 .into();
 
             participant.set_connection_quality(quality);
-            let _ = self.room_emitter.send(RoomEvent::ConnectionQualityChanged {
+            self.dispatcher
-                participant,
+                .lock()
-                quality,
+                .dispatch(&RoomEvent::ConnectionQualityChanged {
-            });
+                    participant,
+                    quality,
+                });
         }
     }
 
@@ -462,7 +477,7 @@ impl SessionInner {
         }
 
         if self.update_connection_state(ConnectionState::Reconnecting) {
-            let _ = self.room_emitter.send(RoomEvent::Reconnecting);
+            self.dispatcher.lock().dispatch(&RoomEvent::Reconnecting);
         }
     }
 
@@ -472,7 +487,7 @@ impl SessionInner {
         let join_response = self.rtc_engine.join_response().unwrap();
 
         self.update_connection_state(ConnectionState::Connected);
-        let _ = self.room_emitter.send(RoomEvent::Reconnected);
+        self.dispatcher.lock().dispatch(&RoomEvent::Reconnected);
 
         if let Some(pi) = join_response.participant {
             self.local_participant.update_info(pi, true); // The sid may have changed
@@ -490,7 +505,7 @@ impl SessionInner {
         }
 
         self.update_connection_state(ConnectionState::Disconnected);
-        let _ = self.room_emitter.send(RoomEvent::Disconnected);
+        self.dispatcher.lock().dispatch(&RoomEvent::Disconnected);
     }
 
     /// Create a new participant
@@ -546,9 +561,11 @@ impl SessionInner {
 
             self.participants.write().remove(&remote_participant.sid());
 
-            let _ = self.room_emitter.send(RoomEvent::ParticipantDisconnected(
+            self.dispatcher
-                remote_participant.clone(),
+                .lock()
-            ));
+                .dispatch(&RoomEvent::ParticipantDisconnected(
+                    remote_participant.clone(),
+                ));
         });
     }