livekit_data_stream/
outgoing.rs

1// Copyright 2025 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 super::{
16    create_random_uuid, ByteStreamInfo, OperationType, SendError, StreamError, StreamProgress,
17    StreamResult, TextStreamInfo,
18};
19use crate::utf8_chunk::Utf8AwareChunkExt;
20use bmrng::unbounded::{UnboundedRequestReceiver, UnboundedRequestSender};
21use chrono::Utc;
22use livekit_common::ParticipantIdentity;
23use livekit_protocol as proto;
24use std::{collections::HashMap, path::Path, sync::Arc};
25use tokio::{io::AsyncReadExt, sync::Mutex};
26
27/// Writer for an open data stream.
28pub trait StreamWriter<'a> {
29    /// Type of input this writer accepts.
30    type Input: 'a;
31
32    /// Information about the underlying data stream.
33    type Info;
34
35    /// Returns a reference to the stream info.
36    fn info(&self) -> &Self::Info;
37
38    /// Writes to the stream.
39    fn write(
40        &self,
41        input: Self::Input,
42    ) -> impl std::future::Future<Output = StreamResult<()>> + Send;
43
44    /// Closes the stream normally.
45    fn close(self) -> impl std::future::Future<Output = StreamResult<()>> + Send;
46
47    /// Closes the stream abnormally, specifying the reason for closure.
48    fn close_with_reason(
49        self,
50        reason: &str,
51    ) -> impl std::future::Future<Output = StreamResult<()>> + Send;
52}
53
54#[derive(Clone)]
55/// Writer for an open byte data stream.
56pub struct ByteStreamWriter {
57    info: Arc<ByteStreamInfo>,
58    stream: Arc<Mutex<RawStream>>,
59}
60
61#[derive(Clone)]
62/// Writer for an open text data stream.
63pub struct TextStreamWriter {
64    info: Arc<TextStreamInfo>,
65    stream: Arc<Mutex<RawStream>>,
66}
67
68impl<'a> StreamWriter<'a> for ByteStreamWriter {
69    type Input = &'a [u8];
70    type Info = ByteStreamInfo;
71
72    fn info(&self) -> &Self::Info {
73        &self.info
74    }
75
76    async fn write(&self, bytes: &'a [u8]) -> StreamResult<()> {
77        let mut stream = self.stream.lock().await;
78        for chunk in bytes.chunks(CHUNK_SIZE) {
79            stream.write_chunk(chunk).await?;
80        }
81        Ok(())
82    }
83
84    async fn close(self) -> StreamResult<()> {
85        self.stream.lock().await.close(None).await
86    }
87
88    async fn close_with_reason(self, reason: &str) -> StreamResult<()> {
89        self.stream.lock().await.close(Some(reason)).await
90    }
91}
92
93impl ByteStreamWriter {
94    /// Writes the contents of the file incrementally.
95    async fn write_file_contents(&self, path: impl AsRef<Path>) -> StreamResult<()> {
96        let mut stream = self.stream.lock().await;
97        let mut file = tokio::fs::File::open(path).await?;
98        let mut buffer = vec![0; 8192]; // 8KB
99        loop {
100            let bytes_read = file.read(&mut buffer).await?;
101            if bytes_read == 0 {
102                break;
103            }
104            stream.write_chunk(&buffer[..bytes_read]).await?;
105        }
106        Ok(())
107    }
108}
109
110impl<'a> StreamWriter<'a> for TextStreamWriter {
111    type Input = &'a str;
112    type Info = TextStreamInfo;
113
114    fn info(&self) -> &Self::Info {
115        &self.info
116    }
117
118    async fn write(&self, text: &'a str) -> StreamResult<()> {
119        let mut stream = self.stream.lock().await;
120        for chunk in text.as_bytes().utf8_aware_chunks(CHUNK_SIZE) {
121            stream.write_chunk(chunk).await?;
122        }
123        Ok(())
124    }
125
126    async fn close(self) -> StreamResult<()> {
127        self.stream.lock().await.close(None).await
128    }
129
130    async fn close_with_reason(self, reason: &str) -> StreamResult<()> {
131        self.stream.lock().await.close(Some(reason)).await
132    }
133}
134
135struct RawStreamOpenOptions {
136    header: proto::data_stream::Header,
137    destination_identities: Vec<ParticipantIdentity>,
138    packet_tx: UnboundedRequestSender<proto::DataPacket, Result<(), SendError>>,
139}
140
141struct RawStream {
142    id: String,
143    progress: StreamProgress,
144    is_closed: bool,
145    /// Request channel for sending packets.
146    packet_tx: UnboundedRequestSender<proto::DataPacket, Result<(), SendError>>,
147}
148
149impl RawStream {
150    async fn open(options: RawStreamOpenOptions) -> StreamResult<Self> {
151        let id = options.header.stream_id.to_string();
152        let bytes_total = options.header.total_length;
153
154        let packet = Self::create_header_packet(options.header, options.destination_identities);
155        Self::send_packet(&options.packet_tx, packet).await?;
156
157        Ok(Self {
158            id,
159            progress: StreamProgress { bytes_total, ..Default::default() },
160            is_closed: false,
161            packet_tx: options.packet_tx,
162        })
163    }
164
165    async fn write_chunk(&mut self, bytes: &[u8]) -> StreamResult<()> {
166        let packet = Self::create_chunk_packet(&self.id, self.progress.chunk_index, bytes);
167        Self::send_packet(&self.packet_tx, packet).await?;
168        self.progress.bytes_processed += bytes.len() as u64;
169        self.progress.chunk_index += 1;
170        Ok(())
171    }
172
173    async fn close(&mut self, reason: Option<&str>) -> StreamResult<()> {
174        if self.is_closed {
175            Err(StreamError::AlreadyClosed)?
176        }
177        let packet = Self::create_trailer_packet(&self.id, reason);
178        Self::send_packet(&self.packet_tx, packet).await?;
179        self.is_closed = true;
180        Ok(())
181    }
182
183    async fn send_packet(
184        tx: &UnboundedRequestSender<proto::DataPacket, Result<(), SendError>>,
185        packet: proto::DataPacket,
186    ) -> StreamResult<()> {
187        tx.send_receive(packet)
188            .await
189            .map_err(|_| StreamError::Internal)? // request channel closed
190            .map_err(|_| StreamError::SendFailed) // data channel error
191    }
192
193    fn create_header_packet(
194        header: proto::data_stream::Header,
195        destination_identities: Vec<ParticipantIdentity>,
196    ) -> proto::DataPacket {
197        proto::DataPacket {
198            kind: proto::data_packet::Kind::Reliable.into(),
199            participant_identity: String::new(), // populate later
200            destination_identities: destination_identities.into_iter().map(|id| id.0).collect(),
201            value: Some(livekit_protocol::data_packet::Value::StreamHeader(header)),
202            // TODO: placeholder for reliable data transport
203            ..Default::default()
204        }
205    }
206
207    fn create_chunk_packet(id: &str, chunk_index: u64, content: &[u8]) -> proto::DataPacket {
208        let chunk = proto::data_stream::Chunk {
209            stream_id: id.to_string(),
210            chunk_index,
211            content: content.to_vec(),
212            ..Default::default()
213        };
214        proto::DataPacket {
215            kind: proto::data_packet::Kind::Reliable.into(),
216            participant_identity: String::new(), // populate later
217            value: Some(livekit_protocol::data_packet::Value::StreamChunk(chunk)),
218            ..Default::default()
219        }
220    }
221
222    fn create_trailer_packet(id: &str, reason: Option<&str>) -> proto::DataPacket {
223        let trailer = proto::data_stream::Trailer {
224            stream_id: id.to_string(),
225            reason: reason.unwrap_or_default().to_owned(),
226            ..Default::default()
227        };
228        proto::DataPacket {
229            kind: proto::data_packet::Kind::Reliable.into(),
230            participant_identity: String::new(), // populate later
231            value: Some(livekit_protocol::data_packet::Value::StreamTrailer(trailer)),
232            ..Default::default()
233        }
234    }
235}
236
237impl Drop for RawStream {
238    /// Close stream normally if not already closed.
239    fn drop(&mut self) {
240        if self.is_closed {
241            return;
242        }
243        let packet = Self::create_trailer_packet(&self.id, None);
244        let packet_tx = self.packet_tx.clone();
245        // Use try_current() instead of assuming a Tokio runtime exists.
246        // The drop can run on a non-Tokio thread (e.g. a GC finalizer in
247        // Unity/.NET) or after the runtime has shut down, in which case
248        // we silently skip the trailer — the connection is going away anyway.
249        if let Ok(handle) = tokio::runtime::Handle::try_current() {
250            handle.spawn(async move { Self::send_packet(&packet_tx, packet).await });
251        }
252    }
253}
254
255/// Options used when opening an outgoing byte data stream.
256#[derive(Clone, Default, Debug, Eq, PartialEq)]
257pub struct StreamByteOptions {
258    pub topic: String,
259    pub attributes: HashMap<String, String>,
260    pub destination_identities: Vec<ParticipantIdentity>,
261    pub id: Option<String>,
262    pub mime_type: Option<String>,
263    pub name: Option<String>,
264    pub total_length: Option<u64>,
265}
266
267/// Options used when opening an outgoing text data stream.
268#[derive(Clone, Default, Debug, Eq, PartialEq)]
269pub struct StreamTextOptions {
270    pub topic: String,
271    pub attributes: HashMap<String, String>,
272    pub destination_identities: Vec<ParticipantIdentity>,
273    pub id: Option<String>,
274    pub operation_type: Option<OperationType>,
275    pub version: Option<i32>,
276    pub reply_to_stream_id: Option<String>,
277    pub attached_stream_ids: Vec<String>,
278    pub generated: Option<bool>,
279}
280
281#[derive(Clone)]
282pub struct OutgoingStreamManager {
283    /// Request channel for sending packets.
284    packet_tx: UnboundedRequestSender<proto::DataPacket, Result<(), SendError>>,
285}
286
287impl OutgoingStreamManager {
288    pub fn new() -> (Self, UnboundedRequestReceiver<proto::DataPacket, Result<(), SendError>>) {
289        let (packet_tx, packet_rx) = bmrng::unbounded_channel();
290        let manager = Self { packet_tx };
291        (manager, packet_rx)
292    }
293
294    pub async fn stream_text(&self, options: StreamTextOptions) -> StreamResult<TextStreamWriter> {
295        let text_header = proto::data_stream::TextHeader {
296            operation_type: options.operation_type.unwrap_or_default() as i32,
297            version: options.version.unwrap_or_default(),
298            reply_to_stream_id: options.reply_to_stream_id.unwrap_or_default(),
299            attached_stream_ids: options.attached_stream_ids,
300            generated: options.generated.unwrap_or_default(),
301        };
302        let header = proto::data_stream::Header {
303            stream_id: options.id.unwrap_or_else(|| create_random_uuid()),
304            timestamp: Utc::now().timestamp_millis(),
305            topic: options.topic,
306            mime_type: TEXT_MIME_TYPE.to_owned(),
307            total_length: None,
308            encryption_type: proto::encryption::Type::None.into(),
309            attributes: options.attributes,
310            content_header: Some(proto::data_stream::header::ContentHeader::TextHeader(
311                text_header.clone(),
312            )),
313            // Data streams v2 fields
314            inline_content: None,
315            compression: proto::data_stream::CompressionType::None as i32,
316        };
317        let open_options = RawStreamOpenOptions {
318            header: header.clone(),
319            destination_identities: options.destination_identities,
320            packet_tx: self.packet_tx.clone(),
321        };
322        let writer = TextStreamWriter {
323            info: Arc::new(TextStreamInfo::from_headers(header, text_header)),
324            stream: Arc::new(Mutex::new(RawStream::open(open_options).await?)),
325        };
326        Ok(writer)
327    }
328
329    pub async fn stream_bytes(&self, options: StreamByteOptions) -> StreamResult<ByteStreamWriter> {
330        let byte_header = proto::data_stream::ByteHeader { name: options.name.unwrap_or_default() };
331        let header = proto::data_stream::Header {
332            stream_id: options.id.unwrap_or_else(|| create_random_uuid()),
333            timestamp: Utc::now().timestamp_millis(),
334            topic: options.topic,
335            mime_type: options.mime_type.unwrap_or_else(|| BYTE_MIME_TYPE.to_owned()),
336            total_length: options.total_length,
337            encryption_type: proto::encryption::Type::None.into(),
338            attributes: options.attributes,
339            content_header: Some(proto::data_stream::header::ContentHeader::ByteHeader(
340                byte_header.clone(),
341            )),
342            // Data streams v2 fields
343            inline_content: None,
344            compression: proto::data_stream::CompressionType::None as i32,
345        };
346
347        let open_options = RawStreamOpenOptions {
348            header: header.clone(),
349            destination_identities: options.destination_identities,
350            packet_tx: self.packet_tx.clone(),
351        };
352        let writer = ByteStreamWriter {
353            info: Arc::new(ByteStreamInfo::from_headers(header, byte_header)),
354            stream: Arc::new(Mutex::new(RawStream::open(open_options).await?)),
355        };
356        Ok(writer)
357    }
358
359    pub async fn send_text(
360        &self,
361        text: &str,
362        options: StreamTextOptions,
363    ) -> StreamResult<TextStreamInfo> {
364        let text_header = proto::data_stream::TextHeader {
365            operation_type: options.operation_type.unwrap_or_default() as i32,
366            version: options.version.unwrap_or_default(),
367            reply_to_stream_id: options.reply_to_stream_id.unwrap_or_default(),
368            attached_stream_ids: options.attached_stream_ids,
369            generated: options.generated.unwrap_or_default(),
370        };
371        let header = proto::data_stream::Header {
372            stream_id: options.id.unwrap_or_else(|| create_random_uuid()),
373            timestamp: Utc::now().timestamp_millis(),
374            topic: options.topic,
375            mime_type: TEXT_MIME_TYPE.to_owned(),
376            total_length: Some(text.bytes().len() as u64),
377            encryption_type: proto::encryption::Type::None.into(),
378            attributes: options.attributes,
379            content_header: Some(proto::data_stream::header::ContentHeader::TextHeader(
380                text_header.clone(),
381            )),
382            // Data streams v2 fields
383            inline_content: None,
384            compression: proto::data_stream::CompressionType::None as i32,
385        };
386        let open_options = RawStreamOpenOptions {
387            header: header.clone(),
388            destination_identities: options.destination_identities,
389            packet_tx: self.packet_tx.clone(),
390        };
391        let writer = TextStreamWriter {
392            info: Arc::new(TextStreamInfo::from_headers(header, text_header)),
393            stream: Arc::new(Mutex::new(RawStream::open(open_options).await?)),
394        };
395
396        let info = (*writer.info).clone();
397        writer.write(text).await?;
398        writer.close().await?;
399
400        Ok(info)
401    }
402
403    /// Send bytes to participants in the room.
404    ///
405    /// This method sends an in-memory blob of bytes to participants in the room
406    /// as a byte stream. It opens a stream using the provided options, writes the
407    /// entire buffer, and closes the stream before returning.
408    ///
409    /// The `total_length` in the header is set from the provided data and is not
410    /// overridable by `options.total_length`.
411    pub async fn send_bytes(
412        &self,
413        data: impl AsRef<[u8]>,
414        options: StreamByteOptions,
415    ) -> StreamResult<ByteStreamInfo> {
416        if options.total_length.is_some() {
417            log::warn!("Ignoring total_length option specified for send_bytes");
418        }
419        let bytes = data.as_ref();
420
421        let byte_header = proto::data_stream::ByteHeader { name: options.name.unwrap_or_default() };
422        let header = proto::data_stream::Header {
423            stream_id: options.id.unwrap_or_else(|| create_random_uuid()),
424            timestamp: Utc::now().timestamp_millis(),
425            topic: options.topic,
426            mime_type: options.mime_type.unwrap_or_else(|| BYTE_MIME_TYPE.to_owned()),
427            total_length: Some(bytes.len() as u64), // not overridable
428            encryption_type: proto::encryption::Type::None.into(),
429            attributes: options.attributes,
430            content_header: Some(proto::data_stream::header::ContentHeader::ByteHeader(
431                byte_header.clone(),
432            )),
433            // Data streams v2 fields
434            inline_content: None,
435            compression: proto::data_stream::CompressionType::None as i32,
436        };
437
438        let open_options = RawStreamOpenOptions {
439            header: header.clone(),
440            destination_identities: options.destination_identities,
441            packet_tx: self.packet_tx.clone(),
442        };
443        let writer = ByteStreamWriter {
444            info: Arc::new(ByteStreamInfo::from_headers(header, byte_header)),
445            stream: Arc::new(Mutex::new(RawStream::open(open_options).await?)),
446        };
447
448        let info = (*writer.info).clone();
449        writer.write(bytes).await?;
450        writer.close().await?;
451
452        Ok(info)
453    }
454
455    pub async fn send_file(
456        &self,
457        path: impl AsRef<Path>,
458        options: StreamByteOptions,
459    ) -> StreamResult<ByteStreamInfo> {
460        let file_size = tokio::fs::metadata(path.as_ref())
461            .await
462            .map(|metadata| metadata.len())
463            .map_err(|e| StreamError::from(e))?;
464        let name =
465            path.as_ref().file_name().and_then(|n| n.to_str()).unwrap_or_default().to_owned();
466
467        let byte_header = proto::data_stream::ByteHeader { name };
468        let header = proto::data_stream::Header {
469            stream_id: options.id.unwrap_or_else(|| create_random_uuid()),
470            timestamp: Utc::now().timestamp_millis(),
471            topic: options.topic,
472            mime_type: options.mime_type.unwrap_or_else(|| BYTE_MIME_TYPE.to_owned()),
473            total_length: Some(file_size as u64), // not overridable
474            encryption_type: proto::encryption::Type::None.into(),
475            attributes: options.attributes,
476            content_header: Some(proto::data_stream::header::ContentHeader::ByteHeader(
477                byte_header.clone(),
478            )),
479            // Data streams v2 fields
480            inline_content: None,
481            compression: proto::data_stream::CompressionType::None as i32,
482        };
483
484        let open_options = RawStreamOpenOptions {
485            header: header.clone(),
486            destination_identities: options.destination_identities,
487            packet_tx: self.packet_tx.clone(),
488        };
489        let writer = ByteStreamWriter {
490            info: Arc::new(ByteStreamInfo::from_headers(header, byte_header)),
491            stream: Arc::new(Mutex::new(RawStream::open(open_options).await?)),
492        };
493
494        let info = (*writer.info).clone();
495        writer.write_file_contents(path).await?;
496        writer.close().await?;
497
498        Ok(info)
499    }
500}
501
502/// Maximum number of bytes to send in a single chunk.
503static CHUNK_SIZE: usize = 15000;
504
505// Default MIME type to use for byte streams.
506static BYTE_MIME_TYPE: &str = "application/octet-stream";
507
508/// Default MIME type to use for text streams.
509static TEXT_MIME_TYPE: &str = "text/plain";
510
511#[cfg(test)]
512mod tests {
513    use super::*;
514
515    type Sent = Arc<std::sync::Mutex<Vec<proto::DataPacket>>>;
516
517    fn setup() -> (OutgoingStreamManager, Sent) {
518        let (manager, mut packet_rx) = OutgoingStreamManager::new();
519        let sent: Sent = Arc::new(std::sync::Mutex::new(Vec::new()));
520        let sink = sent.clone();
521        tokio::spawn(async move {
522            while let Ok((packet, responder)) = packet_rx.recv().await {
523                sink.lock().unwrap().push(packet);
524                let _ = responder.respond(Ok(()));
525            }
526        });
527        (manager, sent)
528    }
529
530    fn ids(list: &[&str]) -> Vec<ParticipantIdentity> {
531        list.iter().map(|s| ParticipantIdentity(s.to_string())).collect()
532    }
533
534    fn text_opts(topic: &str, dests: &[&str]) -> StreamTextOptions {
535        StreamTextOptions {
536            topic: topic.to_string(),
537            destination_identities: ids(dests),
538            ..Default::default()
539        }
540    }
541
542    fn byte_opts(topic: &str, dests: &[&str]) -> StreamByteOptions {
543        StreamByteOptions {
544            topic: topic.to_string(),
545            destination_identities: ids(dests),
546            ..Default::default()
547        }
548    }
549
550    fn header(p: &proto::DataPacket) -> &proto::data_stream::Header {
551        match p.value.as_ref().unwrap() {
552            proto::data_packet::Value::StreamHeader(h) => h,
553            _ => panic!("expected stream header"),
554        }
555    }
556
557    fn chunk(p: &proto::DataPacket) -> &proto::data_stream::Chunk {
558        match p.value.as_ref().unwrap() {
559            proto::data_packet::Value::StreamChunk(c) => c,
560            _ => panic!("expected stream chunk"),
561        }
562    }
563
564    fn is_text_header(h: &proto::data_stream::Header) -> bool {
565        matches!(h.content_header, Some(proto::data_stream::header::ContentHeader::TextHeader(_)))
566    }
567
568    fn is_byte_header(h: &proto::data_stream::Header) -> bool {
569        matches!(h.content_header, Some(proto::data_stream::header::ContentHeader::ByteHeader(_)))
570    }
571
572    fn assert_trailer(p: &proto::DataPacket) {
573        match p.value.as_ref().unwrap() {
574            proto::data_packet::Value::StreamTrailer(t) => assert_eq!(t.reason, ""),
575            _ => panic!("expected stream trailer"),
576        }
577    }
578
579    fn none_i32() -> i32 {
580        proto::data_stream::CompressionType::None as i32
581    }
582
583    #[tokio::test]
584    async fn short_text_is_legacy_multipacket() {
585        let (m, sent) = setup();
586        m.send_text("hello world", text_opts("chat", &[])).await.unwrap();
587        let p = sent.lock().unwrap().clone();
588        assert_eq!(p.len(), 3);
589        let h = header(&p[0]);
590        assert!(is_text_header(h));
591        assert_eq!(h.topic, "chat");
592        assert_eq!(h.compression, none_i32());
593        assert!(h.inline_content.is_none());
594        let c = chunk(&p[1]);
595        assert_eq!(c.chunk_index, 0);
596        assert_eq!(c.content, b"hello world");
597        assert_trailer(&p[2]);
598    }
599
600    #[tokio::test]
601    async fn long_text_splits_at_mtu() {
602        let (m, sent) = setup();
603        let text = "A".repeat(40_000);
604        m.send_text(&text, text_opts("chat", &[])).await.unwrap();
605        let p = sent.lock().unwrap().clone();
606        assert_eq!(p.len(), 5); // header + 3 chunks + trailer
607        assert_eq!(header(&p[0]).compression, none_i32());
608        assert_eq!(chunk(&p[1]).content.len(), 15_000);
609        assert_eq!(chunk(&p[2]).content.len(), 15_000);
610        assert_eq!(chunk(&p[3]).content.len(), 10_000);
611        assert_eq!(chunk(&p[1]).chunk_index, 0);
612        assert_eq!(chunk(&p[3]).chunk_index, 2);
613        assert_trailer(&p[4]);
614    }
615
616    #[tokio::test]
617    async fn bytes_is_legacy_multipacket() {
618        let (m, sent) = setup();
619        m.send_bytes([0u8, 1, 2, 3], byte_opts("blob", &[])).await.unwrap();
620        let p = sent.lock().unwrap().clone();
621        assert_eq!(p.len(), 3);
622        let h = header(&p[0]);
623        assert!(is_byte_header(h));
624        assert_eq!(h.compression, none_i32());
625        assert!(h.inline_content.is_none());
626        assert_eq!(chunk(&p[1]).content, vec![0, 1, 2, 3]);
627        assert_trailer(&p[2]);
628    }
629
630    // Regression test for CLT-2773: dropping a `RawStream` on a thread that has
631    // no Tokio runtime in TLS (e.g. the .NET GC finalizer thread in the Unity
632    // SDK) used to panic because `Drop` called `tokio::spawn` unconditionally.
633    #[test]
634    fn drop_raw_stream_on_non_tokio_thread_does_not_panic() {
635        let rt = tokio::runtime::Runtime::new().unwrap();
636
637        let raw_stream = rt.block_on(async {
638            let (packet_tx, mut packet_rx) =
639                bmrng::unbounded_channel::<proto::DataPacket, Result<(), SendError>>();
640
641            tokio::spawn(async move {
642                while let Ok((_packet, responder)) = packet_rx.recv().await {
643                    let _ = responder.respond(Ok(()));
644                }
645            });
646
647            let header = proto::data_stream::Header {
648                stream_id: "gc-test-stream".to_string(),
649                timestamp: 0,
650                topic: "gc-test-topic".to_string(),
651                mime_type: TEXT_MIME_TYPE.to_owned(),
652                total_length: None,
653                encryption_type: proto::encryption::Type::None.into(),
654                attributes: HashMap::new(),
655                content_header: None,
656                // Data streams v2 fields
657                inline_content: None,
658                compression: proto::data_stream::CompressionType::None as i32,
659            };
660
661            RawStream::open(RawStreamOpenOptions {
662                header,
663                destination_identities: vec![],
664                packet_tx,
665            })
666            .await
667            .expect("RawStream should open")
668        });
669
670        let drop_thread = std::thread::spawn(move || drop(raw_stream));
671
672        drop_thread.join().expect("Dropping RawStream on a non-Tokio thread must not panic");
673    }
674}
livekit_data_stream/outgoing.rs

livekit_data_stream/
outgoing.rs
