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;
19
20use crate::types::OperationType;
21
22pub(crate) mod manager;
23
24mod constants;
25mod raw_stream;
26mod stream_writer;
27
28pub use stream_writer::{ByteStreamWriter, StreamWriter, TextStreamWriter};
29
30/// Options used when opening an outgoing byte data stream.
31#[derive(Clone, Debug, Eq, PartialEq)]
32pub struct StreamByteOptions {
33    pub topic: String,
34    pub attributes: HashMap<String, String>,
35    pub destination_identities: Vec<ParticipantIdentity>,
36    /// The id associated with the stream. If unspecified, a new uuid will be created and used per
37    /// call.
38    pub id: Option<String>,
39    pub mime_type: Option<String>,
40    pub name: Option<String>,
41    pub total_length: Option<u64>,
42    /// Whether to deflate-raw compress the payload when all recipients support it.
43    /// Defaults to `true` (compression opt-out). Ignored by the incremental `stream_bytes`.
44    pub compress: Option<bool>,
45    /// The identity the stream's packets are attributed to. If unspecified, the packets carry
46    /// no explicit identity and the server attributes them to the sending participant. Only
47    /// participants with the appropriate permission (e.g. agents) may impersonate another
48    /// identity.
49    pub sender_identity: Option<ParticipantIdentity>,
50}
51
52impl StreamByteOptions {
53    pub fn new_with_topic(topic: impl Into<String>) -> Self {
54        Self {
55            topic: topic.into(),
56            attributes: HashMap::new(),
57            destination_identities: vec![],
58            id: None,
59            mime_type: None,
60            name: None,
61            total_length: None,
62            compress: None,
63            sender_identity: None,
64        }
65    }
66
67    /// Sets the topic the stream is published to.
68    pub fn with_topic(mut self, topic: String) -> Self {
69        self.topic = topic;
70        self
71    }
72    /// Replaces all attributes attached to the stream.
73    pub fn with_attributes(mut self, attributes: HashMap<String, String>) -> Self {
74        self.attributes = attributes;
75        self
76    }
77    /// Adds a single attribute to the stream, overwriting any existing value for `key`.
78    pub fn with_attribute(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
79        self.attributes.insert(key.into(), value.into());
80        self
81    }
82    /// Replaces the set of participant identities the stream is delivered to.
83    /// An empty list delivers to all participants in the room.
84    pub fn with_destination_identities(
85        mut self,
86        destination_identities: Vec<ParticipantIdentity>,
87    ) -> Self {
88        self.destination_identities = destination_identities;
89        self
90    }
91    /// Adds a single participant identity to the stream's destinations.
92    pub fn with_destination_identity(mut self, identity: impl Into<ParticipantIdentity>) -> Self {
93        self.destination_identities.push(identity.into());
94        self
95    }
96    /// Sets an explicit stream id. If unset, a new uuid is generated per call.
97    pub fn with_id(mut self, id: impl Into<String>) -> Self {
98        self.id = Some(id.into());
99        self
100    }
101    /// Sets the MIME type describing the stream's payload.
102    pub fn with_mime_type(mut self, mime_type: impl Into<String>) -> Self {
103        self.mime_type = Some(mime_type.into());
104        self
105    }
106    /// Sets a human-readable name for the stream (e.g. a file name).
107    pub fn with_name(mut self, name: impl Into<String>) -> Self {
108        self.name = Some(name.into());
109        self
110    }
111    /// Sets the total byte length of the payload, when known ahead of time.
112    pub fn with_total_length(mut self, total_length: u64) -> Self {
113        self.total_length = Some(total_length);
114        self
115    }
116    /// Sets whether to deflate-raw compress the payload when all recipients support it.
117    /// Defaults to `true` (compression opt-out).
118    pub fn with_compress(mut self, compress: bool) -> Self {
119        self.compress = Some(compress);
120        self
121    }
122    /// Sets the identity the stream's packets are attributed to. Only participants with the
123    /// appropriate permission (e.g. agents) may impersonate another identity.
124    pub fn with_sender_identity(mut self, identity: impl Into<ParticipantIdentity>) -> Self {
125        self.sender_identity = Some(identity.into());
126        self
127    }
128}
129
130/// Options used when opening an outgoing text data stream.
131#[derive(Clone, Debug, Eq, PartialEq)]
132pub struct StreamTextOptions {
133    pub topic: String,
134    pub attributes: HashMap<String, String>,
135    pub destination_identities: Vec<ParticipantIdentity>,
136    /// The id associated with the stream. If unspecified, a new uuid will be created and used per
137    /// call.
138    pub id: Option<String>,
139    pub operation_type: Option<OperationType>,
140    pub version: Option<i32>,
141    pub reply_to_stream_id: Option<String>,
142    pub attached_stream_ids: Vec<String>,
143    pub generated: Option<bool>,
144    /// Whether to deflate-raw compress the payload when all recipients support it.
145    /// Defaults to `true` (compression opt-out). Ignored by the incremental `stream_text`.
146    pub compress: Option<bool>,
147    /// The identity the stream's packets are attributed to. If unspecified, the packets carry
148    /// no explicit identity and the server attributes them to the sending participant. Only
149    /// participants with the appropriate permission (e.g. agents) may impersonate another
150    /// identity.
151    pub sender_identity: Option<ParticipantIdentity>,
152}
153
154impl StreamTextOptions {
155    pub fn new_with_topic(topic: impl Into<String>) -> Self {
156        Self {
157            topic: topic.into(),
158            attributes: HashMap::new(),
159            destination_identities: vec![],
160            id: None,
161            operation_type: None,
162            version: None,
163            reply_to_stream_id: None,
164            attached_stream_ids: vec![],
165            generated: None,
166            compress: None,
167            sender_identity: None,
168        }
169    }
170
171    /// Sets the topic the stream is published to.
172    pub fn with_topic(mut self, topic: String) -> Self {
173        self.topic = topic;
174        self
175    }
176    /// Replaces all attributes attached to the stream.
177    pub fn with_attributes(mut self, attributes: HashMap<String, String>) -> Self {
178        self.attributes = attributes;
179        self
180    }
181    /// Adds a single attribute to the stream, overwriting any existing value for `key`.
182    pub fn with_attribute(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
183        self.attributes.insert(key.into(), value.into());
184        self
185    }
186    /// Replaces the set of participant identities the stream is delivered to.
187    /// An empty list delivers to all participants in the room.
188    pub fn with_destination_identities(
189        mut self,
190        destination_identities: Vec<ParticipantIdentity>,
191    ) -> Self {
192        self.destination_identities = destination_identities;
193        self
194    }
195    /// Adds a single participant identity to the stream's destinations.
196    pub fn with_destination_identity(mut self, identity: impl Into<ParticipantIdentity>) -> Self {
197        self.destination_identities.push(identity.into());
198        self
199    }
200    /// Sets an explicit stream id. If unset, a new uuid is generated per call.
201    pub fn with_id(mut self, id: impl Into<String>) -> Self {
202        self.id = Some(id.into());
203        self
204    }
205    /// Sets the operation this text stream represents (e.g. create or update).
206    pub fn with_operation_type(mut self, operation_type: OperationType) -> Self {
207        self.operation_type = Some(operation_type);
208        self
209    }
210    /// Sets the version of the text, used to order updates to the same content.
211    pub fn with_version(mut self, version: i32) -> Self {
212        self.version = Some(version);
213        self
214    }
215    /// Sets the id of the stream this text is a reply to.
216    pub fn with_reply_to_stream_id(mut self, reply_to_stream_id: impl Into<String>) -> Self {
217        self.reply_to_stream_id = Some(reply_to_stream_id.into());
218        self
219    }
220    /// Replaces the set of stream ids attached to this text (e.g. referenced files).
221    pub fn with_attached_stream_ids(mut self, attached_stream_ids: Vec<String>) -> Self {
222        self.attached_stream_ids = attached_stream_ids;
223        self
224    }
225    /// Adds a single attached stream id to this text.
226    pub fn with_attached_stream_id(mut self, attached_stream_id: impl Into<String>) -> Self {
227        self.attached_stream_ids.push(attached_stream_id.into());
228        self
229    }
230    /// Sets whether the text was machine-generated (e.g. by an agent).
231    pub fn with_generated(mut self, generated: bool) -> Self {
232        self.generated = Some(generated);
233        self
234    }
235    /// Sets whether to deflate-raw compress the payload when all recipients support it.
236    /// Defaults to `true` (compression opt-out).
237    pub fn with_compress(mut self, compress: bool) -> Self {
238        self.compress = Some(compress);
239        self
240    }
241    /// Sets the identity the stream's packets are attributed to. Only participants with the
242    /// appropriate permission (e.g. agents) may impersonate another identity.
243    pub fn with_sender_identity(mut self, identity: impl Into<ParticipantIdentity>) -> Self {
244        self.sender_identity = Some(identity.into());
245        self
246    }
247}
livekit_data_stream/outgoing/mod.rs

livekit_data_stream/outgoing/
mod.rs
