livekit_data_stream/types/
packet.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
15use from_variants::FromVariants;
16use livekit_common::EncryptionType;
17use livekit_protocol::data_stream as proto;
18use std::collections::HashMap;
19
20use super::StreamId;
21
22/// Operation type for text streams.
23#[derive(Clone, Copy, Default, Debug, Hash, Eq, PartialEq)]
24pub enum OperationType {
25    #[default]
26    Create,
27    Update,
28    Delete,
29    Reaction,
30}
31
32impl From<proto::OperationType> for OperationType {
33    fn from(value: proto::OperationType) -> Self {
34        match value {
35            proto::OperationType::Create => Self::Create,
36            proto::OperationType::Update => Self::Update,
37            proto::OperationType::Delete => Self::Delete,
38            proto::OperationType::Reaction => Self::Reaction,
39        }
40    }
41}
42
43impl From<OperationType> for proto::OperationType {
44    fn from(value: OperationType) -> Self {
45        match value {
46            OperationType::Create => Self::Create,
47            OperationType::Update => Self::Update,
48            OperationType::Delete => Self::Delete,
49            OperationType::Reaction => Self::Reaction,
50        }
51    }
52}
53
54/// Header information included exclusively in text data streams
55#[derive(Clone, Debug, Default, PartialEq)]
56pub struct TextHeader {
57    pub(crate) operation_type: OperationType,
58    /// Optional: Version for updates/edits
59    pub(crate) version: i32,
60    /// Optional: Reply to specific message
61    pub(crate) reply_to_stream_id: Option<StreamId>,
62    /// file attachments for text streams
63    pub(crate) attached_stream_ids: Vec<StreamId>,
64    /// true if the text has been generated by an agent from a participant's audio transcription
65    pub(crate) generated: bool,
66}
67
68impl From<proto::TextHeader> for TextHeader {
69    fn from(value: proto::TextHeader) -> Self {
70        Self {
71            operation_type: value.operation_type().into(),
72            version: value.version,
73            reply_to_stream_id: if !value.reply_to_stream_id.is_empty() {
74                Some(value.reply_to_stream_id.into())
75            } else {
76                None
77            },
78            attached_stream_ids: value.attached_stream_ids.into_iter().map(Into::into).collect(),
79            generated: value.generated,
80        }
81    }
82}
83
84impl From<TextHeader> for proto::TextHeader {
85    fn from(value: TextHeader) -> Self {
86        Self {
87            operation_type: proto::OperationType::from(value.operation_type) as i32,
88            version: value.version,
89            reply_to_stream_id: value.reply_to_stream_id.map(Into::into).unwrap_or_default(),
90            attached_stream_ids: value.attached_stream_ids.into_iter().map(Into::into).collect(),
91            generated: value.generated,
92        }
93    }
94}
95
96/// Header information included exclusively in byte data streams
97#[derive(Clone, Debug, Default, PartialEq)]
98pub struct ByteHeader {
99    pub(crate) name: String,
100}
101
102impl From<proto::ByteHeader> for ByteHeader {
103    fn from(value: proto::ByteHeader) -> Self {
104        Self { name: value.name }
105    }
106}
107
108impl From<ByteHeader> for proto::ByteHeader {
109    fn from(value: ByteHeader) -> Self {
110        Self { name: value.name }
111    }
112}
113
114#[derive(Clone, Debug, PartialEq, FromVariants)]
115pub enum ContentHeader {
116    TextHeader(TextHeader),
117    ByteHeader(ByteHeader),
118}
119
120impl From<proto::header::ContentHeader> for ContentHeader {
121    fn from(value: proto::header::ContentHeader) -> Self {
122        match value {
123            proto::header::ContentHeader::TextHeader(text_header) => {
124                Self::TextHeader(text_header.into())
125            }
126            proto::header::ContentHeader::ByteHeader(text_header) => {
127                Self::ByteHeader(text_header.into())
128            }
129        }
130    }
131}
132
133impl From<ContentHeader> for proto::header::ContentHeader {
134    fn from(value: ContentHeader) -> Self {
135        match value {
136            ContentHeader::TextHeader(text_header) => Self::TextHeader(text_header.into()),
137            ContentHeader::ByteHeader(byte_header) => Self::ByteHeader(byte_header.into()),
138        }
139    }
140}
141
142/// Type of compression used when sending a data stream chunk
143#[derive(Clone, Debug, Default, PartialEq)]
144pub enum CompressionType {
145    #[default]
146    None,
147    /// DEFLATE_RAW = DEFLATE without header+checksum/trailer
148    DeflateRaw,
149    /// A compression type this SDK version doesn't recognize (i.e. from a future protocol
150    /// version). Streams carrying it cannot be decoded and are dropped on receive; the send
151    /// path never constructs this variant.
152    Unrecognized,
153}
154
155impl From<proto::CompressionType> for CompressionType {
156    fn from(value: proto::CompressionType) -> Self {
157        match value {
158            proto::CompressionType::DeflateRaw => Self::DeflateRaw,
159            proto::CompressionType::None => Self::None,
160        }
161    }
162}
163
164impl From<CompressionType> for proto::CompressionType {
165    fn from(value: CompressionType) -> Self {
166        match value {
167            CompressionType::DeflateRaw => Self::DeflateRaw,
168            // `Unrecognized` only arises from decoding a foreign header and is never sent.
169            CompressionType::None | CompressionType::Unrecognized => Self::None,
170        }
171    }
172}
173
174#[derive(Clone, Debug, Default, PartialEq)]
175pub struct Header {
176    /// Unique identifier for this data stream
177    pub(crate) stream_id: StreamId,
178    /// using int64 for Unix timestamp
179    pub(crate) timestamp: i64,
180    pub(crate) topic: String,
181    pub(crate) mime_type: String,
182    /// only populated for finite streams, if it's a stream of unknown size this stays empty
183    pub(crate) total_length: Option<u64>,
184    /// user defined attributes map that can carry additional info
185    pub(crate) attributes: HashMap<String, String>,
186    /// Optional inline content so that a data stream can be sent as a single packet for short payloads.
187    ///
188    /// content as binary (bytes)
189    pub(crate) inline_content: Option<Vec<u8>>,
190    pub(crate) compression: CompressionType,
191    /// oneof to choose between specific header types
192    pub(crate) content_header: Option<ContentHeader>,
193}
194
195impl From<proto::Header> for Header {
196    fn from(value: proto::Header) -> Self {
197        // Don't use the prost `compression()` accessor here: it silently maps out-of-range
198        // values (a compression type from a future protocol version) to the default `None`,
199        // which would make the receiver deliver still-compressed bytes as content. Preserve
200        // unknown values as `Unrecognized` so the incoming manager can drop the stream.
201        let compression = proto::CompressionType::try_from(value.compression)
202            .map(CompressionType::from)
203            .unwrap_or(CompressionType::Unrecognized);
204        let content_header: Option<ContentHeader> =
205            value.content_header.map(|content_header| content_header.into());
206        Self {
207            stream_id: value.stream_id.into(),
208            timestamp: value.timestamp,
209            topic: value.topic,
210            mime_type: value.mime_type,
211            total_length: value.total_length,
212            attributes: value.attributes,
213            inline_content: value.inline_content,
214            compression,
215            content_header,
216        }
217    }
218}
219
220impl From<Header> for proto::Header {
221    fn from(value: Header) -> Self {
222        // `encryption_type` is deprecated on the proto (it's carried on the DataPacket instead);
223        // `..Default::default()` fills it without naming the deprecated field.
224        Self {
225            stream_id: value.stream_id.into(),
226            timestamp: value.timestamp,
227            topic: value.topic,
228            mime_type: value.mime_type,
229            total_length: value.total_length,
230            attributes: value.attributes,
231            inline_content: value.inline_content,
232            compression: proto::CompressionType::from(value.compression) as i32,
233            content_header: value.content_header.map(Into::into),
234            ..Default::default()
235        }
236    }
237}
238
239#[derive(Clone, Debug, Default, PartialEq)]
240pub struct Chunk {
241    /// Unique identifier for this data stream to map it to the correct header
242    pub(crate) stream_id: StreamId,
243    pub(crate) chunk_index: u64,
244    /// Content as binary (bytes)
245    pub(crate) content: Vec<u8>,
246    /// A version indicating that this chunk_index has been retroactively modified and the original one needs to be replaced
247    pub(crate) version: i32,
248    pub(crate) encryption_type: EncryptionType,
249}
250
251impl From<proto::Chunk> for Chunk {
252    fn from(value: proto::Chunk) -> Self {
253        // The proto carries encryption on the enclosing `DataPacket`, not the chunk, so the
254        // chunk's own `encryption_type` defaults here; the authoritative value rides on `Packet`.
255        Self {
256            stream_id: value.stream_id.into(),
257            chunk_index: value.chunk_index,
258            content: value.content,
259            version: value.version,
260            encryption_type: EncryptionType::default(),
261        }
262    }
263}
264
265impl From<Chunk> for proto::Chunk {
266    fn from(value: Chunk) -> Self {
267        // `iv` is deprecated on the proto (encryption rides on the DataPacket);
268        // `..Default::default()` fills it without naming the deprecated field.
269        Self {
270            stream_id: value.stream_id.into(),
271            chunk_index: value.chunk_index,
272            content: value.content,
273            version: value.version,
274            ..Default::default()
275        }
276    }
277}
278
279#[derive(Clone, Debug, Default, PartialEq)]
280pub struct Trailer {
281    /// Unique identifier for this data stream
282    pub(crate) stream_id: StreamId,
283    /// Reason why the stream was closed (could contain "error" / "interrupted" / empty for expected end)
284    pub(crate) reason: String,
285    /// Any final attribute updates for the stream
286    pub(crate) attributes: HashMap<String, String>,
287}
288
289impl From<proto::Trailer> for Trailer {
290    fn from(value: proto::Trailer) -> Self {
291        Self {
292            stream_id: value.stream_id.into(),
293            reason: value.reason,
294            attributes: value.attributes,
295        }
296    }
297}
298
299impl From<Trailer> for proto::Trailer {
300    fn from(value: Trailer) -> Self {
301        Self {
302            stream_id: value.stream_id.into(),
303            reason: value.reason,
304            attributes: value.attributes,
305        }
306    }
307}
308
309#[derive(Clone, Debug, PartialEq)]
310pub enum Packet {
311    Header { header: Header, encryption_type: EncryptionType },
312    Chunk { chunk: Chunk, encryption_type: EncryptionType },
313    Trailer(Trailer),
314}
livekit_data_stream/types/packet.rs

livekit_data_stream/types/
packet.rs
