fix participant updates (#23)

livekit/src/room/room_session.rs

@@ -14,7 +14,7 @@ use std::sync::Arc;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
-use tracing::{error, instrument, Level};
+use tracing::{error, info, instrument, Level};
 
 #[derive(Debug, Clone, Copy, Eq, PartialEq)]
 pub enum ConnectionState {
@@ -253,7 +253,7 @@ impl SessionInner {
     #[instrument(level = Level::DEBUG)]
     async fn on_engine_event(self: &Arc<Self>, event: EngineEvent) -> RoomResult<()> {
         match event {
-            EngineEvent::ParticipantUpdate(update) => self.handle_participant_update(update),
+            EngineEvent::ParticipantUpdate { updates } => self.handle_participant_update(updates),
             EngineEvent::MediaTrack {
                 track,
                 stream,
@@ -352,8 +352,9 @@ impl SessionInner {
     /// It'll create, update or remove a participant
     /// It also update the participant tracks.
     #[instrument(level = Level::DEBUG)]
-    fn handle_participant_update(self: &Arc<Self>, update: proto::ParticipantUpdate) {
-        for pi in update.participants {
+    fn handle_participant_update(self: &Arc<Self>, updates: Vec<proto::ParticipantInfo>) {
+        for pi in updates {
+            info!("test");
             if pi.sid == self.local_participant.sid()
                 || pi.identity == self.local_participant.identity()
             {
@@ -366,6 +367,7 @@ impl SessionInner {
             if let Some(remote_participant) = remote_participant {
                 if pi.state == participant_info::State::Disconnected as i32 {
                     // Participant disconnected
+                    info!("Participant disconnected: {}", pi.sid);
                     self.clone()
                         .handle_participant_disconnect(remote_participant)
                 } else {
@@ -374,6 +376,7 @@ impl SessionInner {
                 }
             } else {
                 // Create a new participant
+                info!("Participant connected: {}", pi.sid);
                 let remote_participant = {
                     let pi = pi.clone();
                     self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
@@ -478,9 +481,7 @@ impl SessionInner {
             self.local_participant.update_info(pi, true); // The sid may have changed
         }
 
-        self.handle_participant_update(proto::ParticipantUpdate {
-            participants: join_response.other_participants,
-        });
+        self.handle_participant_update(join_response.other_participants);
 
         // TODO(theomonnom): unpublish & republish tracks
     }

livekit/src/rtc_engine/mod.rs

@@ -69,7 +69,9 @@ pub enum EngineError {
 
 #[derive(Debug)]
 pub enum EngineEvent {
-    ParticipantUpdate(ParticipantUpdate),
+    ParticipantUpdate {
+        updates: Vec<proto::ParticipantInfo>,
+    },
     MediaTrack {
         track: MediaStreamTrackHandle,
         stream: MediaStream,
@@ -282,6 +284,12 @@ impl EngineInner {
                     })
                     .await;
             }
+            SessionEvent::ParticipantUpdate { updates } => {
+                let _ = self
+                    .engine_emitter
+                    .send(EngineEvent::ParticipantUpdate { updates })
+                    .await;
+            }
             SessionEvent::SpeakersChanged { speakers } => {
                 let _ = self
                     .engine_emitter

livekit/src/rtc_engine/rtc_session.rs

@@ -43,6 +43,9 @@ pub type SessionEvents = mpsc::UnboundedReceiver<SessionEvent>;
 
 #[derive(Debug)]
 pub enum SessionEvent {
+    ParticipantUpdate {
+        updates: Vec<proto::ParticipantInfo>,
+    },
     Data {
         participant_sid: String,
         payload: Vec<u8>,
@@ -421,6 +424,11 @@ impl SessionInner {
                     true,
                 );
             }
+            signal_response::Message::Update(update) => {
+                let _ = self.emitter.send(SessionEvent::ParticipantUpdate {
+                    updates: update.participants,
+                });
+            }
             signal_response::Message::SpeakersChanged(speaker) => {
                 let _ = self.emitter.send(SessionEvent::SpeakersChanged {
                     speakers: speaker.speakers,