Add data_stream trailer event

livekit/src/room/mod.rs

@@ -176,6 +176,10 @@ pub enum RoomEvent {
         chunk: proto::data_stream::Chunk,
         participant_identity: String,
     },
+    StreamTrailerReceived {
+        trailer: proto::data_stream::Trailer,
+        participant_identity: String,
+    },
     E2eeStateChanged {
         participant: Participant,
         state: EncryptionState,
@@ -734,6 +738,9 @@ impl RoomSession {
             EngineEvent::DataStreamChunk { chunk, participant_identity } => {
                 self.handle_data_stream_chunk(chunk, participant_identity);
             }
+            EngineEvent::DataStreamTrailer { trailer, participant_identity } => {
+                self.handle_data_stream_trailer(chunk, participant_identity);
+            }
             _ => {}
         }
 
@@ -1262,6 +1269,15 @@ impl RoomSession {
         self.dispatcher.dispatch(&event);
     }
 
+    fn handle_data_stream_trailer(
+        &self,
+        trailer: proto::data_stream::Trailer,
+        participant_identity: String,
+    ) {
+        let event = RoomEvent::StreamTrailerReceived { trailer, participant_identity };
+        self.dispatcher.dispatch(&event);
+    }
+
     /// Create a new participant
     /// Also add it to the participants list
     fn create_participant(

livekit/src/rtc_engine/mod.rs

@@ -167,6 +167,10 @@ pub enum EngineEvent {
         chunk: proto::data_stream::Chunk,
         participant_identity: String,
     },
+    DataStreamTrailer {
+        trailer: proto::data_stream::Trailer,
+        participant_identity: String,
+    },
 }
 
 /// Represents a running RtcSession with the ability to close the session
@@ -542,6 +546,11 @@ impl EngineInner {
                     .engine_tx
                     .send(EngineEvent::DataStreamChunk { chunk, participant_identity });
             }
+            SessionEvent::DataStreamTrailer { trailer, participant_identity } => {
+                let _ = self
+                    .engine_tx
+                    .send(EngineEvent::DataStreamTrailer { trailer, participant_identity });
+            }
         }
         Ok(())
     }

livekit/src/rtc_engine/rtc_session.rs

@@ -143,6 +143,10 @@ pub enum SessionEvent {
         chunk: proto::data_stream::Chunk,
         participant_identity: String,
     },
+    DataStreamTrailer {
+        trailer: proto::data_stream::Trailer,
+        participant_identity: String,
+    },
 }
 
 #[derive(Serialize, Deserialize)]
@@ -743,6 +747,12 @@ impl SessionInner {
                                 participant_identity: data.participant_identity.clone(),
                             });
                         }
+                        proto::data_packet::Value::StreamTrailer(message) => {
+                            let _ = self.emitter.send(SessionEvent::DataStreamTrailer {
+                                trailer: message.clone(),
+                                participant_identity: data.participant_identity.clone(),
+                            });
+                        }
                         _ => {}
                     }
                 }