DataReceived event

crates/livekit-core/src/rtc_engine/mod.rs

@@ -73,6 +73,11 @@ pub enum EngineEvent {
         stream: MediaStream,
         receiver: RtpReceiver,
     },
+    Data {
+        participant_sid: String,
+        payload: Vec<u8>,
+        kind: data_packet::Kind,
+    },
     Resuming,
     Resumed,
     Restarting,
@@ -241,7 +246,20 @@ impl EngineInner {
                     self.close().await;
                 }
             }
-            SessionEvent::Data { data: _ } => {}
+            SessionEvent::Data {
+                participant_sid,
+                payload,
+                kind,
+            } => {
+                let _ = self
+                    .engine_emitter
+                    .send(EngineEvent::Data {
+                        participant_sid,
+                        payload,
+                        kind,
+                    })
+                    .await;
+            }
             SessionEvent::MediaTrack {
                 track,
                 stream,

crates/livekit-core/src/rtc_engine/rtc_session.rs

@@ -44,7 +44,9 @@ pub type SessionEvents = mpsc::UnboundedReceiver<SessionEvent>;
 #[derive(Debug)]
 pub enum SessionEvent {
     Data {
-        data: Vec<u8>,
+        participant_sid: String,
+        payload: Vec<u8>,
+        kind: proto::data_packet::Kind,
     },
     MediaTrack {
         track: MediaStreamTrackHandle,
@@ -501,8 +503,12 @@ impl SessionInner {
 
                 let data = DataPacket::decode(&*data)?;
                 match data.value.unwrap() {
-                    Value::User(_user) => {
-                        // TODO(theomonnom) Send event
+                    Value::User(user) => {
+                        let _ = self.emitter.send(SessionEvent::Data {
+                            participant_sid: user.participant_sid,
+                            payload: user.payload,
+                            kind: data_packet::Kind::from_i32(data.kind).unwrap(),
+                        });
                     }
                     Value::Speaker(_) => {
                         // TODO(theomonnonm)