Skip to content

Do not modify raw packets #714

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ladvoc merged 3 commits into from
Sep 25, 2025
Merged

Do not modify raw packets #714

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ec6d743
Do not modify raw packets
ladvoc Sep 23, 2025
7fe485e
Merge remote-tracking branch 'origin/main' into ladvoc/set-iden
ladvoc Sep 25, 2025
9f0c128
Format
ladvoc Sep 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion livekit/src/room/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1730,7 +1730,7 @@ async fn outgoing_data_stream_task(
loop {
tokio::select! {
Ok((packet, responder)) = packet_rx.recv() => {
let result = engine.publish_data(packet, DataPacketKind::Reliable).await;
let result = engine.publish_data(packet, DataPacketKind::Reliable, false).await;
let _ = responder.respond(result);
},
_ = close_rx.recv() => {
Expand Down
38 changes: 29 additions & 9 deletions livekit/src/room/participant/local_participant.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -407,7 +407,7 @@ impl LocalParticipant {
..Default::default()
};

match self.inner.rtc_engine.publish_data(data, DataPacketKind::Reliable).await {
match self.inner.rtc_engine.publish_data(data, DataPacketKind::Reliable, false).await {
Ok(_) => Ok(ChatMessage::from(chat_message)),
Err(e) => Err(Into::into(e)),
}
Expand All @@ -433,7 +433,7 @@ impl LocalParticipant {
..Default::default()
};

match self.inner.rtc_engine.publish_data(data, DataPacketKind::Reliable).await {
match self.inner.rtc_engine.publish_data(data, DataPacketKind::Reliable, false).await {
Ok(_) => Ok(ChatMessage::from(proto_msg)),
Err(e) => Err(Into::into(e)),
}
Expand Down Expand Up @@ -477,7 +477,7 @@ impl LocalParticipant {
true => DataPacketKind::Reliable,
false => DataPacketKind::Lossy,
};
self.inner.rtc_engine.publish_data(packet, kind).await.map_err(Into::into)
self.inner.rtc_engine.publish_data(packet, kind, true).await.map_err(Into::into)
}

pub async fn publish_data(&self, packet: DataPacket) -> RoomResult<()> {
Expand All @@ -498,7 +498,7 @@ impl LocalParticipant {
..Default::default()
};

self.inner.rtc_engine.publish_data(data, kind).await.map_err(Into::into)
self.inner.rtc_engine.publish_data(data, kind, false).await.map_err(Into::into)
}

pub fn set_data_channel_buffered_amount_low_threshold(
Expand Down Expand Up @@ -553,7 +553,11 @@ impl LocalParticipant {
value: Some(proto::data_packet::Value::Transcription(transcription_packet)),
..Default::default()
};
self.inner.rtc_engine.publish_data(data, DataPacketKind::Reliable).await.map_err(Into::into)
self.inner
.rtc_engine
.publish_data(data, DataPacketKind::Reliable, false)
.await
.map_err(Into::into)
}

pub async fn publish_dtmf(&self, dtmf: SipDTMF) -> RoomResult<()> {
Expand All @@ -567,7 +571,11 @@ impl LocalParticipant {
..Default::default()
};

self.inner.rtc_engine.publish_data(data, DataPacketKind::Reliable).await.map_err(Into::into)
self.inner
.rtc_engine
.publish_data(data, DataPacketKind::Reliable, false)
.await
.map_err(Into::into)
}

async fn publish_rpc_request(&self, rpc_request: RpcRequest) -> RoomResult<()> {
Expand All @@ -587,7 +595,11 @@ impl LocalParticipant {
..Default::default()
};

self.inner.rtc_engine.publish_data(data, DataPacketKind::Reliable).await.map_err(Into::into)
self.inner
.rtc_engine
.publish_data(data, DataPacketKind::Reliable, false)
.await
.map_err(Into::into)
}

async fn publish_rpc_response(&self, rpc_response: RpcResponse) -> RoomResult<()> {
Expand All @@ -611,7 +623,11 @@ impl LocalParticipant {
..Default::default()
};

self.inner.rtc_engine.publish_data(data, DataPacketKind::Reliable).await.map_err(Into::into)
self.inner
.rtc_engine
.publish_data(data, DataPacketKind::Reliable, false)
.await
.map_err(Into::into)
}

async fn publish_rpc_ack(&self, rpc_ack: RpcAck) -> RoomResult<()> {
Expand All @@ -625,7 +641,11 @@ impl LocalParticipant {
..Default::default()
};

self.inner.rtc_engine.publish_data(data, DataPacketKind::Reliable).await.map_err(Into::into)
self.inner
.rtc_engine
.publish_data(data, DataPacketKind::Reliable, false)
.await
.map_err(Into::into)
}

pub(crate) async fn update_track_subscription_permissions(&self) {
Expand Down
4 changes: 2 additions & 2 deletions livekit/src/rtc_engine/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,13 +246,13 @@ impl RtcEngine {
&self,
data: proto::DataPacket,
kind: DataPacketKind,
is_raw_packet: bool,
) -> EngineResult<()> {
let (session, _r_lock) = {
let (handle, _r_lock) = self.inner.wait_reconnection().await?;
(handle.session.clone(), _r_lock)
};

session.publish_data(data, kind).await
session.publish_data(data, kind, is_raw_packet).await
}

pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> {
Expand Down
12 changes: 8 additions & 4 deletions livekit/src/rtc_engine/rtc_session.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,7 +213,7 @@ enum DataChannelEventDetail {

#[derive(Debug)]
struct PublishPacketRequest {
/// Unencoded data packewt.
/// Unencoded data packet.
packet: proto::DataPacket,

/// Notifies the caller once the request has been fulfilled.
Expand Down Expand Up @@ -488,8 +488,9 @@ impl RtcSession {
&self,
data: proto::DataPacket,
kind: DataPacketKind,
is_raw_packet: bool,
) -> Result<(), EngineError> {
self.inner.publish_data(data, kind).await
self.inner.publish_data(data, kind, is_raw_packet).await
}

pub async fn restart(&self) -> EngineResult<proto::ReconnectResponse> {
Expand Down Expand Up @@ -1339,12 +1340,15 @@ impl SessionInner {
self: &Arc<Self>,
mut packet: proto::DataPacket,
kind: DataPacketKind,
is_raw_packet: bool,
) -> Result<(), EngineError> {
self.ensure_publisher_connected(kind).await?;

// Populate local participant info fields
packet.participant_identity = self.participant_info.identity.to_string();
packet.participant_sid = self.participant_info.sid.to_string();
if !is_raw_packet {
packet.participant_identity = self.participant_info.identity.to_string();
packet.participant_sid = self.participant_info.sid.to_string();
}

let (completion_tx, completion_rx) = oneshot::channel();
let ev = DataChannelEvent {
Expand Down
Loading