livekit_data_stream/outgoing/mod.rs
1// Copyright 2026 LiveKit, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Outgoing data streams: the [`manager::Manager`] and the writers it produces.
16
17use livekit_common::ParticipantIdentity;
18use std::{collections::HashMap, fmt, sync::Arc};
19
20use crate::{
21 types::{OperationType, StreamId, TextHeader},
22 utils::StreamProgress,
23};
24
25pub(crate) mod manager;
26
27mod constants;
28mod raw_stream;
29mod stream_writer;
30
31pub use stream_writer::{ByteStreamWriter, StreamWriter, TextStreamWriter};
32
33/// Receives outgoing send progress on the thread polling the send; callbacks should return quickly.
34///
35/// Clones share the callback, and equality compares callback identity rather than behavior.
36/// `Debug` prints the placeholder `StreamProgressCallback` without inspecting the closure.
37#[derive(Clone)]
38pub struct StreamProgressCallback(Arc<dyn Fn(StreamProgress) + Send + Sync>);
39
40impl StreamProgressCallback {
41 /// Wraps a callback that receives progress before compression.
42 pub fn new(callback: impl Fn(StreamProgress) + Send + Sync + 'static) -> Self {
43 Self(Arc::new(callback))
44 }
45
46 /// Invokes the callback with the current progress.
47 pub fn call(&self, progress: StreamProgress) {
48 (self.0)(progress);
49 }
50}
51
52impl fmt::Debug for StreamProgressCallback {
53 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
54 f.write_str("StreamProgressCallback")
55 }
56}
57
58impl PartialEq for StreamProgressCallback {
59 fn eq(&self, other: &Self) -> bool {
60 Arc::ptr_eq(&self.0, &other.0)
61 }
62}
63
64impl Eq for StreamProgressCallback {}
65
66macro_rules! common_setters {
67 () => {
68 /// Sets the topic the stream is published to, accepting any value convertible to `String`.
69 pub fn with_topic(mut self, topic: impl Into<String>) -> Self {
70 self.topic = topic.into();
71 self
72 }
73 /// Replaces all attributes attached to the stream.
74 pub fn with_attributes(mut self, attributes: HashMap<String, String>) -> Self {
75 self.attributes = attributes;
76 self
77 }
78 /// Adds a single attribute to the stream, overwriting any existing value for `key`.
79 pub fn with_attribute(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
80 self.attributes.insert(key.into(), value.into());
81 self
82 }
83 /// Replaces the set of participant identities the stream is delivered to.
84 /// An empty list delivers to all participants in the room.
85 pub fn with_destination_identities(
86 mut self,
87 destination_identities: Vec<ParticipantIdentity>,
88 ) -> Self {
89 self.destination_identities = destination_identities;
90 self
91 }
92 /// Adds a single participant identity to the stream's destinations.
93 pub fn with_destination_identity(
94 mut self,
95 identity: impl Into<ParticipantIdentity>,
96 ) -> Self {
97 self.destination_identities.push(identity.into());
98 self
99 }
100 /// Sets an explicit stream id. If unset, a new uuid is generated per call.
101 pub fn with_id(mut self, id: impl Into<String>) -> Self {
102 self.id = Some(id.into());
103 self
104 }
105 /// Sets the identity the stream's packets are attributed to. Only participants with the
106 /// appropriate permission (e.g. agents) may impersonate another identity.
107 pub fn with_sender_identity(mut self, identity: impl Into<ParticipantIdentity>) -> Self {
108 self.sender_identity = Some(identity.into());
109 self
110 }
111 };
112}
113
114macro_rules! text_setters {
115 () => {
116 pub(crate) fn text_header(&self) -> TextHeader {
117 TextHeader {
118 operation_type: self.operation_type.unwrap_or_default(),
119 version: self.version.unwrap_or_default(),
120 reply_to_stream_id: self.reply_to_stream_id.clone().map(StreamId::from),
121 attached_stream_ids: self
122 .attached_stream_ids
123 .clone()
124 .into_iter()
125 .map(StreamId::from)
126 .collect(),
127 generated: self.generated.unwrap_or_default(),
128 }
129 }
130
131 /// Sets the operation this text stream represents (e.g. create or update).
132 pub fn with_operation_type(mut self, operation_type: OperationType) -> Self {
133 self.operation_type = Some(operation_type);
134 self
135 }
136 /// Sets the version of the text, used to order updates to the same content.
137 pub fn with_version(mut self, version: i32) -> Self {
138 self.version = Some(version);
139 self
140 }
141 /// Sets the id of the stream this text is a reply to.
142 pub fn with_reply_to_stream_id(mut self, reply_to_stream_id: impl Into<String>) -> Self {
143 self.reply_to_stream_id = Some(reply_to_stream_id.into());
144 self
145 }
146 /// Replaces the set of stream ids attached to this text (e.g. referenced files).
147 pub fn with_attached_stream_ids(mut self, attached_stream_ids: Vec<String>) -> Self {
148 self.attached_stream_ids = attached_stream_ids;
149 self
150 }
151 /// Adds a single attached stream id to this text.
152 pub fn with_attached_stream_id(mut self, attached_stream_id: impl Into<String>) -> Self {
153 self.attached_stream_ids.push(attached_stream_id.into());
154 self
155 }
156 /// Sets whether the text was machine-generated (e.g. by an agent).
157 pub fn with_generated(mut self, generated: bool) -> Self {
158 self.generated = Some(generated);
159 self
160 }
161 };
162}
163
164macro_rules! byte_setters {
165 () => {
166 /// Sets the MIME type describing the stream's payload.
167 pub fn with_mime_type(mut self, mime_type: impl Into<String>) -> Self {
168 self.mime_type = Some(mime_type.into());
169 self
170 }
171 /// Sets a human-readable name for the stream (e.g. a file name).
172 pub fn with_name(mut self, name: impl Into<String>) -> Self {
173 self.name = Some(name.into());
174 self
175 }
176 };
177}
178
179macro_rules! send_setters {
180 () => {
181 /// Sets whether to deflate-raw compress when all recipients support it.
182 /// Defaults to `true` (compression opt-out).
183 pub fn with_compress(mut self, compress: bool) -> Self {
184 self.compress = Some(compress);
185 self
186 }
187 /// Sets the progress callback; see [`Self::on_progress`] for reporting semantics.
188 pub fn with_on_progress(
189 mut self,
190 callback: impl Fn(StreamProgress) + Send + Sync + 'static,
191 ) -> Self {
192 self.on_progress = Some(StreamProgressCallback::new(callback));
193 self
194 }
195 };
196}
197
198macro_rules! stream_setters {
199 () => {
200 /// Sets the total byte length of the payload, when known ahead of time.
201 pub fn with_total_length(mut self, total_length: u64) -> Self {
202 self.total_length = Some(total_length);
203 self
204 }
205 };
206}
207
208/// Options for [`send_text`](manager::Manager::send_text). Progress is reported once after the complete send is acknowledged.
209#[non_exhaustive]
210#[derive(Clone, Debug, Eq, PartialEq)]
211pub struct SendTextOptions {
212 /// Topic the stream is published to.
213 pub topic: String,
214 /// Attributes attached to the stream.
215 pub attributes: HashMap<String, String>,
216 /// Recipients; an empty list delivers to all participants.
217 pub destination_identities: Vec<ParticipantIdentity>,
218 /// Explicit stream id, or a new UUID per call when unspecified.
219 pub id: Option<String>,
220 /// Operation this text stream represents.
221 pub operation_type: Option<OperationType>,
222 /// Version used to order updates to the same content.
223 pub version: Option<i32>,
224 /// Id of the stream this text replies to.
225 pub reply_to_stream_id: Option<String>,
226 /// Ids of streams attached to this text.
227 pub attached_stream_ids: Vec<String>,
228 /// Whether the text was machine-generated.
229 pub generated: Option<bool>,
230 /// Identity attributed to the packets, or the sending participant when unspecified.
231 /// Impersonating another identity requires the appropriate permission (e.g. agents).
232 pub sender_identity: Option<ParticipantIdentity>,
233 /// Whether to deflate-raw compress when all recipients support it. Defaults to true.
234 pub compress: Option<bool>,
235 /// Reports payload byte length once after the complete send is acknowledged.
236 pub on_progress: Option<StreamProgressCallback>,
237}
238
239impl SendTextOptions {
240 /// Creates options for the given topic. Use the `with_*` setters to customize them.
241 pub fn new_with_topic(topic: impl Into<String>) -> Self {
242 Self {
243 topic: topic.into(),
244 attributes: HashMap::new(),
245 destination_identities: vec![],
246 id: None,
247 operation_type: None,
248 version: None,
249 reply_to_stream_id: None,
250 attached_stream_ids: vec![],
251 generated: None,
252 sender_identity: None,
253 compress: None,
254 on_progress: None,
255 }
256 }
257
258 common_setters!();
259 text_setters!();
260 send_setters!();
261}
262
263/// Options for [`send_bytes`](manager::Manager::send_bytes) and [`send_file`](manager::Manager::send_file).
264///
265/// See [`Self::on_progress`] for progress reporting semantics.
266#[non_exhaustive]
267#[derive(Clone, Debug, Eq, PartialEq)]
268pub struct SendByteOptions {
269 /// Topic the stream is published to.
270 pub topic: String,
271 /// Attributes attached to the stream.
272 pub attributes: HashMap<String, String>,
273 /// Recipients; an empty list delivers to all participants.
274 pub destination_identities: Vec<ParticipantIdentity>,
275 /// Explicit stream id, or a new UUID per call when unspecified.
276 pub id: Option<String>,
277 /// MIME type describing the payload.
278 pub mime_type: Option<String>,
279 /// Human-readable name; `send_file` defaults to the file name.
280 pub name: Option<String>,
281 /// Identity attributed to the packets, or the sending participant when unspecified.
282 /// Impersonating another identity requires the appropriate permission (e.g. agents).
283 pub sender_identity: Option<ParticipantIdentity>,
284 /// Whether to deflate-raw compress when all recipients support it. Defaults to true.
285 pub compress: Option<bool>,
286 /// Byte sends report once after the complete send is acknowledged. Files report input bytes
287 /// before compression, against the file size, after each chunk is sent or each encoder drain.
288 /// Compressed progress can run ahead of chunks sent; the whole file is reported only after
289 /// its last chunk was sent.
290 pub on_progress: Option<StreamProgressCallback>,
291}
292
293impl SendByteOptions {
294 /// Creates options for the given topic. Use the `with_*` setters to customize them.
295 pub fn new_with_topic(topic: impl Into<String>) -> Self {
296 Self {
297 topic: topic.into(),
298 attributes: HashMap::new(),
299 destination_identities: vec![],
300 id: None,
301 mime_type: None,
302 name: None,
303 sender_identity: None,
304 compress: None,
305 on_progress: None,
306 }
307 }
308
309 common_setters!();
310 byte_setters!();
311 send_setters!();
312}
313
314/// Options for [`stream_text`](manager::Manager::stream_text), which never compresses or inlines content.
315#[non_exhaustive]
316#[derive(Clone, Debug, Eq, PartialEq)]
317pub struct StreamTextOptions {
318 /// Topic the stream is published to.
319 pub topic: String,
320 /// Attributes attached to the stream.
321 pub attributes: HashMap<String, String>,
322 /// Recipients; an empty list delivers to all participants.
323 pub destination_identities: Vec<ParticipantIdentity>,
324 /// Explicit stream id, or a new UUID per call when unspecified.
325 pub id: Option<String>,
326 /// Operation this text stream represents.
327 pub operation_type: Option<OperationType>,
328 /// Version used to order updates to the same content.
329 pub version: Option<i32>,
330 /// Id of the stream this text replies to.
331 pub reply_to_stream_id: Option<String>,
332 /// Ids of streams attached to this text.
333 pub attached_stream_ids: Vec<String>,
334 /// Whether the text was machine-generated.
335 pub generated: Option<bool>,
336 /// Identity attributed to the packets, or the sending participant when unspecified.
337 /// Impersonating another identity requires the appropriate permission (e.g. agents).
338 pub sender_identity: Option<ParticipantIdentity>,
339 /// Declared total byte length, when known. The receiver fails the stream on a mismatch.
340 pub total_length: Option<u64>,
341}
342
343impl StreamTextOptions {
344 /// Creates options for the given topic. Use the `with_*` setters to customize them.
345 pub fn new_with_topic(topic: impl Into<String>) -> Self {
346 Self {
347 topic: topic.into(),
348 attributes: HashMap::new(),
349 destination_identities: vec![],
350 id: None,
351 operation_type: None,
352 version: None,
353 reply_to_stream_id: None,
354 attached_stream_ids: vec![],
355 generated: None,
356 sender_identity: None,
357 total_length: None,
358 }
359 }
360
361 common_setters!();
362 text_setters!();
363 stream_setters!();
364}
365
366/// Options for [`stream_bytes`](manager::Manager::stream_bytes), which never compresses or inlines content.
367#[non_exhaustive]
368#[derive(Clone, Debug, Eq, PartialEq)]
369pub struct StreamByteOptions {
370 /// Topic the stream is published to.
371 pub topic: String,
372 /// Attributes attached to the stream.
373 pub attributes: HashMap<String, String>,
374 /// Recipients; an empty list delivers to all participants.
375 pub destination_identities: Vec<ParticipantIdentity>,
376 /// Explicit stream id, or a new UUID per call when unspecified.
377 pub id: Option<String>,
378 /// MIME type describing the payload.
379 pub mime_type: Option<String>,
380 /// Human-readable name for the stream.
381 pub name: Option<String>,
382 /// Identity attributed to the packets, or the sending participant when unspecified.
383 /// Impersonating another identity requires the appropriate permission (e.g. agents).
384 pub sender_identity: Option<ParticipantIdentity>,
385 /// Declared total byte length, when known. The receiver fails the stream on a mismatch.
386 pub total_length: Option<u64>,
387}
388
389impl StreamByteOptions {
390 /// Creates options for the given topic. Use the `with_*` setters to customize them.
391 pub fn new_with_topic(topic: impl Into<String>) -> Self {
392 Self {
393 topic: topic.into(),
394 attributes: HashMap::new(),
395 destination_identities: vec![],
396 id: None,
397 mime_type: None,
398 name: None,
399 sender_identity: None,
400 total_length: None,
401 }
402 }
403
404 common_setters!();
405 byte_setters!();
406 stream_setters!();
407}