async room.sid(&self)

Author: nbsp

livekit-ffi/src/conversion/room.rs

@@ -213,8 +213,9 @@ impl From<proto::AudioEncoding> for AudioEncoding {
 }
 
 impl From<&FfiRoom> for proto::RoomInfo {
-    fn from(value: &FfiRoom) -> Self {
+    #[tokio::main]
+    async fn from(value: &FfiRoom) -> Self {
         let room = &value.inner.room;
-        Self { sid: room.sid().into(), name: room.name(), metadata: room.metadata() }
+        Self { sid: room.sid().await.into(), name: room.name(), metadata: room.metadata() }
     }
 }

livekit/src/room/id.rs

@@ -15,7 +15,7 @@ pub struct ParticipantIdentity(pub String);
 #[derive(Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
 pub struct TrackSid(String);
 
-#[derive(Clone, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
+#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
 pub struct RoomSid(String);
 
 impl From<String> for ParticipantIdentity {

livekit/src/room/mod.rs

@@ -246,9 +246,10 @@ pub struct Room {
 }
 
 impl Debug for Room {
-    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+    #[tokio::main]
+    async fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         f.debug_struct("Room")
-            .field("sid", &self.sid())
+            .field("sid", &self.sid().await)
             .field("name", &self.name())
             .field("connection_state", &self.connection_state())
             .finish()
@@ -262,7 +263,9 @@ struct RoomInfo {
 
 pub(crate) struct RoomSession {
     rtc_engine: Arc<RtcEngine>,
-    sid: RoomSid,
+    sid_tx: AsyncMutex<Option<oneshot::Sender<()>>>,
+    sid_rx: AsyncMutex<oneshot::Receiver<()>>,
+    sid: RwLock<RoomSid>,
     name: String,
     info: RwLock<RoomInfo>,
     dispatcher: Dispatcher<RoomEvent>,
@@ -386,8 +389,11 @@ impl Room {
         });
 
         let room_info = join_response.room.unwrap();
+        let (send, recv) = oneshot::channel();
         let inner = Arc::new(RoomSession {
-            sid: room_info.sid.try_into().unwrap(),
+            sid_tx: AsyncMutex::new(Some(send)),
+            sid_rx: AsyncMutex::new(recv),
+            sid: Default::default(),
             name: room_info.name,
             info: RwLock::new(RoomInfo {
                 state: ConnectionState::Disconnected,
@@ -403,6 +409,8 @@ impl Room {
             room_task: Default::default(),
         });
 
+        inner.set_sid(room_info.sid.try_into().unwrap());
+
         e2ee_manager.on_state_changed({
             let dispatcher = dispatcher.clone();
             let inner = inner.clone();
@@ -476,8 +484,9 @@ impl Room {
         self.inner.dispatcher.register()
     }
 
-    pub fn sid(&self) -> RoomSid {
-        self.inner.sid.clone()
+    pub async fn sid(&self) -> RoomSid {
+        let _ = self.inner.sid_rx.lock().await;
+        self.inner.sid.read().clone()
     }
 
     pub fn name(&self) -> String {
@@ -677,7 +686,12 @@ impl RoomSession {
         let participant_sid: ParticipantSid = participant_sid.to_owned().try_into().unwrap();
         let stream_id = stream_id.to_owned().try_into().unwrap();
 
-        let remote_participant = self.remote_participants.read().values().find(|x| { &x.sid() == &participant_sid }).cloned();
+        let remote_participant = self
+            .remote_participants
+            .read()
+            .values()
+            .find(|x| &x.sid() == &participant_sid)
+            .cloned();
 
         if let Some(remote_participant) = remote_participant {
             livekit_runtime::spawn(async move {
@@ -836,6 +850,7 @@ impl RoomSession {
                 metadata: info.metadata.clone(),
             });
         }
+        self.set_sid(room.sid.try_into().unwrap());
     }
 
     fn handle_resuming(self: &Arc<Self>, tx: oneshot::Sender<()>) {
@@ -1136,7 +1151,7 @@ impl RoomSession {
     }
 
     fn get_participant_by_sid(&self, sid: &ParticipantSid) -> Option<RemoteParticipant> {
-        self.remote_participants.read().values().find(|x| { &x.sid() == sid }).cloned()
+        self.remote_participants.read().values().find(|x| &x.sid() == sid).cloned()
     }
 
     fn get_participant_by_identity(
@@ -1145,6 +1160,22 @@ impl RoomSession {
     ) -> Option<RemoteParticipant> {
         self.remote_participants.read().get(identity).cloned()
     }
+
+    fn set_sid(&self, sid: RoomSid) {
+        let mut guard = self.sid_tx.blocking_lock();
+        let maybe_tx = guard.take();
+        if let Some(tx) = maybe_tx {
+            if !tx.is_closed() && sid != "".to_string().try_into().unwrap() {
+                let mut s = self.sid.write();
+                *s = sid;
+                let _ = tx.send(());
+                // self.sid_tx isn't replaced, remains None
+            } else {
+                // Sender isn't fired, gets put back in place
+                guard.replace(tx);
+            }
+        }
+    }
 }
 
 fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> {