Expose room updates, support MoveParticipant (protocol 15) (#662)

* Expose room updates, support MoveParticipant (protocol 15)

giving clients other information about the room so they could receive

* also fix sid updates

* remove dead var

Author: davidzhao

livekit-api/src/services/sip.rs

@@ -236,6 +236,7 @@ impl SIPClient {
                         // TODO: support these attributes
                         include_headers: Default::default(),
                         media_encryption: Default::default(),
+                        destination_country: Default::default(),
                     }),
                 },
                 self.base.auth_header(

livekit-api/src/signal_client/mod.rs

@@ -47,7 +47,7 @@ pub type SignalEvents = mpsc::UnboundedReceiver<SignalEvent>;
 pub type SignalResult<T> = Result<T, SignalError>;
 
 pub const JOIN_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
-pub const PROTOCOL_VERSION: u32 = 15;
+pub const PROTOCOL_VERSION: u32 = 16;
 
 #[derive(Error, Debug)]
 pub enum SignalError {

livekit-ffi/protocol/participant.proto

@@ -71,4 +71,5 @@ enum DisconnectReason {
   // SIP protocol failure or unexpected response
   SIP_TRUNK_FAILURE = 13;
   CONNECTION_TIMEOUT = 14;
+  MEDIA_FAILURE = 15;
 }

livekit-ffi/protocol/room.proto

@@ -376,6 +376,12 @@ message RoomEvent {
     // Data stream (high level)
     ByteStreamOpened byte_stream_opened = 34;
     TextStreamOpened text_stream_opened = 35;
+    // Room info updated
+    RoomInfo room_updated = 36;
+    // Participant moved to new room
+    RoomInfo moved = 37;
+    // carry over all participant info updates, including sid
+    ParticipantsUpdated participants_updated = 38;
   }
 }
 
@@ -385,13 +391,24 @@ message RoomInfo {
   required string metadata = 3;
   required uint64 lossy_dc_buffered_amount_low_threshold = 4;
   required uint64 reliable_dc_buffered_amount_low_threshold = 5;
+  required uint32 empty_timeout = 6;
+  required uint32 departure_timeout = 7;
+  required uint32 max_participants = 8;
+  required int64 creation_time = 9;
+  required uint32 num_participants = 10;
+  required uint32 num_publishers = 11;
+  required bool active_recording = 12;
 }
 
 message OwnedRoom {
   required FfiOwnedHandle handle = 1;
   required RoomInfo info = 2;
 }
 
+message ParticipantsUpdated {
+  repeated ParticipantInfo participants = 1;
+}
+
 message ParticipantConnected { required OwnedParticipant info = 1; }
 
 message ParticipantDisconnected {

livekit-ffi/src/conversion/participant.rs

@@ -13,12 +13,18 @@
 // limitations under the License.
 
 use crate::{proto, server::participant::FfiParticipant};
+use livekit::prelude::*;
 use livekit::DisconnectReason;
 use livekit::ParticipantKind;
 
 impl From<&FfiParticipant> for proto::ParticipantInfo {
     fn from(value: &FfiParticipant) -> Self {
-        let participant = &value.participant;
+        From::<&Participant>::from(&value.participant)
+    }
+}
+
+impl From<&Participant> for proto::ParticipantInfo {
+    fn from(participant: &Participant) -> Self {
         Self {
             sid: participant.sid().into(),
             name: participant.name(),
@@ -62,6 +68,7 @@ impl From<DisconnectReason> for proto::DisconnectReason {
             DisconnectReason::UserRejected => proto::DisconnectReason::UserRejected,
             DisconnectReason::SipTrunkFailure => proto::DisconnectReason::SipTrunkFailure,
             DisconnectReason::ConnectionTimeout => proto::DisconnectReason::ConnectionTimeout,
+            DisconnectReason::MediaFailure => proto::DisconnectReason::MediaFailure,
         }
     }
 }

livekit-ffi/src/conversion/room.rs

@@ -24,6 +24,7 @@ use livekit::{
         native::frame_cryptor::EncryptionState,
         prelude::{ContinualGatheringPolicy, IceServer, IceTransportsType, RtcConfiguration},
     },
+    RoomInfo,
 };
 
 impl From<EncryptionState> for proto::EncryptionState {
@@ -99,6 +100,7 @@ impl From<DisconnectReason> for proto::DisconnectReason {
             DisconnectReason::UserRejected => Self::UserRejected,
             DisconnectReason::SipTrunkFailure => Self::SipTrunkFailure,
             DisconnectReason::ConnectionTimeout => Self::ConnectionTimeout,
+            DisconnectReason::MediaFailure => Self::MediaFailure,
         }
     }
 }
@@ -249,7 +251,7 @@ impl From<proto::AudioEncoding> for AudioEncoding {
 impl From<&FfiRoom> for proto::RoomInfo {
     fn from(value: &FfiRoom) -> Self {
         let room = &value.inner.room;
-        Self {
+        proto::RoomInfo {
             sid: room.maybe_sid().map(|x| x.to_string()),
             name: room.name(),
             metadata: room.metadata(),
@@ -259,6 +261,36 @@ impl From<&FfiRoom> for proto::RoomInfo {
             reliable_dc_buffered_amount_low_threshold: room
                 .data_channel_options(DataPacketKind::Reliable)
                 .buffered_amount_low_threshold,
+            empty_timeout: room.empty_timeout(),
+            departure_timeout: room.departure_timeout(),
+            max_participants: room.max_participants(),
+            creation_time: room.creation_time(),
+            num_participants: room.num_participants(),
+            num_publishers: room.num_publishers(),
+            active_recording: room.active_recording(),
+        }
+    }
+}
+
+impl From<RoomInfo> for proto::RoomInfo {
+    fn from(room: RoomInfo) -> Self {
+        proto::RoomInfo {
+            sid: room.sid.map(|x| x.to_string()),
+            name: room.name,
+            metadata: room.metadata,
+            lossy_dc_buffered_amount_low_threshold: room
+                .lossy_dc_options
+                .buffered_amount_low_threshold,
+            reliable_dc_buffered_amount_low_threshold: room
+                .reliable_dc_options
+                .buffered_amount_low_threshold,
+            empty_timeout: room.empty_timeout,
+            departure_timeout: room.departure_timeout,
+            max_participants: room.max_participants,
+            creation_time: room.creation_time,
+            num_participants: room.num_participants,
+            num_publishers: room.num_publishers,
+            active_recording: room.active_recording,
         }
     }
 }

livekit-ffi/src/livekit.proto.rs

@@ -1739,6 +1739,7 @@ pub enum DisconnectReason {
     /// SIP protocol failure or unexpected response
     SipTrunkFailure = 13,
     ConnectionTimeout = 14,
+    MediaFailure = 15,
 }
 impl DisconnectReason {
     /// String value of the enum field names used in the ProtoBuf definition.
@@ -1762,6 +1763,7 @@ impl DisconnectReason {
             DisconnectReason::UserRejected => "USER_REJECTED",
             DisconnectReason::SipTrunkFailure => "SIP_TRUNK_FAILURE",
             DisconnectReason::ConnectionTimeout => "CONNECTION_TIMEOUT",
+            DisconnectReason::MediaFailure => "MEDIA_FAILURE",
         }
     }
     /// Creates an enum from field names used in the ProtoBuf definition.
@@ -1782,6 +1784,7 @@ impl DisconnectReason {
             "USER_REJECTED" => Some(Self::UserRejected),
             "SIP_TRUNK_FAILURE" => Some(Self::SipTrunkFailure),
             "CONNECTION_TIMEOUT" => Some(Self::ConnectionTimeout),
+            "MEDIA_FAILURE" => Some(Self::MediaFailure),
             _ => None,
         }
     }
@@ -3329,7 +3332,7 @@ pub struct OwnedBuffer {
 pub struct RoomEvent {
     #[prost(uint64, required, tag="1")]
     pub room_handle: u64,
-    #[prost(oneof="room_event::Message", tags="2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35")]
+    #[prost(oneof="room_event::Message", tags="2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38")]
     pub message: ::core::option::Option<room_event::Message>,
 }
 /// Nested message and enum types in `RoomEvent`.
@@ -3409,6 +3412,15 @@ pub mod room_event {
         ByteStreamOpened(super::ByteStreamOpened),
         #[prost(message, tag="35")]
         TextStreamOpened(super::TextStreamOpened),
+        /// Room info updated
+        #[prost(message, tag="36")]
+        RoomUpdated(super::RoomInfo),
+        /// Participant moved to new room
+        #[prost(message, tag="37")]
+        Moved(super::RoomInfo),
+        /// carry over all participant info updates, including sid
+        #[prost(message, tag="38")]
+        ParticipantsUpdated(super::ParticipantsUpdated),
     }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
@@ -3424,6 +3436,20 @@ pub struct RoomInfo {
     pub lossy_dc_buffered_amount_low_threshold: u64,
     #[prost(uint64, required, tag="5")]
     pub reliable_dc_buffered_amount_low_threshold: u64,
+    #[prost(uint32, required, tag="6")]
+    pub empty_timeout: u32,
+    #[prost(uint32, required, tag="7")]
+    pub departure_timeout: u32,
+    #[prost(uint32, required, tag="8")]
+    pub max_participants: u32,
+    #[prost(int64, required, tag="9")]
+    pub creation_time: i64,
+    #[prost(uint32, required, tag="10")]
+    pub num_participants: u32,
+    #[prost(uint32, required, tag="11")]
+    pub num_publishers: u32,
+    #[prost(bool, required, tag="12")]
+    pub active_recording: bool,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -3435,6 +3461,12 @@ pub struct OwnedRoom {
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ParticipantsUpdated {
+    #[prost(message, repeated, tag="1")]
+    pub participants: ::prost::alloc::vec::Vec<ParticipantInfo>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ParticipantConnected {
     #[prost(message, required, tag="1")]
     pub info: OwnedParticipant,

livekit-ffi/src/server/room.rs

@@ -629,7 +629,6 @@ impl RoomInner {
                     send_chat_message.sender_identity,
                 )
                 .await;
-            let sent_message = res.as_ref().unwrap().clone();
             match res {
                 Ok(message) => {
                     let _ = server.send_event(proto::ffi_event::Message::ChatMessage(
@@ -652,7 +651,6 @@ impl RoomInner {
                     ));
                 }
             }
-            sent_message;
         });
         server.watch_panic(handle);
         proto::SendChatMessageResponse { async_id }
@@ -676,7 +674,6 @@ impl RoomInner {
                     edit_chat_message.sender_identity,
                 )
                 .await;
-            let sent_message: ChatMessage = res.as_ref().unwrap().clone();
             match res {
                 Ok(message) => {
                     let _ = server.send_event(proto::ffi_event::Message::ChatMessage(
@@ -699,7 +696,6 @@ impl RoomInner {
                     ));
                 }
             }
-            sent_message;
         });
         server.watch_panic(handle);
         proto::SendChatMessageResponse { async_id }
@@ -720,6 +716,7 @@ impl RoomInner {
                 send_stream_header.header.into(),
             )
             .into(),
+            ..Default::default()
         };
         let async_id = server.next_id();
         let inner = self.clone();
@@ -748,6 +745,7 @@ impl RoomInner {
                 send_stream_chunk.chunk.into(),
             )
             .into(),
+            ..Default::default()
         };
         let async_id = server.next_id();
         let inner = self.clone();
@@ -777,6 +775,7 @@ impl RoomInner {
                 send_stream_trailer.trailer.into(),
             )
             .into(),
+            ..Default::default()
         };
         let async_id = server.next_id();
         let inner = self.clone();
@@ -1248,7 +1247,7 @@ async fn forward_event(
             ));
         }
         RoomEvent::SipDTMFReceived { code, digit, participant } => {
-            let (sid, identity) = match participant {
+            let (_sid, identity) = match participant {
                 Some(p) => (Some(p.sid().to_string()), p.identity().to_string()),
                 None => (None, String::new()),
             };
@@ -1265,7 +1264,7 @@ async fn forward_event(
         }
 
         RoomEvent::ChatMessage { message, participant } => {
-            let (sid, identity) = match participant {
+            let (_sid, identity) = match participant {
                 Some(p) => (Some(p.sid().to_string()), p.identity().to_string()),
                 None => (None, String::new()),
             };
@@ -1359,6 +1358,22 @@ async fn forward_event(
                 },
             ));
         }
+        RoomEvent::RoomUpdated { room } => {
+            let _ = send_event(proto::room_event::Message::RoomUpdated(room.into()));
+        }
+        RoomEvent::Moved { room } => {
+            let _ = send_event(proto::room_event::Message::Moved(room.into()));
+        }
+        RoomEvent::ParticipantsUpdated { participants } => {
+            let _ = send_event(proto::room_event::Message::ParticipantsUpdated(
+                proto::ParticipantsUpdated {
+                    participants: participants
+                        .into_iter()
+                        .map(|p| proto::ParticipantInfo::from(&p))
+                        .collect(),
+                },
+            ));
+        }
         _ => {
             log::warn!("unhandled room event: {:?}", event);
         }