diff --git a/Cargo.lock b/Cargo.lock
index 3434c4245..853d92f87 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1070,7 +1070,7 @@ dependencies = [
 
 [[package]]
 name = "livekit-ffi"
-version = "0.3.13"
+version = "0.3.15"
 dependencies = [
  "console-subscriber",
  "dashmap",
diff --git a/examples/basic_room/src/main.rs b/examples/basic_room/src/main.rs
index fb8875658..d83c04402 100644
--- a/examples/basic_room/src/main.rs
+++ b/examples/basic_room/src/main.rs
@@ -1,5 +1,5 @@
-use livekit_api::access_token;
 use livekit::prelude::*;
+use livekit_api::access_token;
 use std::env;
 
 // Connect to a room using the specified env variables
@@ -24,18 +24,17 @@ async fn main() {
         .to_jwt()
         .unwrap();
 
-
     let (room, mut rx) = Room::connect(&url, &token, RoomOptions::default())
         .await
         .unwrap();
     log::info!("Connected to room: {} - {}", room.name(), room.sid());
 
     room.local_participant()
-        .publish_data(
-            "Hello world".to_owned().into_bytes(),
-            DataPacketKind::Reliable,
-            Default::default(),
-        )
+        .publish_data(DataPacket {
+            payload: "Hello world".to_owned().into_bytes(),
+            kind: DataPacketKind::Reliable,
+            ..Default::default()
+        })
         .await
         .unwrap();
 
diff --git a/libwebrtc/src/audio_source.rs b/libwebrtc/src/audio_source.rs
index ac35a46ae..f72bb1fad 100644
--- a/libwebrtc/src/audio_source.rs
+++ b/libwebrtc/src/audio_source.rs
@@ -34,6 +34,8 @@ impl RtcAudioSource {
         [Native];
         fn set_audio_options(self: &Self, options: AudioSourceOptions) -> ();
         fn audio_options(self: &Self) -> AudioSourceOptions;
+        fn sample_rate(self: &Self) -> u32;
+        fn num_channels(self: &Self) -> u32;
     );
 }
 
diff --git a/libwebrtc/src/native/audio_source.rs b/libwebrtc/src/native/audio_source.rs
index 64cd58e67..0695d52db 100644
--- a/libwebrtc/src/native/audio_source.rs
+++ b/libwebrtc/src/native/audio_source.rs
@@ -33,6 +33,8 @@ pub struct NativeAudioSource {
 struct AudioSourceInner {
     buf: Box<[i16]>,
 
+    captured_frames: usize,
+
     // Amount of data from the previous frame that hasn't been sent to the libwebrtc source
     // (because it requires 10ms of data)
     len: usize,
@@ -51,10 +53,11 @@ impl NativeAudioSource {
     ) -> NativeAudioSource {
         let samples_10ms = (sample_rate / 100 * num_channels) as usize;
 
-        Self {
+        let source = Self {
             sys_handle: sys_at::ffi::new_audio_track_source(options.into()),
             inner: Arc::new(AsyncMutex::new(AudioSourceInner {
                 buf: vec![0; samples_10ms].into_boxed_slice(),
+                captured_frames: 0,
                 len: 0,
                 read_offset: 0,
                 interval: None, // interval must be created from a tokio runtime context
@@ -62,7 +65,35 @@ impl NativeAudioSource {
             sample_rate,
             num_channels,
             samples_10ms,
-        }
+        };
+
+        tokio::spawn({
+            let source = source.clone();
+            async move {
+                let mut interval = interval(Duration::from_millis(10));
+
+                loop {
+                    // We directly use the sys_handle instead of the capture_frame function
+                    // (We don't want to increase the captured_frames count and no need to buffer)
+                    interval.tick().await;
+
+                    let inner = source.inner.lock().await;
+                    if inner.captured_frames > 0 {
+                        break; // User captured something, stop injecting silence
+                    }
+
+                    let data = vec![0; samples_10ms];
+                    source.sys_handle.on_captured_frame(
+                        &data,
+                        sample_rate,
+                        num_channels,
+                        sample_rate as usize / 100,
+                    );
+                }
+            }
+        });
+
+        source
     }
 
     pub fn sys_handle(&self) -> SharedPtr<sys_at::ffi::AudioTrackSource> {
@@ -131,6 +162,8 @@ impl NativeAudioSource {
         }
 
         let mut inner = self.inner.lock().await;
+        inner.captured_frames += 1;
+
         let mut interval = inner.interval.take().unwrap_or_else(|| {
             let mut interval = interval(Duration::from_millis(10));
             interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
@@ -145,11 +178,12 @@ impl NativeAudioSource {
 
             interval.tick().await;
 
+            // samples per channel = number of frames
             let samples_per_channel = data.len() / self.num_channels as usize;
             self.sys_handle.on_captured_frame(
                 data,
-                self.sample_rate as i32,
-                self.num_channels as usize,
+                self.sample_rate,
+                self.num_channels,
                 samples_per_channel,
             );
         }
diff --git a/libwebrtc/src/native/video_source.rs b/libwebrtc/src/native/video_source.rs
index 1f66ae5bc..4e55048a1 100644
--- a/libwebrtc/src/native/video_source.rs
+++ b/libwebrtc/src/native/video_source.rs
@@ -12,11 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::video_frame::{VideoBuffer, VideoFrame};
+use crate::video_frame::{I420Buffer, VideoBuffer, VideoFrame};
 use crate::video_source::VideoResolution;
 use cxx::SharedPtr;
-use std::time::{SystemTime, UNIX_EPOCH};
+use parking_lot::Mutex;
+use std::sync::Arc;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use webrtc_sys::video_frame as vf_sys;
+use webrtc_sys::video_frame::ffi::VideoRotation;
 use webrtc_sys::video_track as vt_sys;
 
 impl From<vt_sys::ffi::VideoResolution> for VideoResolution {
@@ -40,15 +43,55 @@ impl From<VideoResolution> for vt_sys::ffi::VideoResolution {
 #[derive(Clone)]
 pub struct NativeVideoSource {
     sys_handle: SharedPtr<vt_sys::ffi::VideoTrackSource>,
+    inner: Arc<Mutex<VideoSourceInner>>,
+}
+
+struct VideoSourceInner {
+    captured_frames: usize,
 }
 
 impl NativeVideoSource {
     pub fn new(resolution: VideoResolution) -> NativeVideoSource {
-        Self {
+        let source = Self {
             sys_handle: vt_sys::ffi::new_video_track_source(&vt_sys::ffi::VideoResolution::from(
-                resolution,
+                resolution.clone(),
             )),
-        }
+            inner: Arc::new(Mutex::new(VideoSourceInner { captured_frames: 0 })),
+        };
+
+        tokio::spawn({
+            let source = source.clone();
+            let i420 = I420Buffer::new(resolution.width, resolution.height);
+            async move {
+                let mut interval = tokio::time::interval(Duration::from_millis(100)); // 10 fps
+
+                loop {
+                    interval.tick().await;
+
+                    let inner = source.inner.lock();
+                    if inner.captured_frames > 0 {
+                        break;
+                    }
+
+                    let mut builder = vf_sys::ffi::new_video_frame_builder();
+                    builder
+                        .pin_mut()
+                        .set_rotation(VideoRotation::VideoRotation0);
+                    builder
+                        .pin_mut()
+                        .set_video_frame_buffer(i420.as_ref().sys_handle());
+
+                    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
+                    builder.pin_mut().set_timestamp_us(now.as_micros() as i64);
+
+                    source
+                        .sys_handle
+                        .on_captured_frame(&builder.pin_mut().build());
+                }
+            }
+        });
+
+        source
     }
 
     pub fn sys_handle(&self) -> SharedPtr<vt_sys::ffi::VideoTrackSource> {
@@ -56,6 +99,9 @@ impl NativeVideoSource {
     }
 
     pub fn capture_frame<T: AsRef<dyn VideoBuffer>>(&self, frame: &VideoFrame<T>) {
+        let mut inner = self.inner.lock();
+        inner.captured_frames += 1;
+
         let mut builder = vf_sys::ffi::new_video_frame_builder();
         builder.pin_mut().set_rotation(frame.rotation.into());
         builder
diff --git a/libwebrtc/src/video_source.rs b/libwebrtc/src/video_source.rs
index 114804089..6791e79a2 100644
--- a/libwebrtc/src/video_source.rs
+++ b/libwebrtc/src/video_source.rs
@@ -1,4 +1,4 @@
-// Copyright 2023 LiveKit, Inc.
+// Copyright 2024 LiveKit, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
diff --git a/livekit-ffi/Cargo.toml b/livekit-ffi/Cargo.toml
index ead7095d9..6ac690ccf 100644
--- a/livekit-ffi/Cargo.toml
+++ b/livekit-ffi/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "livekit-ffi"
-version = "0.3.13"
+version = "0.3.15"
 edition = "2021"
 license = "Apache-2.0"
 description = "FFI interface for bindings in other languages"
diff --git a/livekit-ffi/protocol/room.proto b/livekit-ffi/protocol/room.proto
index 7d341faca..64f61444e 100644
--- a/livekit-ffi/protocol/room.proto
+++ b/livekit-ffi/protocol/room.proto
@@ -89,6 +89,7 @@ message PublishDataRequest {
   uint64 data_len = 3;
   DataPacketKind kind = 4;
   repeated string destination_sids = 5; // destination
+  optional string topic = 6;
 }
 message PublishDataResponse {
   uint64 async_id = 1;
@@ -343,6 +344,7 @@ message DataReceived {
   OwnedBuffer data = 1;
   optional string participant_sid = 2; // Can be empty if the data is sent a server SDK
   DataPacketKind kind = 3;
+  optional string topic = 4;
 }
 
 message ConnectionStateChanged { ConnectionState state = 1; }
diff --git a/livekit-ffi/src/livekit.proto.rs b/livekit-ffi/src/livekit.proto.rs
index aca0d99e6..7f809d6bc 100644
--- a/livekit-ffi/src/livekit.proto.rs
+++ b/livekit-ffi/src/livekit.proto.rs
@@ -2157,6 +2157,8 @@ pub struct PublishDataRequest {
     /// destination
     #[prost(string, repeated, tag="5")]
     pub destination_sids: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
+    #[prost(string, optional, tag="6")]
+    pub topic: ::core::option::Option<::prost::alloc::string::String>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -2541,6 +2543,8 @@ pub struct DataReceived {
     pub participant_sid: ::core::option::Option<::prost::alloc::string::String>,
     #[prost(enumeration="DataPacketKind", tag="3")]
     pub kind: i32,
+    #[prost(string, optional, tag="4")]
+    pub topic: ::core::option::Option<::prost::alloc::string::String>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
diff --git a/livekit-ffi/src/server/audio_stream.rs b/livekit-ffi/src/server/audio_stream.rs
index 1f8bc43f0..2d07e6f94 100644
--- a/livekit-ffi/src/server/audio_stream.rs
+++ b/livekit-ffi/src/server/audio_stream.rs
@@ -43,7 +43,9 @@ impl FfiAudioStream {
         server: &'static server::FfiServer,
         new_stream: proto::NewAudioStreamRequest,
     ) -> FfiResult<proto::OwnedAudioStream> {
-        let ffi_track = server.retrieve_handle::<FfiTrack>(new_stream.track_handle)?;
+        let ffi_track = server
+            .retrieve_handle::<FfiTrack>(new_stream.track_handle)?
+            .clone();
         let rtc_track = ffi_track.track.rtc_track();
 
         let MediaStreamTrack::Audio(rtc_track) = rtc_track else {
diff --git a/livekit-ffi/src/server/requests.rs b/livekit-ffi/src/server/requests.rs
index eb54a9aeb..d9e93b2e6 100644
--- a/livekit-ffi/src/server/requests.rs
+++ b/livekit-ffi/src/server/requests.rs
@@ -122,8 +122,10 @@ fn on_set_subscribed(
         server.retrieve_handle::<FfiPublication>(set_subscribed.publication_handle)?;
 
     let TrackPublication::Remote(publication) = &ffi_publication.publication else {
-            return Err(FfiError::InvalidRequest("publication is not a RemotePublication".into()));
-        };
+        return Err(FfiError::InvalidRequest(
+            "publication is not a RemotePublication".into(),
+        ));
+    };
 
     let _guard = server.async_runtime.enter();
     publication.set_subscribed(set_subscribed.subscribe);
@@ -457,7 +459,7 @@ unsafe fn on_to_argb(
             let Some(proto::video_frame_buffer_info::Buffer::Yuv(yuv)) = &buffer.buffer else {
                 return Err(FfiError::InvalidRequest(
                     "invalid i420 buffer description".into(),
-                ))
+                ));
             };
 
             #[rustfmt::skip]
@@ -470,19 +472,19 @@ unsafe fn on_to_argb(
             match rgba_format {
                 proto::VideoFormatType::FormatArgb => {
                     #[rustfmt::skip]
-                    yuv_helper::i420_to_argb(src_y, yuv.stride_y, src_u, yuv.stride_u, src_v, yuv.stride_v, argb, stride, w, h);
+                    yuv_helper::i420_to_bgra(src_y, yuv.stride_y, src_u, yuv.stride_u, src_v, yuv.stride_v, argb, stride, w, h);
                 }
                 proto::VideoFormatType::FormatBgra => {
                     #[rustfmt::skip]
-                    yuv_helper::i420_to_bgra(src_y, yuv.stride_y, src_u, yuv.stride_u, src_v, yuv.stride_v, argb, stride, w, h);
+                    yuv_helper::i420_to_argb(src_y, yuv.stride_y, src_u, yuv.stride_u, src_v, yuv.stride_v, argb, stride, w, h);
                 }
                 proto::VideoFormatType::FormatRgba => {
                     #[rustfmt::skip]
-                    yuv_helper::i420_to_rgba(src_y, yuv.stride_y, src_u, yuv.stride_u, src_v, yuv.stride_v, argb, stride, w, h);
+                    yuv_helper::i420_to_abgr(src_y, yuv.stride_y, src_u, yuv.stride_u, src_v, yuv.stride_v, argb, stride, w, h);
                 }
                 proto::VideoFormatType::FormatAbgr => {
                     #[rustfmt::skip]
-                    yuv_helper::i420_to_abgr(src_y, yuv.stride_y, src_u, yuv.stride_u, src_v, yuv.stride_v, argb, stride, w, h);
+                    yuv_helper::i420_to_rgba(src_y, yuv.stride_y, src_u, yuv.stride_u, src_v, yuv.stride_v, argb, stride, w, h);
                 }
             }
         }
diff --git a/livekit-ffi/src/server/room.rs b/livekit-ffi/src/server/room.rs
index 123b2c467..17d481824 100644
--- a/livekit-ffi/src/server/room.rs
+++ b/livekit-ffi/src/server/room.rs
@@ -59,7 +59,7 @@ pub struct FfiRoom {
 pub struct RoomInner {
     pub room: Room,
     handle_id: FfiHandleId,
-    data_tx: mpsc::UnboundedSender<DataPacket>,
+    data_tx: mpsc::UnboundedSender<FfiDataPacket>,
 
     // local tracks just published, it is used to synchronize the publish events:
     // - make sure LocalTrackPublised is sent *after* the PublishTrack callback)
@@ -74,10 +74,8 @@ struct Handle {
     close_tx: broadcast::Sender<()>,
 }
 
-struct DataPacket {
-    data: Vec<u8>,
-    kind: DataPacketKind,
-    destination_sids: Vec<String>,
+struct FfiDataPacket {
+    payload: DataPacket,
     async_id: u64,
 }
 
@@ -99,9 +97,12 @@ impl FfiRoom {
                 Ok((room, mut events)) => {
                     // Successfully connected to the room
                     // Forward the initial state for the FfiClient
-                    let Some(RoomEvent::Connected { participants_with_tracks}) = events.recv().await else {
-                            unreachable!("Connected event should always be the first event");
-                        };
+                    let Some(RoomEvent::Connected {
+                        participants_with_tracks,
+                    }) = events.recv().await
+                    else {
+                        unreachable!("Connected event should always be the first event");
+                    };
 
                     let (data_tx, data_rx) = mpsc::unbounded_channel();
                     let (close_tx, close_rx) = broadcast::channel(1);
@@ -201,14 +202,20 @@ impl RoomInner {
             slice::from_raw_parts(publish.data_ptr as *const u8, publish.data_len as usize)
         };
         let kind = publish.kind();
-        let destination_sids: Vec<String> = publish.destination_sids;
+        let destination_sids = publish.destination_sids;
         let async_id = server.next_id();
 
         self.data_tx
-            .send(DataPacket {
-                data: data.to_vec(), // Avoid copy?
-                kind: kind.into(),
-                destination_sids,
+            .send(FfiDataPacket {
+                payload: DataPacket {
+                    payload: data.to_vec(), // Avoid copy?
+                    kind: kind.into(),
+                    topic: publish.topic,
+                    destination_sids: destination_sids
+                        .into_iter()
+                        .map(|str| str.try_into().unwrap())
+                        .collect(),
+                },
                 async_id,
             })
             .map_err(|_| FfiError::InvalidRequest("failed to send data packet".into()))?;
@@ -384,17 +391,13 @@ impl RoomInner {
 async fn data_task(
     server: &'static FfiServer,
     inner: Arc<RoomInner>,
-    mut data_rx: mpsc::UnboundedReceiver<DataPacket>,
+    mut data_rx: mpsc::UnboundedReceiver<FfiDataPacket>,
     mut close_rx: broadcast::Receiver<()>,
 ) {
     loop {
         tokio::select! {
             Some(event) = data_rx.recv() => {
-                let res = inner.room.local_participant().publish_data(
-                    event.data,
-                    event.kind,
-                    event.destination_sids,
-                ).await;
+                let res = inner.room.local_participant().publish_data(event.payload).await;
 
                 let cb = proto::PublishDataCallback {
                     async_id: event.async_id,
@@ -727,6 +730,7 @@ async fn forward_event(
             payload,
             kind,
             participant,
+            topic,
         } => {
             let handle_id = server.next_id();
             let buffer_info = proto::BufferInfo {
@@ -749,6 +753,7 @@ async fn forward_event(
                     }),
                     participant_sid: participant.map(|p| p.sid().to_string()),
                     kind: proto::DataPacketKind::from(kind).into(),
+                    topic,
                 },
             ))
             .await;
diff --git a/livekit-ffi/src/server/video_stream.rs b/livekit-ffi/src/server/video_stream.rs
index 2bde3aeb9..8a4c3b4e6 100644
--- a/livekit-ffi/src/server/video_stream.rs
+++ b/livekit-ffi/src/server/video_stream.rs
@@ -43,7 +43,9 @@ impl FfiVideoStream {
         server: &'static server::FfiServer,
         new_stream: proto::NewVideoStreamRequest,
     ) -> FfiResult<proto::OwnedVideoStream> {
-        let ffi_track = server.retrieve_handle::<FfiTrack>(new_stream.track_handle)?;
+        let ffi_track = server
+            .retrieve_handle::<FfiTrack>(new_stream.track_handle)?
+            .clone();
         let rtc_track = ffi_track.track.rtc_track();
 
         let MediaStreamTrack::Video(rtc_track) = rtc_track else {
diff --git a/livekit/src/prelude.rs b/livekit/src/prelude.rs
index 96ab1c67b..0668b9de8 100644
--- a/livekit/src/prelude.rs
+++ b/livekit/src/prelude.rs
@@ -15,7 +15,8 @@
 pub use crate::participant::{ConnectionQuality, LocalParticipant, Participant, RemoteParticipant};
 
 pub use crate::{
-    ConnectionState, DataPacketKind, Room, RoomError, RoomEvent, RoomOptions, RoomResult,
+    ConnectionState, DataPacket, DataPacketKind, Room, RoomError, RoomEvent, RoomOptions,
+    RoomResult,
 };
 
 pub use crate::publication::{LocalTrackPublication, RemoteTrackPublication, TrackPublication};
diff --git a/livekit/src/room/mod.rs b/livekit/src/room/mod.rs
index 7e642c5c0..a120de502 100644
--- a/livekit/src/room/mod.rs
+++ b/livekit/src/room/mod.rs
@@ -132,6 +132,7 @@ pub enum RoomEvent {
     },
     DataReceived {
         payload: Arc<Vec<u8>>,
+        topic: Option<String>,
         kind: DataPacketKind,
         participant: Option<RemoteParticipant>,
     },
@@ -165,6 +166,25 @@ pub enum DataPacketKind {
     Reliable,
 }
 
+#[derive(Debug, Clone)]
+pub struct DataPacket {
+    pub payload: Vec<u8>,
+    pub topic: Option<String>,
+    pub kind: DataPacketKind,
+    pub destination_sids: Vec<ParticipantSid>,
+}
+
+impl Default for DataPacket {
+    fn default() -> Self {
+        Self {
+            payload: Vec::new(),
+            topic: None,
+            kind: DataPacketKind::Reliable,
+            destination_sids: Vec::new(),
+        }
+    }
+}
+
 #[derive(Clone)]
 pub struct RoomOptions {
     pub auto_subscribe: bool,
@@ -427,6 +447,10 @@ impl Room {
         }
     }
 
+    pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> {
+        self.inner.rtc_engine.simulate_scenario(scenario).await
+    }
+
     pub fn subscribe(&self) -> mpsc::UnboundedReceiver<RoomEvent> {
         self.inner.dispatcher.register()
     }
@@ -455,10 +479,6 @@ impl Room {
         self.inner.participants.read().0.clone()
     }
 
-    pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> {
-        self.inner.rtc_engine.simulate_scenario(scenario).await
-    }
-
     pub fn e2ee_manager(&self) -> &E2eeManager {
         &self.inner.e2ee_manager
     }
@@ -557,10 +577,11 @@ impl RoomSession {
             EngineEvent::Disconnected { reason } => self.handle_disconnected(reason),
             EngineEvent::Data {
                 payload,
+                topic,
                 kind,
                 participant_sid,
             } => {
-                self.handle_data(payload, kind, participant_sid);
+                self.handle_data(payload, topic, kind, participant_sid);
             }
             EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers),
             EngineEvent::ConnectionQuality { updates } => {
@@ -970,6 +991,7 @@ impl RoomSession {
     fn handle_data(
         &self,
         payload: Vec<u8>,
+        topic: Option<String>,
         kind: DataPacketKind,
         participant_sid: Option<ParticipantSid>,
     ) {
@@ -985,6 +1007,7 @@ impl RoomSession {
 
         self.dispatcher.dispatch(&RoomEvent::DataReceived {
             payload: Arc::new(payload),
+            topic,
             kind,
             participant,
         });
diff --git a/livekit/src/room/participant/local_participant.rs b/livekit/src/room/participant/local_participant.rs
index 6595f04b3..2c8bff4d0 100644
--- a/livekit/src/room/participant/local_participant.rs
+++ b/livekit/src/room/participant/local_participant.rs
@@ -21,6 +21,7 @@ use crate::options::video_layers_from_encodings;
 use crate::options::TrackPublishOptions;
 use crate::prelude::*;
 use crate::rtc_engine::RtcEngine;
+use crate::DataPacket;
 use crate::DataPacketKind;
 use libwebrtc::rtp_parameters::RtpEncodingParameters;
 use livekit_protocol as proto;
@@ -28,6 +29,8 @@ use parking_lot::Mutex;
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::sync::Arc;
+use std::time::Duration;
+use tokio::time::sleep;
 
 type LocalTrackPublishedHandler = Box<dyn Fn(LocalParticipant, LocalTrackPublication) + Send>;
 type LocalTrackUnpublishedHandler = Box<dyn Fn(LocalParticipant, LocalTrackPublication) + Send>;
@@ -226,7 +229,6 @@ impl LocalParticipant {
         {
             local_track_published(self.clone(), publication.clone());
         }
-
         track.enable();
 
         Ok(publication)
@@ -286,24 +288,24 @@ impl LocalParticipant {
         }
     }
 
-    pub async fn publish_data(
-        &self,
-        data: Vec<u8>,
-        kind: DataPacketKind,
-        destination_sids: Vec<String>,
-    ) -> RoomResult<()> {
+    pub async fn publish_data(&self, packet: DataPacket) -> RoomResult<()> {
         let data = proto::DataPacket {
-            kind: kind as i32,
+            kind: DataPacketKind::from(packet.kind) as i32,
             value: Some(proto::data_packet::Value::User(proto::UserPacket {
-                payload: data,
-                destination_sids: destination_sids.to_owned(),
+                payload: packet.payload,
+                topic: packet.topic,
+                destination_sids: packet
+                    .destination_sids
+                    .into_iter()
+                    .map(Into::into)
+                    .collect(),
                 ..Default::default()
             })),
         };
 
         self.inner
             .rtc_engine
-            .publish_data(&data, kind)
+            .publish_data(&data, packet.kind)
             .await
             .map_err(Into::into)
     }
diff --git a/livekit/src/rtc_engine/mod.rs b/livekit/src/rtc_engine/mod.rs
index d3341c359..d48a39fe3 100644
--- a/livekit/src/rtc_engine/mod.rs
+++ b/livekit/src/rtc_engine/mod.rs
@@ -90,6 +90,7 @@ pub enum EngineEvent {
     Data {
         participant_sid: Option<ParticipantSid>,
         payload: Vec<u8>,
+        topic: Option<String>,
         kind: DataPacketKind,
     },
     SpeakersChanged {
@@ -383,11 +384,13 @@ impl EngineInner {
             SessionEvent::Data {
                 participant_sid,
                 payload,
+                topic,
                 kind,
             } => {
                 let _ = self.engine_tx.send(EngineEvent::Data {
                     participant_sid,
                     payload,
+                    topic,
                     kind,
                 });
             }
diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs
index 49cec8cf6..f45ed028c 100644
--- a/livekit/src/rtc_engine/rtc_session.rs
+++ b/livekit/src/rtc_engine/rtc_session.rs
@@ -59,6 +59,7 @@ pub enum SessionEvent {
         // None when the data comes from the ServerSDK (So no real participant)
         participant_sid: Option<ParticipantSid>,
         payload: Vec<u8>,
+        topic: Option<String>,
         kind: DataPacketKind,
     },
     MediaTrack {
@@ -573,6 +574,7 @@ impl SessionInner {
                             kind: data.kind().into(),
                             participant_sid: participant_sid.map(|s| s.try_into().unwrap()),
                             payload: user.payload.clone(),
+                            topic: user.topic.clone(),
                         });
                     }
                     proto::data_packet::Value::Speaker(_) => {}
diff --git a/webrtc-sys/include/livekit/audio_track.h b/webrtc-sys/include/livekit/audio_track.h
index 0d08de653..a67ca8116 100644
--- a/webrtc-sys/include/livekit/audio_track.h
+++ b/webrtc-sys/include/livekit/audio_track.h
@@ -94,8 +94,8 @@ class AudioTrackSource {
     // AudioFrame should always contain 10 ms worth of data (see index.md of
     // acm)
     void on_captured_frame(rust::Slice<const int16_t> audio_data,
-                           int sample_rate,
-                           size_t number_of_channels,
+                           uint32_t sample_rate,
+                           uint32_t number_of_channels,
                            size_t number_of_frames);
 
    private:
@@ -112,8 +112,8 @@ class AudioTrackSource {
   void set_audio_options(const AudioSourceOptions& options) const;
 
   void on_captured_frame(rust::Slice<const int16_t> audio_data,
-                         int sample_rate,
-                         size_t number_of_channels,
+                         uint32_t sample_rate,
+                         uint32_t number_of_channels,
                          size_t number_of_frames) const;
 
   rtc::scoped_refptr<InternalSource> get() const;
diff --git a/webrtc-sys/src/audio_track.cpp b/webrtc-sys/src/audio_track.cpp
index eff92a483..951208d24 100644
--- a/webrtc-sys/src/audio_track.cpp
+++ b/webrtc-sys/src/audio_track.cpp
@@ -131,8 +131,8 @@ void AudioTrackSource::InternalSource::RemoveSink(
 
 void AudioTrackSource::InternalSource::on_captured_frame(
     rust::Slice<const int16_t> data,
-    int sample_rate,
-    size_t number_of_channels,
+    uint32_t sample_rate,
+    uint32_t number_of_channels,
     size_t number_of_frames) {
   webrtc::MutexLock lock(&mutex_);
   for (auto sink : sinks_) {
@@ -156,8 +156,8 @@ void AudioTrackSource::set_audio_options(
 }
 
 void AudioTrackSource::on_captured_frame(rust::Slice<const int16_t> audio_data,
-                                         int sample_rate,
-                                         size_t number_of_channels,
+                                         uint32_t sample_rate,
+                                         uint32_t number_of_channels,
                                          size_t number_of_frames) const {
   source_->on_captured_frame(audio_data, sample_rate, number_of_channels,
                              number_of_frames);
diff --git a/webrtc-sys/src/audio_track.rs b/webrtc-sys/src/audio_track.rs
index dd92dd4b4..c3c35cd43 100644
--- a/webrtc-sys/src/audio_track.rs
+++ b/webrtc-sys/src/audio_track.rs
@@ -44,8 +44,8 @@ pub mod ffi {
         fn on_captured_frame(
             self: &AudioTrackSource,
             data: &[i16],
-            sample_rate: i32,
-            nb_channels: usize,
+            sample_rate: u32,
+            nb_channels: u32,
             nb_frames: usize,
         );
         fn audio_options(self: &AudioTrackSource) -> AudioSourceOptions;