diff --git a/.changeset/data_streams_v2_uniffi.md b/.changeset/data_streams_v2_uniffi.md new file mode 100644 index 000000000..423326be3 --- /dev/null +++ b/.changeset/data_streams_v2_uniffi.md @@ -0,0 +1,9 @@ +--- +livekit: patch +livekit-data-stream: patch +livekit-ffi: patch +livekit-uniffi: patch +livekit-datatrack: patch +--- + +Add data streams v2 to exposed uniffi interface - #1286 (@1egoman) diff --git a/Cargo.lock b/Cargo.lock index 6801c9546..81f4ed127 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4111,6 +4111,8 @@ dependencies = [ "camino", "futures-util", "livekit-api", + "livekit-common", + "livekit-data-stream", "livekit-datatrack", "livekit-protocol", "log", diff --git a/Cargo.toml b/Cargo.toml index f7943db61..db3f43d77 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -87,6 +87,15 @@ serde_json = "1.0" thiserror = "2" tokio = { version = "1", default-features = false } tokio-stream = "0.1" +# Test on a 64-bit ARM device before you change this version. +# +# The Kotlin bindings from uniffi 0.31.2 and 0.32.0 compare each checksum incorrectly on 64-bit +# ARM. Every affected method then fails. See https://github.com/mozilla/uniffi-rs/pull/2897, which +# introduced the defect. Version 0.31.1 has a related defect on 32-bit ARM, and it also does not +# build here, because uniffi-dart requires 0.31.2 or later. mozilla/uniffi-rs#2935 corrects both +# defects, but no release (as of mid august 2026) contains that change. +# +# For this reason, livekit-uniffi sets `omit_checksums` for Kotlin. See livekit-uniffi/uniffi.toml. uniffi = "0.31" # For examples diff --git a/datastream_uniffi_test.py b/datastream_uniffi_test.py new file mode 100644 index 000000000..f5cb4cd75 --- /dev/null +++ b/datastream_uniffi_test.py @@ -0,0 +1,73 @@ +import asyncio +import livekit_uniffi + +class OutgoingDelegate(livekit_uniffi.OutgoingDataStreamManagerDelegate): + def on_packets_available(self, packets): + print('PACKETS:', packets) + +class RemoteParticipantRegistry(livekit_uniffi.RemoteParticipantRegistryDelegate): + def remote_capabilities(self, identity): + return [] # typing.List[ClientCapability] + + def remote_client_protocol(self, identity): + return 2 + + def remote_identities(self): + return ["alice", "bob", "randy"] + +class IncomingDelegate(livekit_uniffi.IncomingDataStreamManagerDelegate): + """Forwards opened readers onto the main asyncio loop. + + Delegate callbacks fire on a Rust tokio thread, so they must not block or await; + hand the reader off to the main loop and let it drive the async reads. + """ + + def __init__(self, loop: asyncio.AbstractEventLoop, opened: asyncio.Queue): + self._loop = loop + self._opened = opened + + def on_byte_stream_opened(self, reader, identity: str): + self._loop.call_soon_threadsafe(self._opened.put_nowait, ("byte", reader, identity)) + + def on_text_stream_opened(self, reader, identity: str): + self._loop.call_soon_threadsafe(self._opened.put_nowait, ("text", reader, identity)) + +# Encoded livekit.DataPacket envelopes (participant_identity = "alice") carrying a +# DataStream.Header / Chunk / Trailer for an 11-byte "hello world" text stream. +DATA_STREAM_HEADER_BYTES = b'"\x05alicej@\n\x11example-stream-id\x10\xad\xf5\xcb\xae\xf93\x1a\x08my-topic"\ntext/plain(\x0bB\n\n\x03foo\x12\x03barJ\x00' +DATA_STREAM_CHUNK_BYTES = b'"\x05alicer \n\x11example-stream-id\x1a\x0bhello world' +DATA_STREAM_TRAILER_BYTES = b'"\x05alicez\'\n\x11example-stream-id\x1a\x12\n\x06status\x12\x08complete' + +async def main(): + opened = asyncio.Queue() + + print("--- OUTGOING:") + outgoing_delegate = OutgoingDelegate() + remote_participant_registry = RemoteParticipantRegistry() + outgoing = livekit_uniffi.OutgoingDataStreamManager(outgoing_delegate, remote_participant_registry) + await outgoing.send_text('hello world', livekit_uniffi.StreamTextOptions( + topic="test", + attributes={}, + # destination_identities: 'typing.List[str]' = , + # id: 'typing.Optional[str]' = , + # operation_type: 'typing.Optional[OperationType]' = , + # version: 'typing.Optional[int]' = , + # reply_to_stream_id: 'typing.Optional[str]' = , + # attached_stream_ids: 'typing.List[str]' = , + # generated: 'typing.Optional[bool]' = , + # compress: 'typing.Optional[bool]' = , + # sender_identity: 'typing.Optional[str]' = + )) + + print("--- INCOMING:") + incoming_delegate = IncomingDelegate(asyncio.get_running_loop(), opened) + incoming = livekit_uniffi.IncomingDataStreamManager(incoming_delegate, [], None) + incoming.handle_packet_received(DATA_STREAM_HEADER_BYTES) + incoming.handle_packet_received(DATA_STREAM_CHUNK_BYTES) + incoming.handle_packet_received(DATA_STREAM_TRAILER_BYTES) + + kind, reader, identity = await asyncio.wait_for(opened.get(), timeout=5) + print(f"{kind.upper()} STREAM OPENED:", identity, "CONTENTS:", await reader.read_all()) + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/livekit-data-stream/src/incoming/events.rs b/livekit-data-stream/src/incoming/events.rs index 2effa426e..1b5fa36c4 100644 --- a/livekit-data-stream/src/incoming/events.rs +++ b/livekit-data-stream/src/incoming/events.rs @@ -17,7 +17,7 @@ use livekit_common::ParticipantIdentity; use crate::{ incoming::AnyStreamReader, - types::{Chunk, Packet, Trailer}, + types::{Chunk, Packet, StreamId, Trailer}, }; pub struct PacketReceived { @@ -39,6 +39,14 @@ pub enum InputEvent { PacketReceived(PacketReceived), /// Abort every open stream sent by this participant (they disconnected mid-send). AbortStreamsFrom(ParticipantIdentity), + /// Abort every open stream (e.g. the local connection is going away). Unlike + /// [`InputEvent::Shutdown`], the run loop keeps going so streams opened later are still handled. + AbortAllStreams, + /// Reply with the number of currently open streams (registered by a header and awaiting more + /// packets). Processed in order with the other events, so the answer reflects everything + /// enqueued before it. + #[from_variants(skip)] + QueryOpenStreamCount(tokio::sync::oneshot::Sender), /// Stop the run loop. Shutdown, } @@ -50,11 +58,29 @@ pub struct StreamOpened { pub participant_identity: ParticipantIdentity, } +/// A stream previously announced via [`StreamOpened`] has terminated and will produce no further +/// data: its trailer arrived, its inline payload completed, it failed with an error, or it was +/// aborted. +/// +/// Emitted exactly once per opened stream. Hosts delivering streams on ordered topics use this to +/// know when a stream's handler can be considered finished on the wire (a trailer alone is not +/// enough: inline single-packet streams never receive one). +pub struct StreamClosed { + pub stream_id: StreamId, + pub participant_identity: ParticipantIdentity, + /// Topic the stream was opened on. + pub topic: String, +} + /// A "raw chunk received" notification, which is used to trigger /// the deprecated [RoomEvent:::StreamChunkReceived] event. pub struct ChunkReceived { pub chunk: Chunk, pub participant_identity: ParticipantIdentity, + + /// Topic of the stream this chunk belongs to, or `None` if the associated stream id could + /// not be mapped to a topic. + pub topic: Option, } /// A "raw trailer received" notification, which is used to trigger @@ -62,6 +88,12 @@ pub struct ChunkReceived { pub struct TrailerReceived { pub trailer: Trailer, pub participant_identity: ParticipantIdentity, + + /// Topic of the stream this chunk belongs to, or `None` if the associated stream id could + /// not be mapped to a topic. + /// + /// See [`ChunkReceived::topic`]. + pub topic: Option, } /// An event emitted by [`IncomingStreamManager::run`] for the host crate to surface. The manager @@ -69,6 +101,7 @@ pub struct TrailerReceived { #[derive(FromVariants)] pub enum OutputEvent { StreamOpened(StreamOpened), + StreamClosed(StreamClosed), ChunkReceived(ChunkReceived), TrailerReceived(TrailerReceived), } diff --git a/livekit-data-stream/src/incoming/manager.rs b/livekit-data-stream/src/incoming/manager.rs index af257b0e7..a018cc429 100644 --- a/livekit-data-stream/src/incoming/manager.rs +++ b/livekit-data-stream/src/incoming/manager.rs @@ -29,7 +29,8 @@ use crate::{ use super::{ events::{ - ChunkReceived, InputEvent, OutputEvent, PacketReceived, StreamOpened, TrailerReceived, + ChunkReceived, InputEvent, OutputEvent, PacketReceived, StreamClosed, StreamOpened, + TrailerReceived, }, stream_reader::AnyStreamReader, }; @@ -46,7 +47,9 @@ struct Descriptor { /// Identity of the participant sending this stream; used to abort the stream /// if that participant disconnects mid-send. sender_identity: ParticipantIdentity, - is_internal: bool, + /// Topic this stream was opened on, reported on chunk/trailer events so the host can recognize + /// streams it handles internally (chunk and trailer packets carry only a stream id). + topic: String, /// Whether this is a text stream (decompressed output is reframed on UTF-8 boundaries). is_text: bool, /// Per-stream deflate-raw decompressor; `Some` if the header declared `DEFLATE_RAW`. @@ -182,24 +185,18 @@ impl ManagerInput { pub struct Manager { inner: ManagerInner, input_rx: UnboundedReceiver, - output_tx: UnboundedSender, - - /// Topics whose streams are handled internally by the SDK (e.g. RPC) and never surfaced as - /// application events. Supplied by the host crate so this crate stays decoupled from RPC. - reserved_topics: Vec<&'static str>, /// Max number of bytes that a data stream can contain before it is deemed to be malicious max_payload_byte_length: usize, } -#[derive(Default)] struct ManagerInner { open_streams: HashMap, + output_tx: UnboundedSender, } impl Manager { pub fn new( - reserved_topics: Vec<&'static str>, max_payload_byte_length: Option, ) -> (Self, ManagerInput, UnboundedReceiver) { // Unbounded: inbound wire packets must never be dropped (a dropped chunk is an @@ -207,11 +204,9 @@ impl Manager { let (input_tx, input_rx) = mpsc::unbounded_channel(); let (output_tx, output_rx) = mpsc::unbounded_channel(); let manager = Self { - inner: ManagerInner::default(), + inner: ManagerInner { open_streams: HashMap::new(), output_tx }, input_rx, - output_tx, - reserved_topics, max_payload_byte_length: max_payload_byte_length .unwrap_or(DEFAULT_MAX_PAYLOAD_BYTE_LENGTH), }; @@ -238,6 +233,10 @@ impl Manager { } } InputEvent::AbortStreamsFrom(identity) => self.handle_abort(identity), + InputEvent::AbortAllStreams => self.handle_abort_all(), + InputEvent::QueryOpenStreamCount(respond_to) => { + let _ = respond_to.send(self.inner.open_streams.len()); + } InputEvent::Shutdown => break, } } @@ -250,7 +249,7 @@ impl Manager { participant_identity: ParticipantIdentity, encryption_type: EncryptionType, ) { - let is_internal = self.is_internal_topic(&header.topic); + let topic = header.topic.clone(); // A compression type from a future protocol version can't be decoded; drop the stream // (a conforming sender never sends compression a recipient didn't advertise support for, @@ -291,18 +290,20 @@ impl Manager { } let (stream_reader, chunk_tx, progress_tx) = AnyStreamReader::from(info); - let _ = self.output_tx.send( + let _ = self.inner.output_tx.send( StreamOpened { stream_reader, participant_identity: participant_identity.clone() } .into(), ); if bytes_total.is_some_and(|total| total > self.max_payload_byte_length as u64) { let _ = chunk_tx.send(Err(StreamError::PayloadTooLarge)); + self.inner.emit_stream_closed(&id, participant_identity, topic); return; } // Inline single-packet stream: synthesize the complete content now; no chunk/trailer - // packets will follow, so we never register an open descriptor. + // packets will follow, so we never register an open descriptor. Every path below + // terminates the stream, so each emits `StreamClosed` (there is no trailer to do it). if let Some(content) = inline_content { let content = if is_compressed { match inflate_raw(&content, self.max_payload_byte_length).await { @@ -311,12 +312,14 @@ impl Manager { // Defensive: a conforming sender never sends a compressed stream we // can't read, but drop gracefully if it happens. let _ = chunk_tx.send(Err(error)); + self.inner.emit_stream_closed(&id, participant_identity, topic); return; } } } else { if content.len() > self.max_payload_byte_length { let _ = chunk_tx.send(Err(StreamError::PayloadTooLarge)); + self.inner.emit_stream_closed(&id, participant_identity, topic); return; } content @@ -332,6 +335,7 @@ impl Manager { let _ = chunk_tx.send(Ok(Bytes::from(content))); } // Dropping `chunk_tx` closes the reader. + self.inner.emit_stream_closed(&id, participant_identity, topic); return; } @@ -341,7 +345,7 @@ impl Manager { progress_tx, encryption_type: stream_encryption_type, sender_identity: participant_identity, - is_internal, + topic, is_text, decompressor: is_compressed .then(|| DeflateDecompressState::new(self.max_payload_byte_length)), @@ -351,18 +355,12 @@ impl Manager { self.inner.open_streams.insert(id, descriptor); } - /// Returns whether a given streams is handled internally by the SDK - /// (e.g. `lk.rpc_request`) and associated events should not be surfaced to the application. - fn is_internal(&self, id: &StreamId) -> bool { - self.inner.open_streams.get(id).is_some_and(|d| d.is_internal) - } - - /// Returns whether streams created on the given topic are handled internally by the SDK - /// (e.g. `lk.rpc_request`) and should not be surfaced to the application. + /// Returns the topic of an open stream, or `None` if no stream with this id is open. /// - /// When possible, prefer [`Self::is_internal`] instead. - fn is_internal_topic(&self, topic: &str) -> bool { - self.reserved_topics.iter().any(|t| t == &topic) + /// Reported on chunk/trailer events so the host can apply its own topic policy (e.g. hiding + /// `lk.rpc_request`); this crate deliberately holds no notion of which topics are internal. + fn topic_associated_with_stream_id(&self, id: &StreamId) -> Option { + self.inner.open_streams.get(id).map(|d| d.topic.clone()) } /// Handles an incoming chunk packet. @@ -373,20 +371,23 @@ impl Manager { encryption_type: EncryptionType, ) { let id = chunk.stream_id.clone(); - if !self.is_internal(&id) { - let _ = self.output_tx.send(OutputEvent::ChunkReceived(ChunkReceived { - chunk: chunk.clone(), - participant_identity, - })); - } + let _ = self.inner.output_tx.send(OutputEvent::ChunkReceived(ChunkReceived { + chunk: chunk.clone(), + participant_identity, + topic: self.topic_associated_with_stream_id(&id), + })); let inner = &mut self.inner; let Some(descriptor) = inner.open_streams.get_mut(&id) else { return; }; - if descriptor.encryption_type != encryption_type.into() { - inner.close_stream_with_error(&id, StreamError::EncryptionTypeMismatch); + if descriptor.encryption_type != encryption_type { + let expected = descriptor.encryption_type; + inner.close_stream_with_error( + &id, + StreamError::EncryptionTypeMismatch { expected, received: encryption_type }, + ); return; } @@ -476,11 +477,14 @@ impl Manager { /// Handles an incoming trailer packet. fn handle_trailer(&mut self, trailer: Trailer, participant_identity: ParticipantIdentity) { let id = trailer.stream_id.clone(); - if !self.is_internal(&id) { - let _ = self - .output_tx - .send(TrailerReceived { trailer: trailer.clone(), participant_identity }.into()); - } + let _ = self.inner.output_tx.send( + TrailerReceived { + trailer: trailer.clone(), + participant_identity, + topic: self.topic_associated_with_stream_id(&id), + } + .into(), + ); let inner = &mut self.inner; let Some(descriptor) = inner.open_streams.get_mut(&id) else { @@ -526,6 +530,16 @@ impl Manager { } }); } + + /// Aborts every open stream, erroring each reader with [`StreamError::AbnormalEnd`]. Unlike + /// [`Self::handle_abort`] this isn't scoped to one participant; the host calls it when the + /// connection is torn down so no reader hangs waiting for chunks that will never arrive. + /// The run loop keeps going, so streams opened after (e.g. a reconnect) are still handled. + fn handle_abort_all(&mut self) { + self.inner.close_matching_streams_with_error(|_id, _descriptor| { + Err(StreamError::AbnormalEnd("Data stream connection closed".to_string())) + }); + } } impl ManagerInner { @@ -550,12 +564,15 @@ impl ManagerInner { fn close_stream(&mut self, id: &StreamId) { // Dropping the sender closes the channel. - self.open_streams.remove(id); + if let Some(descriptor) = self.open_streams.remove(id) { + self.emit_stream_closed(id, descriptor.sender_identity, descriptor.topic); + } } fn close_stream_with_error(&mut self, id: &StreamId, error: StreamError) { if let Some(descriptor) = self.open_streams.remove(id) { let _ = descriptor.chunk_tx.send(Err(error)); + self.emit_stream_closed(id, descriptor.sender_identity, descriptor.topic); } } @@ -563,14 +580,29 @@ impl ManagerInner { &mut self, checker: impl Fn(&StreamId, &Descriptor) -> Result<(), StreamError>, ) { - self.open_streams.retain(|id, descriptor| match checker(id, &descriptor) { + let Self { open_streams, output_tx } = self; + open_streams.retain(|id, descriptor| match checker(id, &descriptor) { Ok(_) => true, Err(error) => { let _ = descriptor.chunk_tx.send(Err(error)); + let _ = output_tx.send(OutputEvent::StreamClosed(StreamClosed { + stream_id: id.clone(), + participant_identity: descriptor.sender_identity.clone(), + topic: descriptor.topic.clone(), + })); false } }); } + + /// Announces that a stream previously announced via [`StreamOpened`] is terminated. + fn emit_stream_closed(&self, id: &StreamId, identity: ParticipantIdentity, topic: String) { + let _ = self.output_tx.send(OutputEvent::StreamClosed(StreamClosed { + stream_id: id.clone(), + participant_identity: identity, + topic, + })); + } } #[cfg(test)] @@ -681,16 +713,12 @@ mod tests { } impl Harness { - fn new(reserved_topics: Vec<&'static str>) -> Self { - Self::new_with_max_payload_length(reserved_topics, None) + fn new() -> Self { + Self::new_with_max_payload_length(None) } - fn new_with_max_payload_length( - reserved_topics: Vec<&'static str>, - max_payload_byte_length: Option, - ) -> Self { - let (manager, input, output_rx) = - Manager::new(reserved_topics, max_payload_byte_length); + fn new_with_max_payload_length(max_payload_byte_length: Option) -> Self { + let (manager, input, output_rx) = Manager::new(max_payload_byte_length); tokio::spawn(manager.run()); Self { input, output_rx } } @@ -711,6 +739,13 @@ mod tests { self.input.send(InputEvent::AbortStreamsFrom(identity)).unwrap(); } + /// Queries the number of currently open (descriptor-registered) streams. + async fn open_stream_count(&self) -> usize { + let (respond_to, response) = tokio::sync::oneshot::channel(); + self.input.send(InputEvent::QueryOpenStreamCount(respond_to)).unwrap(); + response.await.expect("the manager should answer the query") + } + /// Awaits the next opened stream's reader (skipping back-compat chunk/trailer outputs). async fn next_opened(&mut self) -> (AnyStreamReader, ParticipantIdentity) { loop { @@ -725,6 +760,22 @@ mod tests { } } } + + /// Awaits the next closed stream, returning its id, sender identity, and topic. + async fn next_closed(&mut self) -> (String, String, String) { + loop { + match self.output_rx.recv().await.expect("a stream should be closed") { + OutputEvent::StreamClosed(StreamClosed { + stream_id, + participant_identity, + topic, + }) => { + return (stream_id.to_string(), participant_identity.to_string(), topic); + } + _ => continue, + } + } + } } mod v1_legacy_multi_packet { @@ -732,7 +783,7 @@ mod tests { #[tokio::test] async fn v1_text_stream_round_trips() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "hello world"; h.send_packet(Packet::Header { header: text_header( @@ -757,7 +808,7 @@ mod tests { #[tokio::test] async fn v1_byte_stream_round_trips() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); h.send_packet(Packet::Header { header: byte_header("s1", Some(4), None, CompressionType::None), encryption_type: EncryptionType::None, @@ -773,7 +824,7 @@ mod tests { #[tokio::test] async fn v1_merges_trailer_attributes() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "hi"; h.send_packet(Packet::Header { header: text_header( @@ -803,7 +854,7 @@ mod tests { #[tokio::test] async fn v1_errors_when_too_few_bytes() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); h.send_packet(Packet::Header { header: text_header("s1", Some(5), HashMap::new(), None, CompressionType::None), encryption_type: EncryptionType::None, @@ -819,7 +870,7 @@ mod tests { #[tokio::test] async fn v1_errors_when_too_many_bytes() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); h.send_packet(Packet::Header { header: byte_header("s1", Some(3), None, CompressionType::None), encryption_type: EncryptionType::None, @@ -836,7 +887,7 @@ mod tests { #[tokio::test] async fn v1_max_payload_size_breached_with_unknown_total() { // A stream with no declared total must still be bounded by the receiver's cap. - let mut h = Harness::new_with_max_payload_length(vec![], Some(1_000)); + let mut h = Harness::new_with_max_payload_length(Some(1_000)); h.send_packet(Packet::Header { header: byte_header("s1", None, None, CompressionType::None), encryption_type: EncryptionType::None, @@ -854,7 +905,7 @@ mod tests { #[tokio::test] async fn v1_max_payload_size_fast_fails_on_declared_total() { // A header declaring a total above the cap is rejected before any chunks arrive. - let mut h = Harness::new_with_max_payload_length(vec![], Some(1_000)); + let mut h = Harness::new_with_max_payload_length(Some(1_000)); h.send_packet(Packet::Header { header: byte_header("s1", Some(2_000), None, CompressionType::None), encryption_type: EncryptionType::None, @@ -866,7 +917,7 @@ mod tests { #[tokio::test] async fn v1_payload_exactly_at_max_payload_size_succeeds() { // The cap is inclusive: a payload of exactly max_payload_byte_length is accepted. - let mut h = Harness::new_with_max_payload_length(vec![], Some(1_000)); + let mut h = Harness::new_with_max_payload_length(Some(1_000)); h.send_packet(Packet::Header { header: byte_header("s1", Some(1_000), None, CompressionType::None), encryption_type: EncryptionType::None, @@ -882,7 +933,7 @@ mod tests { #[tokio::test] async fn v1_drops_on_encryption_type_mismatch() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); h.send_packet(Packet::Header { header: text_header("s1", Some(2), HashMap::new(), None, CompressionType::None), encryption_type: EncryptionType::None, @@ -892,12 +943,18 @@ mod tests { chunk: chunk("s1", 0, vec![b'h', b'i']), encryption_type: EncryptionType::Gcm, }); - assert!(matches!(read_text(reader).await, Err(StreamError::EncryptionTypeMismatch))); + assert!(matches!( + read_text(reader).await, + Err(StreamError::EncryptionTypeMismatch { + expected: EncryptionType::None, + received: EncryptionType::Gcm, + }) + )); } #[tokio::test] async fn v1_trailer_attributes_merged_after_close() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "hello world"; h.send_packet(Packet::Header { header: text_header( @@ -934,7 +991,7 @@ mod tests { #[tokio::test] async fn v2_inline_uncompressed_text() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "inline hello"; h.send_packet(Packet::Header { header: text_header( @@ -954,7 +1011,7 @@ mod tests { #[tokio::test] async fn v2_inline_uncompressed_byte() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); h.send_packet(Packet::Header { header: byte_header("s1", Some(3), Some(vec![1, 2, 3]), CompressionType::None), encryption_type: EncryptionType::None, @@ -965,7 +1022,7 @@ mod tests { #[tokio::test] async fn v2_inline_compressed_text() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "hello hello compressible world"; let compressed = deflate_raw(text.as_bytes()).await; h.send_packet(Packet::Header { @@ -985,7 +1042,7 @@ mod tests { #[tokio::test] async fn v2_inline_compressed_byte() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let payload: Vec = (0..2000).map(|i| (i % 7) as u8).collect(); let compressed = deflate_raw(&payload).await; h.send_packet(Packet::Header { @@ -1005,7 +1062,7 @@ mod tests { async fn v2_inline_compressed_max_payload_size_breached() { // A tiny compressed inline payload that inflates far past the configured cap must be // rejected (decompression-bomb guard on the inline path). - let mut h = Harness::new_with_max_payload_length(vec![], Some(1_000)); + let mut h = Harness::new_with_max_payload_length(Some(1_000)); let text = pseudo_random_text(50_000); let compressed = deflate_raw(text.as_bytes()).await; h.send_packet(Packet::Header { @@ -1026,7 +1083,7 @@ mod tests { async fn v2_inline_uncompressed_max_payload_size_breached() { // The cap applies to uncompressed inline payloads too. No declared total, so the // inline content check (not the header fast-fail) is what trips. - let mut h = Harness::new_with_max_payload_length(vec![], Some(1_000)); + let mut h = Harness::new_with_max_payload_length(Some(1_000)); h.send_packet(Packet::Header { header: byte_header("s1", None, Some(vec![0u8; 2_000]), CompressionType::None), encryption_type: EncryptionType::None, @@ -1037,7 +1094,7 @@ mod tests { #[tokio::test] async fn v2_inline_zero_length_text() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); h.send_packet(Packet::Header { header: text_header( "s1", @@ -1060,7 +1117,7 @@ mod tests { #[tokio::test] async fn v2_multipacket_compressed_text() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); // ~60 KB of pseudo-random lowercase so the compressed output spans multiple chunks. let text = pseudo_random_text(60_000); let compressed = deflate_raw(text.as_bytes()).await; @@ -1090,7 +1147,7 @@ mod tests { #[tokio::test] async fn errors_open_streams_on_sender_disconnect() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); h.send_packet(Packet::Header { header: text_header("s1", Some(10), HashMap::new(), None, CompressionType::None), encryption_type: EncryptionType::None, @@ -1107,7 +1164,7 @@ mod tests { #[tokio::test] async fn abort_only_affects_matching_sender() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); h.send_packet_from( Packet::Header { header: text_header("s1", Some(5), HashMap::new(), None, CompressionType::None), @@ -1131,7 +1188,7 @@ mod tests { #[tokio::test] async fn v2_compressed_gap_errors() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = pseudo_random_text(60_000); let compressed = deflate_raw(text.as_bytes()).await; let pieces: Vec<&[u8]> = compressed.chunks(15_000).collect(); @@ -1165,8 +1222,7 @@ mod tests { let compressed = deflate_raw(text.as_bytes()).await; // Use a max payload size one byte below the size of the compressed data - let mut h = - Harness::new_with_max_payload_length(vec![], Some(50_000 /* less than 60k */)); + let mut h = Harness::new_with_max_payload_length(Some(50_000 /* less than 60k */)); // Feed all data in h.send_packet(Packet::Header { @@ -1193,7 +1249,7 @@ mod tests { #[tokio::test] async fn v2_multipacket_compressed_byte_stream() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let data = pseudo_random_text(60_000).into_bytes(); let compressed = deflate_raw(&data).await; let pieces: Vec<&[u8]> = compressed.chunks(15_000).collect(); @@ -1221,7 +1277,7 @@ mod tests { #[tokio::test] async fn v2_compressed_errors_when_too_few_bytes() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "hello world"; // 11 bytes decompressed let compressed = deflate_raw(text.as_bytes()).await; h.send_packet(Packet::Header { @@ -1246,7 +1302,7 @@ mod tests { #[tokio::test] async fn v2_compressed_errors_when_too_many_bytes() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "hello world"; // 11 bytes decompressed let compressed = deflate_raw(text.as_bytes()).await; h.send_packet(Packet::Header { @@ -1270,7 +1326,7 @@ mod tests { #[tokio::test] async fn v2_compressed_duplicate_chunk_dropped() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = pseudo_random_text(60_000); let compressed = deflate_raw(text.as_bytes()).await; let pieces: Vec<&[u8]> = compressed.chunks(15_000).collect(); @@ -1309,7 +1365,7 @@ mod tests { #[tokio::test] async fn v2_compressed_text_reframes_multibyte_utf8() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "πŸ˜€δ½ ε₯½δΈ–η•Œ cafΓ© β€” ‘ñandΓΊ! ".repeat(500); let compressed = deflate_raw(text.as_bytes()).await; // Split the compressed bytes at an arbitrary midpoint: the decompressor's output at the @@ -1341,7 +1397,7 @@ mod tests { #[tokio::test] async fn v2_unknown_compression_type_is_ignored() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); // A compression type from a future protocol version arrives at the proto layer; the // receiver can't decode it, so per the spec's defensive-drop behavior (mirroring the web // SDK) the stream must be ignored rather than delivered as if uncompressed. @@ -1380,7 +1436,7 @@ mod tests { #[tokio::test] async fn v2_compressed_merges_trailer_attributes() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "hello world"; let compressed = deflate_raw(text.as_bytes()).await; h.send_packet(Packet::Header { @@ -1448,7 +1504,7 @@ mod tests { #[tokio::test] async fn progress_reports_completion_uncompressed_bytes() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let total = 12u64; h.send_packet(Packet::Header { header: byte_header("s1", Some(total), None, CompressionType::None), @@ -1475,7 +1531,7 @@ mod tests { #[tokio::test] async fn progress_reports_completion_compressed_text() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = pseudo_random_text(60_000); let total = text.len() as u64; let compressed = deflate_raw(text.as_bytes()).await; @@ -1509,7 +1565,7 @@ mod tests { #[tokio::test] async fn progress_reports_completion_inline() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "inline hello"; let total = text.len() as u64; h.send_packet(Packet::Header { @@ -1530,9 +1586,167 @@ mod tests { } } + /// The open-stream count reflects descriptor-registered streams, answered in order with the + /// events enqueued before the query β€” tests use it to wait for a header/abort to land. + mod open_stream_count { + use super::*; + + #[tokio::test] + async fn counts_streams_across_their_lifecycle() { + let mut h = Harness::new(); + assert_eq!(h.open_stream_count().await, 0); + + h.send_packet(Packet::Header { + header: text_header("s1", Some(5), HashMap::new(), None, CompressionType::None), + encryption_type: EncryptionType::None, + }); + assert_eq!(h.open_stream_count().await, 1); + + h.send_packet(Packet::Header { + header: text_header("s2", Some(5), HashMap::new(), None, CompressionType::None), + encryption_type: EncryptionType::None, + }); + assert_eq!(h.open_stream_count().await, 2); + + // Keep the readers alive so the streams aren't closed by reader drop. + let (reader1, _) = h.next_opened().await; + let (reader2, _) = h.next_opened().await; + + h.send_packet(Packet::Chunk { + chunk: chunk("s1", 0, b"hello".to_vec()), + encryption_type: EncryptionType::None, + }); + h.send_packet(Packet::Trailer(trailer("s1"))); + assert_eq!(h.open_stream_count().await, 1); + + h.abort(ParticipantIdentity::from(SENDER)); + assert_eq!(h.open_stream_count().await, 0); + drop((reader1, reader2)); + } + + #[tokio::test] + async fn inline_streams_are_never_counted() { + let mut h = Harness::new(); + h.send_packet(Packet::Header { + header: text_header( + "s1", + Some(5), + HashMap::new(), + Some(b"hello".to_vec()), + CompressionType::None, + ), + encryption_type: EncryptionType::None, + }); + let (reader, _) = h.next_opened().await; + // The inline stream completes during header handling; no descriptor is registered. + assert_eq!(h.open_stream_count().await, 0); + assert_eq!(read_text(reader).await.unwrap(), "hello"); + } + } + + /// Every opened stream terminates with exactly one `StreamClosed`, whatever the terminal + /// path β€” hosts rely on it to sequence handler invocations for ordered topics. + mod stream_closed { + use super::*; + + #[tokio::test] + async fn trailer_close_emits_stream_closed() { + let mut h = Harness::new(); + let text = "hello world"; + h.send_packet(Packet::Header { + header: text_header( + "s1", + Some(text.len() as u64), + HashMap::new(), + None, + CompressionType::None, + ), + encryption_type: EncryptionType::None, + }); + let (reader, _) = h.next_opened().await; + h.send_packet(Packet::Chunk { + chunk: chunk("s1", 0, text.as_bytes().to_vec()), + encryption_type: EncryptionType::None, + }); + h.send_packet(Packet::Trailer(trailer("s1"))); + assert_eq!(h.next_closed().await, ("s1".into(), SENDER.into(), "topic".into())); + assert_eq!(read_text(reader).await.unwrap(), text); + } + + #[tokio::test] + async fn inline_stream_emits_stream_closed() { + // Inline single-packet streams never receive a trailer, so the closed signal must be + // synthesized when the inline payload completes. + let mut h = Harness::new(); + h.send_packet(Packet::Header { + header: text_header( + "s1", + Some(5), + HashMap::new(), + Some(b"hello".to_vec()), + CompressionType::None, + ), + encryption_type: EncryptionType::None, + }); + let (reader, _) = h.next_opened().await; + assert_eq!(h.next_closed().await, ("s1".into(), SENDER.into(), "topic".into())); + assert_eq!(read_text(reader).await.unwrap(), "hello"); + } + + #[tokio::test] + async fn error_close_emits_stream_closed() { + let mut h = Harness::new(); + h.send_packet(Packet::Header { + header: text_header("s1", Some(10), HashMap::new(), None, CompressionType::None), + encryption_type: EncryptionType::None, + }); + let (reader, _) = h.next_opened().await; + // A chunk-index gap closes the stream with `MissedChunk`. + h.send_packet(Packet::Chunk { + chunk: chunk("s1", 5, b"hello".to_vec()), + encryption_type: EncryptionType::None, + }); + assert_eq!(h.next_closed().await, ("s1".into(), SENDER.into(), "topic".into())); + assert!(matches!(read_text(reader).await, Err(StreamError::MissedChunk))); + } + + #[tokio::test] + async fn abort_emits_stream_closed() { + let mut h = Harness::new(); + h.send_packet(Packet::Header { + header: text_header("s1", Some(10), HashMap::new(), None, CompressionType::None), + encryption_type: EncryptionType::None, + }); + let (reader, _) = h.next_opened().await; + h.abort(ParticipantIdentity::from(SENDER)); + assert_eq!(h.next_closed().await, ("s1".into(), SENDER.into(), "topic".into())); + assert!(matches!(read_text(reader).await, Err(StreamError::AbnormalEnd(_)))); + } + + #[tokio::test] + async fn trailer_for_unopened_stream_emits_no_stream_closed() { + let mut h = Harness::new(); + h.send_packet(Packet::Trailer(trailer("never-opened"))); + // A second, well-formed inline stream: if the orphan trailer had produced a closed + // event, it would be observed before this stream's. + h.send_packet(Packet::Header { + header: text_header( + "s2", + Some(2), + HashMap::new(), + Some(b"hi".to_vec()), + CompressionType::None, + ), + encryption_type: EncryptionType::None, + }); + let (closed_id, _, _) = h.next_closed().await; + assert_eq!(closed_id, "s2"); + } + } + #[tokio::test] async fn empty_chunks_are_ignored() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "hello world"; h.send_packet(Packet::Header { header: text_header( @@ -1560,7 +1774,7 @@ mod tests { #[tokio::test] async fn trailer_with_reason_errors_abnormal_end() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); h.send_packet(Packet::Header { header: text_header("s1", Some(5), HashMap::new(), None, CompressionType::None), encryption_type: EncryptionType::None, @@ -1582,7 +1796,7 @@ mod tests { #[tokio::test] async fn text_stream_with_attachments_round_trips() { - let mut h = Harness::new(vec![]); + let mut h = Harness::new(); let text = "hello world"; // Text stream whose header references an attachment stream id, body inline. @@ -1619,4 +1833,60 @@ mod tests { h.send_packet(Packet::Trailer(trailer("att1"))); assert_eq!(read_bytes(byte_reader).await.unwrap(), Bytes::from(vec![1u8, 2, 3])); } + + /// Chunk and trailer packets carry only a stream id, so the manager reports the topic of the + /// stream they belong to. Hosts rely on this to filter events for topics they handle + /// internally (e.g. RPC), so it is the only signal available to them for these two events. + mod reported_topic { + use super::*; + + /// Awaits the next chunk/trailer output, returning the topic it reported. + async fn next_raw_topic(h: &mut Harness) -> Option { + loop { + match h.output_rx.recv().await.expect("an output event should be emitted") { + OutputEvent::ChunkReceived(ChunkReceived { topic, .. }) + | OutputEvent::TrailerReceived(TrailerReceived { topic, .. }) => { + return topic; + } + _ => continue, + } + } + } + + #[tokio::test] + async fn chunk_and_trailer_report_the_topic_of_their_stream() { + let mut h = Harness::new(); + h.send_packet(Packet::Header { + header: Header { + topic: "lk.rpc_request".to_string(), + ..text_header("s1", Some(2), HashMap::new(), None, CompressionType::None) + }, + encryption_type: EncryptionType::None, + }); + let (reader, _) = h.next_opened().await; + + h.send_packet(Packet::Chunk { + chunk: chunk("s1", 0, b"hi".to_vec()), + encryption_type: EncryptionType::None, + }); + assert_eq!(next_raw_topic(&mut h).await.as_deref(), Some("lk.rpc_request")); + + h.send_packet(Packet::Trailer(trailer("s1"))); + assert_eq!(next_raw_topic(&mut h).await.as_deref(), Some("lk.rpc_request")); + + // The stream itself still opens and reads normally; reporting the topic does not + // suppress anything in this crate. + assert_eq!(read_text(reader).await.unwrap(), "hi"); + } + + #[tokio::test] + async fn chunk_for_an_unopened_stream_reports_no_topic() { + let mut h = Harness::new(); + h.send_packet(Packet::Chunk { + chunk: chunk("never-opened", 0, b"hi".to_vec()), + encryption_type: EncryptionType::None, + }); + assert_eq!(next_raw_topic(&mut h).await, None); + } + } } diff --git a/livekit-data-stream/src/outgoing/manager.rs b/livekit-data-stream/src/outgoing/manager.rs index 0764b5593..48e1cd00e 100644 --- a/livekit-data-stream/src/outgoing/manager.rs +++ b/livekit-data-stream/src/outgoing/manager.rs @@ -43,12 +43,15 @@ fn create_random_uuid() -> String { #[derive(Clone)] pub struct Manager { - /// Request channel for sending packets. - packet_tx: UnboundedRequestSender>, + /// Request channel for sending packet batches. Each request is an ordered batch the + /// transport acknowledges as a whole: one-shot sends (`send_text`/`send_bytes`) emit their + /// entire stream as a single request, while every other call site sends one packet at a time. + packet_tx: UnboundedRequestSender, Result<(), SendError>>, } impl Manager { - pub fn new() -> (Self, UnboundedRequestReceiver>) { + pub fn new() -> (Self, UnboundedRequestReceiver, Result<(), SendError>>) + { let (packet_tx, packet_rx) = bmrng::unbounded_channel(); let manager = Self { packet_tx }; (manager, packet_rx) @@ -159,27 +162,30 @@ impl Manager { return Ok(TextStreamInfo::from_headers(header, text_header)); } - // 2/3. Chunked, compressed when eligible else uncompressed. + // 2/3. Chunked, compressed when eligible else uncompressed. The entire stream β€” header, + // chunks, trailer β€” goes out as one transport request. header.inline_content = None; enforce_header_size(&header, &options.destination_identities)?; - let open_options = RawStreamOpenOptions { - header: header.clone(), - destination_identities: options.destination_identities, - sender_identity: options.sender_identity, - packet_tx: self.packet_tx.clone(), - }; - let info = TextStreamInfo::from_headers(header, text_header); - let mut stream = RawStream::open(open_options).await?; + let info = TextStreamInfo::from_headers(header.clone(), text_header); if use_compression { let compressed_bytes = maybe_compressed.as_bytes().await?; - stream.write_raw_chunks(compressed_bytes).await?; + self.send_one_shot_stream( + header, + compressed_bytes.chunks(constants::STREAM_CHUNK_SIZE_BYTES), + options.destination_identities, + options.sender_identity, + ) + .await?; } else { - for chunk in text_bytes.utf8_aware_chunks(constants::STREAM_CHUNK_SIZE_BYTES) { - stream.write_chunk(chunk).await?; - } + self.send_one_shot_stream( + header, + text_bytes.utf8_aware_chunks(constants::STREAM_CHUNK_SIZE_BYTES), + options.destination_identities, + options.sender_identity, + ) + .await?; } - stream.close(None, None).await?; Ok(info) } @@ -253,25 +259,20 @@ impl Manager { return Ok(ByteStreamInfo::from_headers(header, byte_header)); } - // 2/3. Chunked, compressed when eligible else uncompressed. + // 2/3. Chunked, compressed when eligible else uncompressed. The entire stream β€” header, + // chunks, trailer β€” goes out as one transport request. header.inline_content = None; enforce_header_size(&header, &options.destination_identities)?; - let open_options = RawStreamOpenOptions { - header: header.clone(), - destination_identities: options.destination_identities, - sender_identity: options.sender_identity, - packet_tx: self.packet_tx.clone(), - }; - let info = ByteStreamInfo::from_headers(header, byte_header); - let mut stream = RawStream::open(open_options).await?; - if use_compression { - let compressed_bytes = maybe_compressed.as_bytes().await?; - stream.write_raw_chunks(compressed_bytes).await?; - } else { - stream.write_raw_chunks(bytes).await?; - } - stream.close(None, None).await?; + let info = ByteStreamInfo::from_headers(header.clone(), byte_header); + let content = if use_compression { maybe_compressed.as_bytes().await? } else { bytes }; + self.send_one_shot_stream( + header, + content.chunks(constants::STREAM_CHUNK_SIZE_BYTES), + options.destination_identities, + options.sender_identity, + ) + .await?; Ok(info) } @@ -318,6 +319,36 @@ impl Manager { stream.close(None, None).await?; Ok(info) } + + /// Sends a complete one-shot stream β€” header, pre-split content chunks, trailer β€” as a + /// single transport request acknowledged as a whole, rather than one request per packet. + /// + /// Only for sends whose full content is already in memory (`send_text`/`send_bytes`); + /// incremental writers and `send_file` stream packet-by-packet instead. + async fn send_one_shot_stream<'a>( + &self, + header: Header, + chunks: impl IntoIterator, + destination_identities: Vec, + sender_identity: Option, + ) -> StreamResult<()> { + let stream_id = header.stream_id.to_string(); + let mut packets = + vec![RawStream::create_header_packet(header.into(), destination_identities)]; + packets.extend( + chunks.into_iter().enumerate().map(|(index, chunk)| { + RawStream::create_chunk_packet(&stream_id, index as u64, chunk) + }), + ); + packets.push(RawStream::create_trailer_packet(&stream_id, None, None)); + if let Some(sender_identity) = sender_identity { + let identity: String = sender_identity.into(); + for packet in &mut packets { + packet.participant_identity = identity.clone(); + } + } + RawStream::send_packets(&self.packet_tx, packets).await + } } /// Inline / compression eligibility evaluated over a send's recipients. @@ -537,14 +568,29 @@ mod tests { // --- Capture harness ----------------------------------------------------------------- type Sent = Arc>>; + type SentBatches = Arc>>>; fn setup() -> (Manager, Sent) { let (manager, mut packet_rx) = Manager::new(); let sent: Sent = Arc::new(StdMutex::new(Vec::new())); let sink = sent.clone(); tokio::spawn(async move { - while let Ok((packet, responder)) = packet_rx.recv().await { - sink.lock().unwrap().push(packet); + while let Ok((packets, responder)) = packet_rx.recv().await { + sink.lock().unwrap().extend(packets); + let _ = responder.respond(Ok(())); + } + }); + (manager, sent) + } + + /// Like [`setup`], but records the batch boundaries of each transport request. + fn setup_batched() -> (Manager, SentBatches) { + let (manager, mut packet_rx) = Manager::new(); + let sent: SentBatches = Arc::new(StdMutex::new(Vec::new())); + let sink = sent.clone(); + tokio::spawn(async move { + while let Ok((packets, responder)) = packet_rx.recv().await { + sink.lock().unwrap().push(packets); let _ = responder.respond(Ok(())); } }); @@ -1263,7 +1309,7 @@ mod tests { let raw_stream = rt.block_on(async { let (packet_tx, mut packet_rx) = - bmrng::unbounded_channel::>(); + bmrng::unbounded_channel::, Result<(), SendError>>(); tokio::spawn(async move { while let Ok((_packet, responder)) = packet_rx.recv().await { @@ -1299,6 +1345,78 @@ mod tests { drop_thread.join().expect("Dropping RawStream on a non-Tokio thread must not panic"); } + // --- Batching --------------------------------------------------------------------------- + + mod packet_batching { + use super::*; + + #[tokio::test] + async fn one_shot_send_text_is_a_single_transport_request() { + let (m, sent) = setup_batched(); + // 40 KB uncompressed to a pre-v2 room: header + 3 chunks (15k/15k/10k) + trailer, + // delivered as ONE transport request rather than one per packet. + let text = "A".repeat(40_000); + m.send_text(&text, text_opts("chat", &[]), &pre_v2_room()).await.unwrap(); + let batches = sent.lock().unwrap().clone(); + assert_eq!(batches.iter().map(Vec::len).collect::>(), vec![5]); + let batch = &batches[0]; + assert!(matches!(batch[0].value, Some(proto::data_packet::Value::StreamHeader(_)))); + for (i, packet) in batch[1..4].iter().enumerate() { + assert_eq!(chunk(packet).chunk_index, i as u64); + } + assert_trailer(&batch[4]); + } + + #[tokio::test] + async fn one_shot_send_bytes_is_a_single_transport_request() { + let (m, sent) = setup_batched(); + let payload = vec![0x07u8; 40_000]; + let opts = byte_opts("blob", &["alice", "bob"]).with_compress(false); + m.send_bytes(&payload, opts, &all_v2_room()).await.unwrap(); + let batches = sent.lock().unwrap().clone(); + assert_eq!(batches.iter().map(Vec::len).collect::>(), vec![5]); + } + + #[tokio::test] + async fn one_shot_send_with_sender_identity_stamps_every_packet() { + let (m, sent) = setup_batched(); + let opts = text_opts("chat", &[]).with_sender_identity("impostor"); + m.send_text(&"A".repeat(20_000), opts, &pre_v2_room()).await.unwrap(); + let batches = sent.lock().unwrap().clone(); + assert_eq!(batches.iter().map(Vec::len).collect::>(), vec![4]); + assert!(batches[0].iter().all(|pkt| pkt.participant_identity == "impostor")); + } + + #[tokio::test] + async fn incremental_writer_sends_per_write() { + let (m, sent) = setup_batched(); + let writer = m.stream_text(text_opts("chat", &[])).await.unwrap(); + writer.write("hello").await.unwrap(); + writer.write("world").await.unwrap(); + writer.close().await.unwrap(); + // Incremental writes are flushed as they happen β€” never coalesced across writes. + let batches = sent.lock().unwrap().clone(); + assert_eq!(batches.iter().map(Vec::len).collect::>(), vec![1, 1, 1, 1]); + } + + #[tokio::test] + async fn send_file_streams_one_packet_per_request() { + // send_file deliberately never buffers the whole file, so it keeps per-packet + // requests instead of the one-shot batch. + let (m, sent) = setup_batched(); + let path = + std::env::temp_dir().join(format!("lk_ds_batch_{}.bin", create_random_uuid())); + tokio::fs::write(&path, vec![0x07u8; 20_000]).await.unwrap(); + m.send_file(&path, byte_opts("file", &[]).with_compress(false), &all_v2_room()) + .await + .unwrap(); + let _ = tokio::fs::remove_file(&path).await; + let batches = sent.lock().unwrap().clone(); + // Header + 15k chunk + 5k chunk + trailer, each its own request. + assert_eq!(batches.iter().map(Vec::len).collect::>(), vec![1, 1, 1, 1]); + } + } + // --- Additional spec-conformance cases ------------------------------------------------ mod stream_text_bytes { diff --git a/livekit-data-stream/src/outgoing/raw_stream.rs b/livekit-data-stream/src/outgoing/raw_stream.rs index 8f9055fe5..c75055f7a 100644 --- a/livekit-data-stream/src/outgoing/raw_stream.rs +++ b/livekit-data-stream/src/outgoing/raw_stream.rs @@ -30,7 +30,7 @@ pub(crate) struct RawStreamOpenOptions { /// Identity the stream's packets are attributed to; empty means the server attributes /// them to the sending participant. pub(crate) sender_identity: Option, - pub(crate) packet_tx: UnboundedRequestSender>, + pub(crate) packet_tx: UnboundedRequestSender, Result<(), SendError>>, } pub(crate) struct RawStream { @@ -38,8 +38,8 @@ pub(crate) struct RawStream { sender_identity: Option, progress: StreamProgress, is_closed: bool, - /// Request channel for sending packets. - packet_tx: UnboundedRequestSender>, + /// Request channel for sending packet batches. + packet_tx: UnboundedRequestSender, Result<(), SendError>>, } impl RawStream { @@ -63,18 +63,28 @@ impl RawStream { }) } + pub(crate) fn is_closed(&self) -> bool { + self.is_closed + } + pub(crate) async fn write_chunk(&mut self, bytes: &[u8]) -> StreamResult<()> { let mut packet = Self::create_chunk_packet(&self.id, self.progress.chunk_index, bytes); if let Some(sender_identity) = self.sender_identity.as_ref() { packet.participant_identity = sender_identity.clone().into(); } - Self::send_packet(&self.packet_tx, packet).await?; + if let Err(error) = Self::send_packet(&self.packet_tx, packet).await { + // A failed send makes the stream unusable; mark it closed so readers/writers stop + // treating it as open. + self.is_closed = true; + return Err(error); + } self.progress.bytes_processed += bytes.len() as u64; self.progress.chunk_index += 1; Ok(()) } - /// Writes opaque bytes split into MTU-sized chunks on raw byte boundaries. + /// Writes opaque bytes split into MTU-sized chunks on raw byte boundaries, one transport + /// request per chunk. /// /// Used for byte payloads and for compressed (deflate-raw) content, where the bytes /// are opaque and must not be split on UTF-8 boundaries. @@ -155,16 +165,25 @@ impl RawStream { if let Some(sender_identity) = self.sender_identity.as_ref() { packet.participant_identity = sender_identity.clone().into(); } - Self::send_packet(&self.packet_tx, packet).await?; + // The stream is done after a close attempt regardless of whether the trailer send succeeds. self.is_closed = true; + Self::send_packet(&self.packet_tx, packet).await?; Ok(()) } pub(crate) async fn send_packet( - tx: &UnboundedRequestSender>, + tx: &UnboundedRequestSender, Result<(), SendError>>, packet: proto::DataPacket, ) -> StreamResult<()> { - tx.send_receive(packet) + Self::send_packets(tx, vec![packet]).await + } + + /// Sends a batch of packets as a single transport request, acknowledged as a whole. + pub(crate) async fn send_packets( + tx: &UnboundedRequestSender, Result<(), SendError>>, + packets: Vec, + ) -> StreamResult<()> { + tx.send_receive(packets) .await .map_err(|_| StreamError::Internal)? // request channel closed .map_err(|_| StreamError::SendFailed) // data channel error diff --git a/livekit-data-stream/src/outgoing/stream_writer.rs b/livekit-data-stream/src/outgoing/stream_writer.rs index fa2165e7e..c5a823f30 100644 --- a/livekit-data-stream/src/outgoing/stream_writer.rs +++ b/livekit-data-stream/src/outgoing/stream_writer.rs @@ -68,6 +68,11 @@ impl ByteStreamWriter { pub(crate) fn new(info: Arc, stream: Arc>) -> Self { Self { info, stream } } + + /// Whether the stream has been closed β€” either locally (via `close`) or because a send failed. + pub async fn is_closed(&self) -> bool { + self.stream.lock().await.is_closed() + } } #[derive(Clone)] @@ -81,6 +86,11 @@ impl TextStreamWriter { pub(crate) fn new(info: Arc, stream: Arc>) -> Self { Self { info, stream } } + + /// Whether the stream has been closed β€” either locally (via `close`) or because a send failed. + pub async fn is_closed(&self) -> bool { + self.stream.lock().await.is_closed() + } } impl<'a> StreamWriter<'a> for ByteStreamWriter { diff --git a/livekit-data-stream/src/utils.rs b/livekit-data-stream/src/utils.rs index bb1a49c12..e1efebe3d 100644 --- a/livekit-data-stream/src/utils.rs +++ b/livekit-data-stream/src/utils.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use livekit_common::EncryptionType; use thiserror::Error; /// Error returned by the packet transport when a data-stream packet fails to send. @@ -59,8 +60,13 @@ pub enum StreamError { #[error("internal error")] Internal, - #[error("encryption type mismatch")] - EncryptionTypeMismatch, + #[error("encryption type mismatch: expected {expected:?}, received {received:?}")] + EncryptionTypeMismatch { + /// The encryption type the stream's header declared. + expected: EncryptionType, + /// The encryption type of the offending chunk. + received: EncryptionType, + }, #[error("stream header exceeds maximum size")] HeaderTooLarge, diff --git a/livekit-datatrack/uniffi.toml b/livekit-datatrack/uniffi.toml index e138669b9..5148c72db 100644 --- a/livekit-datatrack/uniffi.toml +++ b/livekit-datatrack/uniffi.toml @@ -1,2 +1,8 @@ [bindings.swift] ffi_module_name = "RustLiveKitDataTrack" + +[bindings.kotlin] +# The Kotlin checksum test is off because it fails on ARM devices. uniffi 0.31.2 and 0.32.0 do not +# mask the upper 16 bits of the checksum. For more info, see the similar comment in the root +# Cargo.toml. +omit_checksums = true diff --git a/livekit-uniffi/Cargo.toml b/livekit-uniffi/Cargo.toml index f8a915ee4..51846f4ea 100644 --- a/livekit-uniffi/Cargo.toml +++ b/livekit-uniffi/Cargo.toml @@ -15,6 +15,8 @@ publish = false livekit-protocol = { workspace = true } livekit-api = { workspace = true, default-features = false, features = ["access-token"] } livekit-datatrack = { workspace = true, features = ["uniffi"] } +livekit-data-stream = { workspace = true } +livekit-common = { workspace = true } uniffi = { workspace = true, features = ["scaffolding-ffi-buffer-fns", "tokio"] } log = { workspace = true } tokio = { workspace = true, features = ["sync", "rt-multi-thread"] } @@ -27,7 +29,13 @@ thiserror = { workspace = true } # Dart binding generator. Not published to crates.io, so pinned by git rev. The # rev must target the same uniffi-rs release (0.31) as the `uniffi` dependency # above, or it cannot read this crate's compiled metadata. -uniffi-dart = { git = "https://github.com/Uniffi-Dart/uniffi-dart", rev = "90f2c6f29cbf88c8bc2cf515e6a0c2314a48844c", optional = true } +# +# Temporarily a fork: upstream cannot generate compiling bindings for this crate +# -- custom types leak into FFI signatures, object converter statics collide +# with Rust methods named `write`, and each crate re-declares the runtime +# scaffolding, so livekit-datatrack's types will not cross into livekit-uniffi. +# Point back at Uniffi-Dart/uniffi-dart once those fixes are upstreamed. +uniffi-dart = { git = "https://github.com/1egoman/uniffi-dart", rev = "4633a7d3c93186ac6a96007b5f109268b4746190", optional = true } camino = { version = "1", optional = true } [features] diff --git a/livekit-uniffi/src/data_stream/common.rs b/livekit-uniffi/src/data_stream/common.rs new file mode 100644 index 000000000..0cd23d2db --- /dev/null +++ b/livekit-uniffi/src/data_stream/common.rs @@ -0,0 +1,416 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Types that cross the FFI boundary for data streams: info/option records, enums, the error +//! wrapper, and the wire-packet decode helper. +//! +//! `Bytes` is already registered as a custom type by [`crate::data_track::common`]; it is reused +//! here rather than redefined (a second `custom_type!` in the same crate would conflict). +//! Participant identities cross as plain `String`. + +use std::collections::HashMap; + +use livekit_common as common; +use livekit_data_stream::{api as ds_api, backend as ds}; +use livekit_protocol as proto; +use prost::Message; + +// MARK: - Enums + +/// Encryption applied to a data stream, mirroring [`common::EncryptionType`]. +#[derive(uniffi::Enum, Clone, Copy, Debug, PartialEq, Eq)] +pub enum EncryptionType { + None, + Gcm, + Custom, +} + +impl From for EncryptionType { + fn from(value: common::EncryptionType) -> Self { + match value { + common::EncryptionType::None => Self::None, + common::EncryptionType::Gcm => Self::Gcm, + common::EncryptionType::Custom => Self::Custom, + } + } +} + +impl From for common::EncryptionType { + fn from(value: EncryptionType) -> Self { + match value { + EncryptionType::None => Self::None, + EncryptionType::Gcm => Self::Gcm, + EncryptionType::Custom => Self::Custom, + } + } +} + +/// Operation type for text streams, mirroring [`ds_api::OperationType`]. +#[derive(uniffi::Enum, Clone, Copy, Debug, PartialEq, Eq)] +pub enum OperationType { + Create, + Update, + Delete, + Reaction, +} + +impl From for OperationType { + fn from(value: ds_api::OperationType) -> Self { + match value { + ds_api::OperationType::Create => Self::Create, + ds_api::OperationType::Update => Self::Update, + ds_api::OperationType::Delete => Self::Delete, + ds_api::OperationType::Reaction => Self::Reaction, + } + } +} + +impl From for ds_api::OperationType { + fn from(value: OperationType) -> Self { + match value { + OperationType::Create => Self::Create, + OperationType::Update => Self::Update, + OperationType::Delete => Self::Delete, + OperationType::Reaction => Self::Reaction, + } + } +} + +/// A capability a remote participant's client advertises, mirroring [`common::ClientCapability`]. +#[derive(uniffi::Enum, Clone, Copy, Debug, PartialEq, Eq)] +pub enum ClientCapability { + Unused, + PacketTrailer, + CompressionDeflateRaw, +} + +impl From for ClientCapability { + fn from(value: common::ClientCapability) -> Self { + match value { + common::ClientCapability::Unused => Self::Unused, + common::ClientCapability::PacketTrailer => Self::PacketTrailer, + common::ClientCapability::CompressionDeflateRaw => Self::CompressionDeflateRaw, + // `common::ClientCapability` is `#[non_exhaustive]`; treat anything newer as unusable. + _ => Self::Unused, + } + } +} + +impl From for common::ClientCapability { + fn from(value: ClientCapability) -> Self { + match value { + ClientCapability::Unused => Self::Unused, + ClientCapability::PacketTrailer => Self::PacketTrailer, + ClientCapability::CompressionDeflateRaw => Self::CompressionDeflateRaw, + } + } +} + +// MARK: - Info records + +/// Information about a byte data stream. FFI wrapper around [`ds_api::ByteStreamInfo`]. +#[derive(uniffi::Record, Clone, Debug)] +pub struct ByteStreamInfo { + pub id: String, + pub topic: String, + /// Unix timestamp in milliseconds. + pub timestamp_ms: i64, + pub total_length: Option, + pub attributes: HashMap, + pub mime_type: String, + pub name: String, + pub encryption_type: EncryptionType, +} + +impl From for ByteStreamInfo { + fn from(info: ds_api::ByteStreamInfo) -> Self { + let attributes = info.attributes(); + Self { + id: info.id, + topic: info.topic, + timestamp_ms: info.timestamp.timestamp_millis(), + total_length: info.total_length, + attributes, + mime_type: info.mime_type, + name: info.name, + encryption_type: info.encryption_type.into(), + } + } +} + +/// Information about a text data stream. FFI wrapper around [`ds_api::TextStreamInfo`]. +#[derive(uniffi::Record, Clone, Debug)] +pub struct TextStreamInfo { + pub id: String, + pub topic: String, + /// Unix timestamp in milliseconds. + pub timestamp_ms: i64, + pub total_length: Option, + pub attributes: HashMap, + pub mime_type: String, + pub operation_type: OperationType, + pub version: i32, + pub reply_to_stream_id: Option, + pub attached_stream_ids: Vec, + pub generated: bool, + pub encryption_type: EncryptionType, +} + +impl From for TextStreamInfo { + fn from(info: ds_api::TextStreamInfo) -> Self { + let attributes = info.attributes(); + Self { + id: info.id, + topic: info.topic, + timestamp_ms: info.timestamp.timestamp_millis(), + total_length: info.total_length, + attributes, + mime_type: info.mime_type, + operation_type: info.operation_type.into(), + version: info.version, + reply_to_stream_id: info.reply_to_stream_id, + attached_stream_ids: info.attached_stream_ids, + generated: info.generated, + encryption_type: info.encryption_type.into(), + } + } +} + +// MARK: - Option records + +/// Options for sending a byte stream. FFI wrapper around [`ds_api::StreamByteOptions`]. +#[derive(uniffi::Record, Clone, Debug, Default)] +pub struct StreamByteOptions { + pub topic: String, + pub attributes: HashMap, + #[uniffi(default = [])] + pub destination_identities: Vec, + #[uniffi(default = None)] + pub id: Option, + #[uniffi(default = None)] + pub mime_type: Option, + #[uniffi(default = None)] + pub name: Option, + #[uniffi(default = None)] + pub total_length: Option, + #[uniffi(default = None)] + pub compress: Option, + #[uniffi(default = None)] + pub sender_identity: Option, +} + +impl From for ds_api::StreamByteOptions { + fn from(options: StreamByteOptions) -> Self { + Self { + topic: options.topic, + attributes: options.attributes, + destination_identities: options + .destination_identities + .into_iter() + .map(Into::into) + .collect(), + id: options.id, + mime_type: options.mime_type, + name: options.name, + total_length: options.total_length, + compress: options.compress, + sender_identity: options.sender_identity.map(Into::into), + } + } +} + +/// Options for sending a text stream. FFI wrapper around [`ds_api::StreamTextOptions`]. +#[derive(uniffi::Record, Clone, Debug, Default)] +pub struct StreamTextOptions { + pub topic: String, + pub attributes: HashMap, + #[uniffi(default = [])] + pub destination_identities: Vec, + #[uniffi(default = None)] + pub id: Option, + #[uniffi(default = None)] + pub operation_type: Option, + #[uniffi(default = None)] + pub version: Option, + #[uniffi(default = None)] + pub reply_to_stream_id: Option, + #[uniffi(default = [])] + pub attached_stream_ids: Vec, + #[uniffi(default = None)] + pub generated: Option, + #[uniffi(default = None)] + pub compress: Option, + #[uniffi(default = None)] + pub sender_identity: Option, +} + +impl From for ds_api::StreamTextOptions { + fn from(options: StreamTextOptions) -> Self { + Self { + topic: options.topic, + attributes: options.attributes, + destination_identities: options + .destination_identities + .into_iter() + .map(Into::into) + .collect(), + id: options.id, + operation_type: options.operation_type.map(Into::into), + version: options.version, + reply_to_stream_id: options.reply_to_stream_id, + attached_stream_ids: options.attached_stream_ids, + generated: options.generated, + compress: options.compress, + sender_identity: options.sender_identity.map(Into::into), + } + } +} + +// MARK: - Error + +/// A data stream operation failed. Structured mirror of [`ds_api::StreamError`] so foreign callers +/// can map each case to their own error type; variants carrying a message forward it as `message`. +#[derive(uniffi::Error, thiserror::Error, Debug)] +pub enum DataStreamError { + #[error("stream has already been closed")] + AlreadyClosed, + + // Named `reason` rather than `message`: in Kotlin a variant field called `message` collides + // with the `message` uniffi overrides from Throwable, and the collision cannot be renamed away + // from uniffi.toml (renames of enum members declared in a submodule are silently dropped). + #[error("stream closed abnormally: {reason}")] + AbnormalEnd { reason: String }, + + #[error("UTF-8 decoding error: {reason}")] + Utf8 { reason: String }, + + #[error("incoming header was invalid")] + InvalidHeader, + + #[error("expected chunk index to be exactly one more than the previous")] + MissedChunk, + + #[error("read length exceeded total length specified in stream header")] + LengthExceeded, + + #[error("stream data is incomplete")] + Incomplete, + + #[error("unable to send packet")] + SendFailed, + + #[error("I/O error: {reason}")] + Io { reason: String }, + + #[error("internal error")] + Internal, + + #[error("encryption type mismatch: expected {expected:?}, received {received:?}")] + EncryptionTypeMismatch { expected: EncryptionType, received: EncryptionType }, + + #[error("stream header exceeds maximum size")] + HeaderTooLarge, + + #[error("stream payload exceeds maximum size")] + PayloadTooLarge, + + #[error("decompression failed")] + Decompression, + + #[error("file name must be a plain file name without path separators or '..'")] + InvalidFileName, +} + +/// A foreign transport failed to deliver outbound packets; thrown by hosts from +/// [`OutgoingDataStreamManagerDelegate::on_packets_available`](super::outgoing::OutgoingDataStreamManagerDelegate::on_packets_available). +/// +/// Morally `struct PacketDeliveryError(String)`, but uniffi error types must be enums, so the +/// string travels as the single variant's `reason` (free-form host context: logged, not parsed). +#[derive(uniffi::Error, thiserror::Error, Debug)] +pub enum PacketDeliveryError { + #[error("failed to deliver packets: {reason}")] + Failed { reason: String }, +} + +// Required because foreign code implements delegate methods returning this error: an exception +// that is NOT a `PacketDeliveryError` surfaces through this catch-all rather than aborting. +impl From for PacketDeliveryError { + fn from(error: uniffi::UnexpectedUniFFICallbackError) -> Self { + Self::Failed { reason: error.reason } + } +} + +impl From for DataStreamError { + fn from(error: PacketDeliveryError) -> Self { + log::error!("outbound packet delivery failed: {error}"); + Self::Internal + } +} + +impl From for DataStreamError { + fn from(error: ds_api::StreamError) -> Self { + match error { + ds_api::StreamError::AlreadyClosed => Self::AlreadyClosed, + ds_api::StreamError::AbnormalEnd(reason) => Self::AbnormalEnd { reason }, + ds_api::StreamError::Utf8(error) => Self::Utf8 { reason: error.to_string() }, + ds_api::StreamError::InvalidHeader => Self::InvalidHeader, + ds_api::StreamError::MissedChunk => Self::MissedChunk, + ds_api::StreamError::LengthExceeded => Self::LengthExceeded, + ds_api::StreamError::Incomplete => Self::Incomplete, + ds_api::StreamError::SendFailed => Self::SendFailed, + ds_api::StreamError::Io(error) => Self::Io { reason: error.to_string() }, + ds_api::StreamError::Internal => Self::Internal, + ds_api::StreamError::EncryptionTypeMismatch { expected, received } => { + Self::EncryptionTypeMismatch { + expected: expected.into(), + received: received.into(), + } + } + ds_api::StreamError::HeaderTooLarge => Self::HeaderTooLarge, + ds_api::StreamError::PayloadTooLarge => Self::PayloadTooLarge, + ds_api::StreamError::Decompression => Self::Decompression, + ds_api::StreamError::InvalidFileName => Self::InvalidFileName, + } + } +} + +// MARK: - Wire decode + +/// Decodes a serialized [`proto::DataPacket`] carrying a data-stream header/chunk/trailer into an +/// incoming-manager input event. Returns `None` if the bytes don't decode or the packet isn't a +/// data-stream packet. +/// +/// The encryption type must be supplied by the caller and CANNOT be recovered from the bytes: +/// `encrypted_packet` is a member of the `DataPacket.value` oneof, so a host decrypting E2EE +/// traffic replaces it with the decrypted stream header/chunk/trailer β€” by the time these bytes +/// arrive, the field is absent from the wire format. Hosts without E2EE pass +/// [`common::EncryptionType::None`]. +pub(crate) fn decode_data_packet( + bytes: &[u8], + encryption_type: common::EncryptionType, +) -> Option { + let mut packet = proto::DataPacket::decode(bytes).ok()?; + let identity: common::ParticipantIdentity = packet.participant_identity.clone().into(); + let ds_packet = match packet.value.take()? { + proto::data_packet::Value::StreamHeader(header) => { + ds::Packet::Header { header: header.into(), encryption_type } + } + proto::data_packet::Value::StreamChunk(chunk) => { + ds::Packet::Chunk { chunk: chunk.into(), encryption_type } + } + proto::data_packet::Value::StreamTrailer(trailer) => ds::Packet::Trailer(trailer.into()), + _ => return None, + }; + Some(ds::incoming::PacketReceived::new(ds_packet, identity)) +} diff --git a/livekit-uniffi/src/data_stream/incoming.rs b/livekit-uniffi/src/data_stream/incoming.rs new file mode 100644 index 000000000..16bc16143 --- /dev/null +++ b/livekit-uniffi/src/data_stream/incoming.rs @@ -0,0 +1,266 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::sync::Arc; + +use bytes::{Bytes, BytesMut}; +use futures_util::StreamExt; +use livekit_data_stream::{api as ds_api, backend as ds}; +use tokio::sync::mpsc::UnboundedReceiver; +use tokio::sync::Mutex; +use tokio_util::sync::{CancellationToken, DropGuard}; + +use super::common::{ + decode_data_packet, ByteStreamInfo, DataStreamError, EncryptionType, TextStreamInfo, +}; +use ds_api::StreamReader as _; + +/// Receives inbound data-stream packets and processes them on the incoming manager's actor loop, +/// surfacing opened readers through a foreign delegate. +/// +/// Mirrors [`crate::data_track::remote::RemoteDataTrackManager`]: `handle_packet_received` is a +/// cheap synchronous enqueue (safe to call from a native data-channel callback), while +/// decompression and reassembly happen on the spawned `run` task in packet order. +#[derive(uniffi::Object)] +pub struct IncomingDataStreamManager { + input: ds::incoming::ManagerInput, + _guard: DropGuard, +} + +/// Delegate for receiving output events from [`IncomingDataStreamManager`]. +/// +/// Only stream lifecycle events (opened/closed) are surfaced. The manager's deprecated v1 raw +/// chunk/trailer notifications are intentionally not forwarded over the FFI boundary. +#[uniffi::export(with_foreign)] +pub trait IncomingDataStreamManagerDelegate: Send + Sync { + /// A byte stream was opened by `identity` and is ready to be read. + fn on_byte_stream_opened(&self, reader: Arc, identity: String); + + /// A text stream was opened by `identity` and is ready to be read. + fn on_text_stream_opened(&self, reader: Arc, identity: String); + + /// A previously opened stream terminated on the wire and will produce no further data: its + /// trailer arrived, its (single-packet) inline payload completed, it failed, or it was + /// aborted. Emitted exactly once per opened stream, after the corresponding open event. + /// + /// Hosts delivering streams on ordered topics use this to know when a stream's handler chain + /// can advance β€” a stream that is still open must not block streams opened after it forever. + fn on_stream_closed(&self, stream_id: String, identity: String); +} + +#[uniffi::export(async_runtime = "tokio")] +impl IncomingDataStreamManager { + /// Creates a manager that surfaces opened streams to `delegate`. + /// + /// `max_payload_byte_length` caps the decompressed size of a single incoming stream + /// (`None` = default, 5 GB) and is **fixed for the lifetime of the manager**. If the host + /// sources it from per-connection options that aren't final until connect β€” and can differ + /// between sessions of the same host object β€” construct a fresh manager for each session + /// rather than lazily memoizing one, or the first session's value is silently pinned. (Same + /// class of rough edge as a single room instance being `connect()`ed multiple times.) + /// Rebuilding is cheap and safe: dropping the manager cancels its tasks, and handler wiring + /// lives on the foreign side. + #[uniffi::constructor] + pub fn new( + delegate: Arc, + max_payload_byte_length: Option, + ) -> Arc { + let token = CancellationToken::new(); + let (manager, input, output) = + ds::incoming::Manager::new(max_payload_byte_length.map(|n| n as usize)); + + let rt = crate::runtime::runtime(); + rt.spawn(shutdown_forward_task(input.clone(), token.clone())); + let delegate_forward = DelegateForwardTask { output, delegate, token: token.clone() }; + rt.spawn(delegate_forward.run()); + rt.spawn(manager.run()); + + Self { input, _guard: token.drop_guard() }.into() + } + + /// Handles an encoded [`livekit_protocol::DataPacket`] received over the data channel. + /// + /// Fire-and-forget: the packet is decoded and enqueued in order; processing happens on the + /// manager's run loop. Non-data-stream or undecodable packets are ignored. + /// + /// `encryption_type` is how this packet arrived on the wire, and must be passed by the host + /// because it cannot be recovered from the bytes: `encrypted_packet` is a member of the + /// `DataPacket.value` oneof, so decrypting replaces it with the decrypted stream packet. + /// Hosts without end-to-end encryption pass [`EncryptionType::None`]; hosts with E2EE pass + /// the type they decrypted with, letting the manager reject chunks whose encryption doesn't + /// match their stream's header ([`DataStreamError::EncryptionTypeMismatch`]). + pub fn handle_packet_received(&self, packet: Bytes, encryption_type: EncryptionType) { + if let Some(event) = decode_data_packet(&packet, encryption_type.into()) { + let _ = self.input.send(event.into()); + } + } + + /// Aborts all open incoming streams so their readers error instead of hanging (e.g. on + /// disconnect). Handler wiring on the foreign side survives, so streams that arrive later + /// (e.g. after a reconnect) are still processed. + pub fn abort_all_streams(&self) { + let _ = self.input.send(ds::incoming::InputEvent::AbortAllStreams); + } + + /// Aborts open incoming streams sent by `identity` (e.g. when that participant disconnects + /// mid-send), so their readers error instead of hanging. + pub fn abort_streams_from(&self, identity: String) { + let _ = self.input.send(ds::incoming::InputEvent::AbortStreamsFrom(identity.into())); + } + + /// Number of currently open incoming streams: streams announced by a header that are still + /// awaiting more packets. Inline (single-packet) streams complete immediately and are never + /// counted. + /// + /// The query runs on the manager's loop in order with previously submitted events, so a + /// packet or abort passed beforehand is reflected in the answer β€” useful in tests to wait for + /// a header to register (or an abort to land) without racing the run loop. + pub async fn open_stream_count(&self) -> u64 { + let (respond_to, response) = tokio::sync::oneshot::channel(); + if self.input.send(ds::incoming::InputEvent::QueryOpenStreamCount(respond_to)).is_err() { + return 0; + } + response.await.unwrap_or(0) as u64 + } +} + +/// Reader for an incoming byte data stream. +#[derive(uniffi::Object)] +pub struct ByteStreamReader { + info: ByteStreamInfo, + inner: Mutex, +} + +impl ByteStreamReader { + fn new(reader: ds_api::ByteStreamReader) -> Self { + Self { info: reader.info().clone().into(), inner: Mutex::new(reader) } + } +} + +#[uniffi::export(async_runtime = "tokio")] +impl ByteStreamReader { + /// Information about the underlying stream. + pub fn info(&self) -> ByteStreamInfo { + self.info.clone() + } + + /// Returns the next chunk, or `None` once the stream has closed. + pub async fn next(&self) -> Result, DataStreamError> { + Ok(self.inner.lock().await.next().await.transpose()?) + } + + /// Reads every chunk, concatenating them into a single buffer returned once the stream closes. + pub async fn read_all(&self) -> Result { + let mut reader = self.inner.lock().await; + let mut buffer = BytesMut::new(); + while let Some(chunk) = reader.next().await { + buffer.extend_from_slice(&chunk?); + } + Ok(buffer.freeze()) + } +} + +/// Reader for an incoming text data stream. +#[derive(uniffi::Object)] +pub struct TextStreamReader { + info: TextStreamInfo, + inner: Mutex, +} + +impl TextStreamReader { + fn new(reader: ds_api::TextStreamReader) -> Self { + Self { info: reader.info().clone().into(), inner: Mutex::new(reader) } + } +} + +#[uniffi::export(async_runtime = "tokio")] +impl TextStreamReader { + /// Information about the underlying stream. + pub fn info(&self) -> TextStreamInfo { + self.info.clone() + } + + /// Returns the next chunk, or `None` once the stream has closed. + pub async fn next(&self) -> Result, DataStreamError> { + Ok(self.inner.lock().await.next().await.transpose()?) + } + + /// Reads every chunk, concatenating them into a single string returned once the stream closes. + pub async fn read_all(&self) -> Result { + let mut reader = self.inner.lock().await; + let mut result = String::new(); + while let Some(chunk) = reader.next().await { + result.push_str(&chunk?); + } + Ok(result) + } +} + +/// Forwards manager output events to the foreign [`IncomingDataStreamManagerDelegate`]. +struct DelegateForwardTask { + output: UnboundedReceiver, + delegate: Arc, + token: CancellationToken, +} + +impl DelegateForwardTask { + async fn run(mut self) { + loop { + tokio::select! { + _ = self.token.cancelled() => break, + event = self.output.recv() => match event { + Some(event) => self.forward_event(event), + None => break, + } + } + } + } + + fn forward_event(&self, event: ds::incoming::OutputEvent) { + match event { + ds::incoming::OutputEvent::StreamOpened(ds::incoming::StreamOpened { + stream_reader, + participant_identity, + }) => { + let identity = participant_identity.to_string(); + match stream_reader { + ds_api::AnyStreamReader::Byte(reader) => { + let reader = Arc::new(ByteStreamReader::new(reader)); + self.delegate.on_byte_stream_opened(reader, identity); + } + ds_api::AnyStreamReader::Text(reader) => { + let reader = Arc::new(TextStreamReader::new(reader)); + self.delegate.on_text_stream_opened(reader, identity); + } + } + } + ds::incoming::OutputEvent::StreamClosed(ds::incoming::StreamClosed { + stream_id, + participant_identity, + topic: _, + }) => { + self.delegate + .on_stream_closed(stream_id.to_string(), participant_identity.to_string()); + } + // Deprecated v1 raw chunk/trailer notifications are not surfaced over the FFI boundary. + ds::incoming::OutputEvent::ChunkReceived(_) + | ds::incoming::OutputEvent::TrailerReceived(_) => {} + } + } +} + +async fn shutdown_forward_task(input: ds::incoming::ManagerInput, token: CancellationToken) { + token.cancelled().await; + let _ = input.send(ds::incoming::InputEvent::Shutdown); +} diff --git a/livekit-uniffi/src/data_stream/mod.rs b/livekit-uniffi/src/data_stream/mod.rs new file mode 100644 index 000000000..52f75f851 --- /dev/null +++ b/livekit-uniffi/src/data_stream/mod.rs @@ -0,0 +1,34 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! UniFFI bindings for data streams v2 from [`livekit-data-stream`]. +//! +//! Mirrors the [`crate::data_track`] pattern: +//! - [`incoming::IncomingDataStreamManager`] wraps the incoming actor. Packets are fed in via a +//! synchronous `handle_packet_received` (safe to call from a native data-channel callback), and +//! opened readers / v1 back-compat events are pushed out through a foreign delegate. +//! - [`outgoing::OutgoingDataStreamManager`] wraps the outgoing manager as an object with async +//! `send_*`/`stream_*` methods. Outbound packets are handed to a foreign delegate, and remote +//! participant protocol/capabilities are read through a foreign registry callback. +//! +//! [`polled`] adapts both managers for bindings that cannot accept a delegate call from an +//! arbitrary thread β€” see its module docs. + +pub mod common; +pub mod incoming; +pub mod outgoing; +pub mod polled; + +#[cfg(test)] +mod tests; diff --git a/livekit-uniffi/src/data_stream/outgoing.rs b/livekit-uniffi/src/data_stream/outgoing.rs new file mode 100644 index 000000000..a3f29e592 --- /dev/null +++ b/livekit-uniffi/src/data_stream/outgoing.rs @@ -0,0 +1,236 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::sync::Arc; + +use bytes::Bytes; +use livekit_common as lk_common; +use livekit_data_stream::{api as ds_api, backend as ds}; +use prost::Message as _; +use tokio_util::sync::{CancellationToken, DropGuard}; + +use super::common::{ + ByteStreamInfo, ClientCapability, DataStreamError, PacketDeliveryError, StreamByteOptions, + StreamTextOptions, TextStreamInfo, +}; +use ds_api::StreamWriter as _; + +/// Sends data streams, choosing v2 single-packet/compression or legacy multi-packet framing based +/// on recipient capabilities. Outbound packets are handed to a foreign delegate for transport. +#[derive(uniffi::Object)] +pub struct OutgoingDataStreamManager { + manager: ds::outgoing::Manager, + registry: Arc, + _guard: DropGuard, +} + +/// Delegate for receiving outbound packets from [`OutgoingDataStreamManager`]. +#[uniffi::export(with_foreign)] +pub trait OutgoingDataStreamManagerDelegate: Send + Sync { + /// Encoded [`livekit_protocol::DataPacket`]s to be sent over the data channel transport, in + /// order. One-shot sends (`send_text`/`send_bytes`) deliver their entire stream β€” header, + /// chunks, trailer β€” in a single call; incremental writers and `send_file` deliver one packet + /// per call. + /// + /// Return only once the packets have been handed to the transport: the originating + /// `send_*`/`write` call stays pending until then, which is what bounds how fast a producer + /// can enqueue. Throwing [`PacketDeliveryError`] fails that call with + /// [`DataStreamError::SendFailed`](super::common::DataStreamError::SendFailed) and closes the + /// affected stream (`is_open` becomes false for writers). + fn on_packets_available(&self, packets: Vec) -> Result<(), PacketDeliveryError>; +} + +/// Read access to remote participants' advertised protocol and capabilities, implemented by the +/// foreign side. Mirrors [`lk_common::RemoteParticipantRegistry`]; used to decide inline/compression +/// eligibility per send. +#[uniffi::export(with_foreign)] +pub trait RemoteParticipantRegistryDelegate: Send + Sync { + /// A remote participant's `client_protocol`, or `0` (`CLIENT_PROTOCOL_DEFAULT`) if unknown. + fn remote_client_protocol(&self, identity: String) -> i32; + + /// A remote participant's advertised capabilities, or empty if unknown. + fn remote_capabilities(&self, identity: String) -> Vec; + + /// The identities of every remote participant, used to resolve a broadcast send. + fn remote_identities(&self) -> Vec; +} + +/// Adapts the foreign [`RemoteParticipantRegistryDelegate`] to the crate-internal +/// [`lk_common::RemoteParticipantRegistry`] the outgoing manager consumes. +struct ForeignRegistry(Arc); + +impl lk_common::RemoteParticipantRegistry for ForeignRegistry { + fn remote_client_protocol(&self, identity: &lk_common::ParticipantIdentity) -> i32 { + self.0.remote_client_protocol(identity.to_string()) + } + + fn remote_capabilities( + &self, + identity: &lk_common::ParticipantIdentity, + ) -> Vec { + self.0.remote_capabilities(identity.to_string()).into_iter().map(Into::into).collect() + } + + fn remote_identities(&self) -> Vec { + self.0.remote_identities().into_iter().map(Into::into).collect() + } +} + +#[uniffi::export(async_runtime = "tokio")] +impl OutgoingDataStreamManager { + #[uniffi::constructor] + pub fn new( + delegate: Arc, + registry: Arc, + ) -> Arc { + let token = CancellationToken::new(); + let (manager, mut packet_rx) = ds::outgoing::Manager::new(); + + // Forward each outbound packet batch to the transport delegate, acknowledging the send + // with the delegate's own result so wire failures propagate back to the originating + // `send_*`/`write` call (which surfaces them as `DataStreamError::SendFailed`). + let forward_token = token.clone(); + crate::runtime::runtime().spawn(async move { + loop { + tokio::select! { + _ = forward_token.cancelled() => break, + recv = packet_rx.recv() => match recv { + Ok((packets, responder)) => { + let encoded = packets + .iter() + .map(|packet| Bytes::from(packet.encode_to_vec())) + .collect(); + let result = delegate + .on_packets_available(encoded) + .map_err(|_| ds_api::SendError); + let _ = responder.respond(result); + } + Err(_) => break, + } + } + } + }); + + let registry: Arc = + Arc::new(ForeignRegistry(registry)); + Self { manager, registry, _guard: token.drop_guard() }.into() + } + + /// Sends a complete text payload, returning info about the created stream. + pub async fn send_text( + &self, + text: String, + options: StreamTextOptions, + ) -> Result { + Ok(self.manager.send_text(&text, options.into(), &*self.registry).await?.into()) + } + + /// Sends a complete byte payload, returning info about the created stream. + pub async fn send_bytes( + &self, + data: Bytes, + options: StreamByteOptions, + ) -> Result { + Ok(self.manager.send_bytes(data, options.into(), &*self.registry).await?.into()) + } + + /// Streams a file from disk, returning info about the created stream. + pub async fn send_file( + &self, + path: String, + options: StreamByteOptions, + ) -> Result { + Ok(self.manager.send_file(path, options.into(), &*self.registry).await?.into()) + } + + /// Opens an incremental text stream writer (never compressed or inlined). + pub async fn stream_text( + &self, + options: StreamTextOptions, + ) -> Result { + Ok(TextStreamWriter(self.manager.stream_text(options.into()).await?)) + } + + /// Opens an incremental byte stream writer (never compressed or inlined). + pub async fn stream_bytes( + &self, + options: StreamByteOptions, + ) -> Result { + Ok(ByteStreamWriter(self.manager.stream_bytes(options.into()).await?)) + } +} + +/// Writer for an open text data stream. +#[derive(uniffi::Object)] +pub struct TextStreamWriter(ds_api::TextStreamWriter); + +#[uniffi::export(async_runtime = "tokio")] +impl TextStreamWriter { + /// Information about the underlying stream. + pub fn info(&self) -> TextStreamInfo { + self.0.info().clone().into() + } + + /// Whether the stream is still open β€” false once it has been closed locally or a send has failed. + pub async fn is_open(&self) -> bool { + !self.0.is_closed().await + } + + /// Appends text to the stream. + pub async fn write(&self, text: String) -> Result<(), DataStreamError> { + Ok(self.0.write(&text).await?) + } + + /// Closes the stream normally. + pub async fn close(&self) -> Result<(), DataStreamError> { + Ok(self.0.clone().close().await?) + } + + /// Closes the stream abnormally with a reason. + pub async fn close_with_reason(&self, reason: String) -> Result<(), DataStreamError> { + Ok(self.0.clone().close_with_reason(&reason).await?) + } +} + +/// Writer for an open byte data stream. +#[derive(uniffi::Object)] +pub struct ByteStreamWriter(ds_api::ByteStreamWriter); + +#[uniffi::export(async_runtime = "tokio")] +impl ByteStreamWriter { + /// Information about the underlying stream. + pub fn info(&self) -> ByteStreamInfo { + self.0.info().clone().into() + } + + /// Whether the stream is still open β€” false once it has been closed locally or a send has failed. + pub async fn is_open(&self) -> bool { + !self.0.is_closed().await + } + + /// Appends bytes to the stream. + pub async fn write(&self, data: Bytes) -> Result<(), DataStreamError> { + Ok(self.0.write(data.as_ref()).await?) + } + + /// Closes the stream normally. + pub async fn close(&self) -> Result<(), DataStreamError> { + Ok(self.0.clone().close().await?) + } + + /// Closes the stream abnormally with a reason. + pub async fn close_with_reason(&self, reason: String) -> Result<(), DataStreamError> { + Ok(self.0.clone().close_with_reason(&reason).await?) + } +} diff --git a/livekit-uniffi/src/data_stream/polled.rs b/livekit-uniffi/src/data_stream/polled.rs new file mode 100644 index 000000000..926031109 --- /dev/null +++ b/livekit-uniffi/src/data_stream/polled.rs @@ -0,0 +1,293 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Pull-based adapters over the push delegates, for bindings whose callbacks are thread-affine. +//! +//! [`incoming`](super::incoming) and [`outgoing`](super::outgoing) surface their output by calling +//! a foreign delegate from this crate's tokio runtime. Some bindings cannot accept that. Dart is +//! the motivating case: uniffi compiles a callback interface to `Pointer.fromFunction`, which is +//! only valid on the thread owning the isolate, so a delegate invoked from a tokio worker aborts +//! the VM outright with "Cannot invoke native callback outside an isolate" β€” not a catchable +//! error. (Dart's thread-safe callback form, `NativeCallable.listener`, is asynchronous and cannot +//! return a value, so it can't satisfy uniffi's synchronous callback ABI either.) +//! +//! The fix is to keep the delegate on the Rust side. Each type here implements the relevant +//! delegate trait, buffers what it receives into a channel, and exposes an `async fn next_*` the +//! foreign side awaits. Nothing crosses the FFI until that await resolves, and uniffi polls those +//! futures from whichever thread called `rust_future_poll` β€” the binding's own. Delegate +//! invocation still happens on a tokio thread, which is fine precisely because the implementation +//! is Rust. +//! +//! Note that [`RemoteParticipantRegistryDelegate`] needs no adapter: it is only ever called +//! synchronously inside a `send_*` future, so it already runs on the polling thread. +//! +//! Bindings that can take callbacks from any thread (Swift, Kotlin) should ignore this module and +//! construct the managers directly. + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +use bytes::Bytes; +use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}; +use tokio::sync::Mutex; +use tokio_util::sync::CancellationToken; + +use super::common::PacketDeliveryError; +use super::incoming::{ + ByteStreamReader, IncomingDataStreamManager, IncomingDataStreamManagerDelegate, + TextStreamReader, +}; +use super::outgoing::{ + OutgoingDataStreamManager, OutgoingDataStreamManagerDelegate, RemoteParticipantRegistryDelegate, +}; + +/// Queue depth at which we start warning. The channels are unbounded so a stalled consumer can't +/// deadlock the manager, which means the only backstop against unbounded growth is noticing. +const QUEUE_DEPTH_WARN: usize = 1024; + +fn warn_if_deep(kind: &str, depth: usize) { + if depth == QUEUE_DEPTH_WARN { + log::warn!( + "{kind} queue has reached {depth} pending items; the foreign side is not draining it \ + fast enough" + ); + } +} + +// MARK: - Outgoing + +/// Buffers outbound packets so they can be pulled instead of pushed. +/// +/// Implements [`OutgoingDataStreamManagerDelegate`] in Rust; see the module docs. +#[derive(uniffi::Object)] +pub struct OutgoingPacketQueue { + tx: UnboundedSender, + rx: Mutex>, + depth: AtomicUsize, + shutdown: CancellationToken, +} + +impl OutgoingPacketQueue { + fn new() -> Self { + let (tx, rx) = unbounded_channel(); + Self { + tx, + rx: Mutex::new(rx), + depth: AtomicUsize::new(0), + shutdown: CancellationToken::new(), + } + } +} + +impl OutgoingDataStreamManagerDelegate for OutgoingPacketQueue { + // Acknowledges once buffered: a pull adapter has no synchronous transport feedback, so send + // failures observed while draining must be handled host-side. + fn on_packets_available(&self, packets: Vec) -> Result<(), PacketDeliveryError> { + for packet in packets { + if self.tx.send(packet).is_ok() { + warn_if_deep("outgoing packet", self.depth.fetch_add(1, Ordering::Relaxed) + 1); + } + } + Ok(()) + } +} + +#[uniffi::export(async_runtime = "tokio")] +impl OutgoingPacketQueue { + /// Awaits the next batch of encoded `livekit.DataPacket`s to put on the wire. + /// + /// Returns `None` once the manager has shut down, which ends the caller's drain loop. + /// Everything already queued is returned together, so a burst costs one FFI crossing rather + /// than one per packet. + pub async fn next_packets(&self) -> Option> { + let mut rx = self.rx.lock().await; + let first = tokio::select! { + _ = self.shutdown.cancelled() => return None, + received = rx.recv() => received?, + }; + let mut batch = vec![first]; + while let Ok(next) = rx.try_recv() { + batch.push(next); + } + self.depth.fetch_sub(batch.len(), Ordering::Relaxed); + Some(batch) + } + + /// Wakes a pending [`Self::next_packets`] with `None` so the caller's drain loop can exit. + /// + /// Call this before releasing the queue: a caller blocked in `next_packets` is holding a + /// pointer to it, so freeing it first is a use-after-free. + pub fn close(&self) { + self.shutdown.cancel(); + } +} + +/// An [`OutgoingDataStreamManager`] and the queue draining it, already connected. +#[derive(uniffi::Record)] +pub struct PolledOutgoingDataStreamManager { + pub manager: Arc, + pub packets: Arc, +} + +/// Builds an outgoing manager whose packets are pulled rather than pushed. +/// +/// The two halves are wired together here rather than by the caller: passing a Rust object where +/// `Arc` is expected is awkward-to-impossible from some +/// bindings, and unnecessary β€” it's ordinary Rust on this side. +#[uniffi::export] +pub fn polled_outgoing_data_stream_manager( + registry: Arc, +) -> PolledOutgoingDataStreamManager { + let packets = Arc::new(OutgoingPacketQueue::new()); + let manager = OutgoingDataStreamManager::new(packets.clone(), registry); + PolledOutgoingDataStreamManager { manager, packets } +} + +// MARK: - Incoming + +/// A stream opened by a remote participant. +/// +/// Exactly one of the two readers is set; which one tells you the stream's kind. Two `Option`s +/// rather than an enum keeps the shape trivial in every binding. +#[derive(uniffi::Record)] +pub struct OpenedStream { + /// Identity of the participant that opened the stream. + pub identity: String, + /// Set when the stream carries bytes. + pub byte_reader: Option>, + /// Set when the stream carries text. + pub text_reader: Option>, +} + +/// A stream closed by a remote participant (or terminated by an error/abort); see +/// [`IncomingDataStreamManagerDelegate::on_stream_closed`]. +#[derive(uniffi::Record)] +pub struct ClosedStream { + /// Id of the stream that closed. + pub stream_id: String, + /// Identity of the participant that opened the stream. + pub identity: String, +} + +/// Buffers opened and closed streams so they can be pulled instead of pushed. +/// +/// Implements [`IncomingDataStreamManagerDelegate`] in Rust; see the module docs. +#[derive(uniffi::Object)] +pub struct IncomingStreamQueue { + tx: UnboundedSender, + rx: Mutex>, + depth: AtomicUsize, + closed_tx: UnboundedSender, + closed_rx: Mutex>, + closed_depth: AtomicUsize, + shutdown: CancellationToken, +} + +impl IncomingStreamQueue { + fn new() -> Self { + let (tx, rx) = unbounded_channel(); + let (closed_tx, closed_rx) = unbounded_channel(); + Self { + tx, + rx: Mutex::new(rx), + depth: AtomicUsize::new(0), + closed_tx, + closed_rx: Mutex::new(closed_rx), + closed_depth: AtomicUsize::new(0), + shutdown: CancellationToken::new(), + } + } + + fn push(&self, opened: OpenedStream) { + if self.tx.send(opened).is_ok() { + warn_if_deep("incoming stream", self.depth.fetch_add(1, Ordering::Relaxed) + 1); + } + } +} + +impl IncomingDataStreamManagerDelegate for IncomingStreamQueue { + fn on_byte_stream_opened(&self, reader: Arc, identity: String) { + self.push(OpenedStream { identity, byte_reader: Some(reader), text_reader: None }); + } + + fn on_text_stream_opened(&self, reader: Arc, identity: String) { + self.push(OpenedStream { identity, byte_reader: None, text_reader: Some(reader) }); + } + + fn on_stream_closed(&self, stream_id: String, identity: String) { + if self.closed_tx.send(ClosedStream { stream_id, identity }).is_ok() { + warn_if_deep("closed stream", self.closed_depth.fetch_add(1, Ordering::Relaxed) + 1); + } + } +} + +#[uniffi::export(async_runtime = "tokio")] +impl IncomingStreamQueue { + /// Awaits the next stream opened by a remote participant. + /// + /// Returns `None` once the manager has shut down, which ends the caller's drain loop. Unlike + /// [`OutgoingPacketQueue::next_packets`] this yields one at a time: each carries a reader the + /// caller has to route to a handler, so batching would only defer that work. + pub async fn next_opened_stream(&self) -> Option { + let mut rx = self.rx.lock().await; + let opened = tokio::select! { + _ = self.shutdown.cancelled() => return None, + received = rx.recv() => received?, + }; + self.depth.fetch_sub(1, Ordering::Relaxed); + Some(opened) + } + + /// Awaits the next stream-closed notification. + /// + /// Pulled independently of [`Self::next_opened_stream`], so ordering across the two queues is + /// not guaranteed β€” correlate by `stream_id` (a close always follows its open on the push + /// side). Returns `None` once the manager has shut down. + pub async fn next_closed_stream(&self) -> Option { + let mut rx = self.closed_rx.lock().await; + let closed = tokio::select! { + _ = self.shutdown.cancelled() => return None, + received = rx.recv() => received?, + }; + self.closed_depth.fetch_sub(1, Ordering::Relaxed); + Some(closed) + } + + /// Wakes a pending [`Self::next_opened_stream`] or [`Self::next_closed_stream`] with `None`. + /// See [`OutgoingPacketQueue::close`]. + pub fn close(&self) { + self.shutdown.cancel(); + } +} + +/// An [`IncomingDataStreamManager`] and the queue draining it, already connected. +#[derive(uniffi::Record)] +pub struct PolledIncomingDataStreamManager { + pub manager: Arc, + pub streams: Arc, +} + +/// Builds an incoming manager whose opened streams are pulled rather than pushed. +/// +/// `max_payload_byte_length` is fixed for the lifetime of the manager; see +/// [`IncomingDataStreamManager::new`]. Hosts sourcing it from per-connection options should build +/// a fresh manager per session rather than memoizing one. +#[uniffi::export] +pub fn polled_incoming_data_stream_manager( + max_payload_byte_length: Option, +) -> PolledIncomingDataStreamManager { + let streams = Arc::new(IncomingStreamQueue::new()); + let manager = IncomingDataStreamManager::new(streams.clone(), max_payload_byte_length); + PolledIncomingDataStreamManager { manager, streams } +} diff --git a/livekit-uniffi/src/data_stream/tests.rs b/livekit-uniffi/src/data_stream/tests.rs new file mode 100644 index 000000000..75cf1ff8d --- /dev/null +++ b/livekit-uniffi/src/data_stream/tests.rs @@ -0,0 +1,473 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Round-trip tests driving the FFI wrappers through mock foreign delegates on the global runtime. + +use std::sync::{Arc, Mutex}; + +use bytes::Bytes; +use livekit_protocol as proto; +use prost::Message as _; +use tokio::sync::oneshot; + +use super::common::{ + ClientCapability, DataStreamError, EncryptionType, PacketDeliveryError, StreamTextOptions, +}; +use super::incoming::{ + ByteStreamReader, IncomingDataStreamManager, IncomingDataStreamManagerDelegate, + TextStreamReader, +}; +use super::outgoing::{ + OutgoingDataStreamManager, OutgoingDataStreamManagerDelegate, RemoteParticipantRegistryDelegate, +}; + +/// Builds an encoded v2 inline (single-packet) text `DataPacket`. +fn inline_text_packet(identity: &str, topic: &str, text: &str) -> Bytes { + let header = proto::data_stream::Header { + stream_id: "s1".to_string(), + topic: topic.to_string(), + mime_type: "text/plain".to_string(), + timestamp: 0, + total_length: Some(text.len() as u64), + inline_content: Some(text.as_bytes().to_vec()), + content_header: Some(proto::data_stream::header::ContentHeader::TextHeader( + proto::data_stream::TextHeader::default(), + )), + ..Default::default() + }; + let packet = proto::DataPacket { + participant_identity: identity.to_string(), + value: Some(proto::data_packet::Value::StreamHeader(header)), + ..Default::default() + }; + Bytes::from(packet.encode_to_vec()) +} + +/// Builds an encoded v1 multi-packet text stream header `DataPacket` (no inline content). +fn multipacket_text_header_packet(identity: &str, topic: &str, total_length: u64) -> Bytes { + let header = proto::data_stream::Header { + stream_id: "s1".to_string(), + topic: topic.to_string(), + mime_type: "text/plain".to_string(), + timestamp: 0, + total_length: Some(total_length), + content_header: Some(proto::data_stream::header::ContentHeader::TextHeader( + proto::data_stream::TextHeader::default(), + )), + ..Default::default() + }; + let packet = proto::DataPacket { + participant_identity: identity.to_string(), + value: Some(proto::data_packet::Value::StreamHeader(header)), + ..Default::default() + }; + Bytes::from(packet.encode_to_vec()) +} + +/// Builds an encoded chunk `DataPacket` for stream `s1`. +fn chunk_packet(identity: &str, chunk_index: u64, content: &[u8]) -> Bytes { + let chunk = proto::data_stream::Chunk { + stream_id: "s1".to_string(), + chunk_index, + content: content.to_vec(), + ..Default::default() + }; + let packet = proto::DataPacket { + participant_identity: identity.to_string(), + value: Some(proto::data_packet::Value::StreamChunk(chunk)), + ..Default::default() + }; + Bytes::from(packet.encode_to_vec()) +} + +/// Builds an encoded trailer `DataPacket` for stream `s1`. +fn trailer_packet(identity: &str) -> Bytes { + let trailer = proto::data_stream::Trailer { stream_id: "s1".to_string(), ..Default::default() }; + let packet = proto::DataPacket { + participant_identity: identity.to_string(), + value: Some(proto::data_packet::Value::StreamTrailer(trailer)), + ..Default::default() + }; + Bytes::from(packet.encode_to_vec()) +} + +/// Captures the first opened text reader. +struct TextCapture(Mutex, String)>>>); + +impl IncomingDataStreamManagerDelegate for TextCapture { + fn on_byte_stream_opened(&self, _reader: Arc, _identity: String) {} + + fn on_text_stream_opened(&self, reader: Arc, identity: String) { + if let Some(tx) = self.0.lock().unwrap().take() { + let _ = tx.send((reader, identity)); + } + } + + fn on_stream_closed(&self, _stream_id: String, _identity: String) {} +} + +#[test] +fn incoming_inline_text_stream_roundtrips() { + crate::runtime::runtime().block_on(async { + let (tx, rx) = oneshot::channel(); + let delegate = Arc::new(TextCapture(Mutex::new(Some(tx)))); + let manager = IncomingDataStreamManager::new(delegate, None); + + manager.handle_packet_received( + inline_text_packet("alice", "my-topic", "hello world"), + EncryptionType::None, + ); + + let (reader, identity) = rx.await.expect("a stream should open"); + assert_eq!(identity, "alice"); + assert_eq!(reader.info().topic, "my-topic"); + assert_eq!(reader.read_all().await.unwrap(), "hello world"); + }); +} + +#[test] +fn incoming_chunk_with_mismatched_encryption_errors_reader() { + crate::runtime::runtime().block_on(async { + let (tx, rx) = oneshot::channel(); + let delegate = Arc::new(TextCapture(Mutex::new(Some(tx)))); + let manager = IncomingDataStreamManager::new(delegate, None); + + // The stream is announced unencrypted, but a chunk arrives claiming GCM encryption. + manager.handle_packet_received( + multipacket_text_header_packet("alice", "my-topic", 5), + EncryptionType::None, + ); + let (reader, _) = rx.await.expect("a stream should open"); + manager.handle_packet_received(chunk_packet("alice", 0, b"hello"), EncryptionType::Gcm); + + let result = reader.read_all().await; + assert!(matches!( + result, + Err(DataStreamError::EncryptionTypeMismatch { + expected: EncryptionType::None, + received: EncryptionType::Gcm, + }) + )); + }); +} + +#[test] +fn incoming_open_stream_count_tracks_headers_and_aborts() { + crate::runtime::runtime().block_on(async { + let (tx, _rx) = oneshot::channel(); + let delegate = Arc::new(TextCapture(Mutex::new(Some(tx)))); + let manager = IncomingDataStreamManager::new(delegate, None); + assert_eq!(manager.open_stream_count().await, 0); + + // The count query is processed in order with the packets enqueued before it, so this + // waits for the (orphaned) header to register without racing the run loop β€” exactly what + // exercising the abort paths requires. + manager.handle_packet_received( + multipacket_text_header_packet("alice", "my-topic", 5), + EncryptionType::None, + ); + assert_eq!(manager.open_stream_count().await, 1); + + manager.abort_all_streams(); + assert_eq!(manager.open_stream_count().await, 0); + }); +} + +/// Captures the first stream-closed notification. +struct ClosedCapture(Mutex>>); + +impl IncomingDataStreamManagerDelegate for ClosedCapture { + fn on_byte_stream_opened(&self, _reader: Arc, _identity: String) {} + + fn on_text_stream_opened(&self, _reader: Arc, _identity: String) {} + + fn on_stream_closed(&self, stream_id: String, identity: String) { + if let Some(tx) = self.0.lock().unwrap().take() { + let _ = tx.send((stream_id, identity)); + } + } +} + +#[test] +fn incoming_trailer_fires_stream_closed() { + crate::runtime::runtime().block_on(async { + let (tx, rx) = oneshot::channel(); + let delegate = Arc::new(ClosedCapture(Mutex::new(Some(tx)))); + let manager = IncomingDataStreamManager::new(delegate, None); + + manager.handle_packet_received( + multipacket_text_header_packet("alice", "my-topic", 5), + EncryptionType::None, + ); + manager.handle_packet_received(chunk_packet("alice", 0, b"hello"), EncryptionType::None); + manager.handle_packet_received(trailer_packet("alice"), EncryptionType::None); + + let (stream_id, identity) = rx.await.expect("the stream should close"); + assert_eq!(stream_id, "s1"); + assert_eq!(identity, "alice"); + }); +} + +#[test] +fn incoming_inline_stream_fires_stream_closed() { + // Inline single-packet streams never receive a trailer, so the closed signal must still fire + // once their payload is delivered. + crate::runtime::runtime().block_on(async { + let (tx, rx) = oneshot::channel(); + let delegate = Arc::new(ClosedCapture(Mutex::new(Some(tx)))); + let manager = IncomingDataStreamManager::new(delegate, None); + + manager.handle_packet_received( + inline_text_packet("alice", "my-topic", "hello world"), + EncryptionType::None, + ); + + let (stream_id, identity) = rx.await.expect("the stream should close"); + assert_eq!(stream_id, "s1"); + assert_eq!(identity, "alice"); + }); +} + +#[test] +fn incoming_abort_fires_stream_closed() { + crate::runtime::runtime().block_on(async { + let (tx, rx) = oneshot::channel(); + let delegate = Arc::new(ClosedCapture(Mutex::new(Some(tx)))); + let manager = IncomingDataStreamManager::new(delegate, None); + + // Announce a multi-packet stream, then abort before its trailer ever arrives. + manager.handle_packet_received( + multipacket_text_header_packet("alice", "my-topic", 5), + EncryptionType::None, + ); + manager.abort_all_streams(); + + let (stream_id, identity) = rx.await.expect("the stream should close"); + assert_eq!(stream_id, "s1"); + assert_eq!(identity, "alice"); + }); +} + +/// Collects every outbound packet the manager emits. +struct PacketCapture(Mutex>); + +impl OutgoingDataStreamManagerDelegate for PacketCapture { + fn on_packets_available(&self, packets: Vec) -> Result<(), PacketDeliveryError> { + self.0.lock().unwrap().extend(packets); + Ok(()) + } +} + +/// A room where every recipient is v2 and advertises deflate-raw compression. +struct AllV2Registry; + +impl RemoteParticipantRegistryDelegate for AllV2Registry { + fn remote_client_protocol(&self, _identity: String) -> i32 { + livekit_common::CLIENT_PROTOCOL_DATA_STREAM_V2 + } + + fn remote_capabilities(&self, _identity: String) -> Vec { + vec![ClientCapability::CompressionDeflateRaw] + } + + fn remote_identities(&self) -> Vec { + vec!["bob".to_string()] + } +} + +#[test] +fn outgoing_all_v2_text_inlines_compressed() { + crate::runtime::runtime().block_on(async { + let delegate = Arc::new(PacketCapture(Mutex::new(Vec::new()))); + let manager = OutgoingDataStreamManager::new(delegate.clone(), Arc::new(AllV2Registry)); + + let options = StreamTextOptions { + topic: "chat".to_string(), + destination_identities: vec!["bob".to_string()], + ..Default::default() + }; + let info = manager + .send_text("hello hello compressible world".to_string(), options) + .await + .expect("send_text should succeed"); + assert_eq!(info.topic, "chat"); + + // send_text awaits the transport responder, which the forward task fulfills only after + // invoking the delegate β€” so the packet is already captured here. + let packets = delegate.0.lock().unwrap(); + assert_eq!(packets.len(), 1, "expected a single inline header packet"); + + let decoded = proto::DataPacket::decode(packets[0].as_ref()).unwrap(); + let Some(proto::data_packet::Value::StreamHeader(header)) = decoded.value else { + panic!("expected a stream header packet"); + }; + assert_eq!(header.compression(), proto::data_stream::CompressionType::DeflateRaw); + let inline = header.inline_content.expect("inline content should be present"); + assert_ne!(inline.as_slice(), b"hello hello compressible world", "should be compressed"); + }); +} + +/// A transport delegate that accepts a fixed number of calls, then fails every subsequent one. +struct FailingTransport(std::sync::atomic::AtomicUsize); + +impl FailingTransport { + fn failing_after(successful_calls: usize) -> Self { + Self(std::sync::atomic::AtomicUsize::new(successful_calls)) + } +} + +impl OutgoingDataStreamManagerDelegate for FailingTransport { + fn on_packets_available(&self, _packets: Vec) -> Result<(), PacketDeliveryError> { + let remaining = &self.0; + if remaining + .fetch_update( + std::sync::atomic::Ordering::SeqCst, + std::sync::atomic::Ordering::SeqCst, + |n| n.checked_sub(1), + ) + .is_ok() + { + Ok(()) + } else { + Err(PacketDeliveryError::Failed { reason: "transport is down".to_string() }) + } + } +} + +/// Collects the batch boundaries of each delegate invocation. +struct BatchCapture(Mutex>); + +impl OutgoingDataStreamManagerDelegate for BatchCapture { + fn on_packets_available(&self, packets: Vec) -> Result<(), PacketDeliveryError> { + self.0.lock().unwrap().push(packets.len()); + Ok(()) + } +} + +#[test] +fn outgoing_send_failure_propagates() { + crate::runtime::runtime().block_on(async { + let delegate = Arc::new(FailingTransport::failing_after(0)); + let manager = OutgoingDataStreamManager::new(delegate, Arc::new(AllV2Registry)); + + let options = StreamTextOptions { topic: "chat".to_string(), ..Default::default() }; + let result = manager.send_text("hello".to_string(), options).await; + assert!(matches!(result, Err(DataStreamError::SendFailed))); + }); +} + +#[test] +fn outgoing_write_failure_errors_and_closes_writer() { + crate::runtime::runtime().block_on(async { + // Allow the header through, then fail: the failure lands on the write. + let delegate = Arc::new(FailingTransport::failing_after(1)); + let manager = OutgoingDataStreamManager::new(delegate, Arc::new(AllV2Registry)); + + let options = StreamTextOptions { topic: "chat".to_string(), ..Default::default() }; + let writer = manager.stream_text(options).await.expect("opening the stream should work"); + assert!(writer.is_open().await); + + let result = writer.write("hello".to_string()).await; + assert!(matches!(result, Err(DataStreamError::SendFailed))); + assert!(!writer.is_open().await, "a failed send should close the stream"); + }); +} + +#[test] +fn outgoing_one_shot_send_is_a_single_delegate_call() { + crate::runtime::runtime().block_on(async { + let delegate = Arc::new(BatchCapture(Mutex::new(Vec::new()))); + let manager = OutgoingDataStreamManager::new(delegate.clone(), Arc::new(PreV2Registry)); + + // 40 KB to a pre-v2 recipient: legacy framing, header + 3 chunks + trailer β€” the whole + // stream must arrive as ONE delegate call, not one call per packet. + let options = StreamTextOptions { + topic: "chat".to_string(), + destination_identities: vec!["bob".to_string()], + ..Default::default() + }; + manager.send_text("A".repeat(40_000), options).await.expect("send_text should succeed"); + assert_eq!(*delegate.0.lock().unwrap(), vec![5]); + }); +} + +/// A room where every recipient predates v2. +struct PreV2Registry; + +impl RemoteParticipantRegistryDelegate for PreV2Registry { + fn remote_client_protocol(&self, _identity: String) -> i32 { + livekit_common::CLIENT_PROTOCOL_DEFAULT + } + + fn remote_capabilities(&self, _identity: String) -> Vec { + vec![] + } + + fn remote_identities(&self) -> Vec { + vec!["bob".to_string()] + } +} + +/// Drives both managers through the pull adapters in [`super::polled`] β€” the path thread-affine +/// bindings take β€” and checks a payload survives the round trip. +async fn polled_roundtrip( + registry: Arc, + text: &str, +) -> (usize, String) { + let outgoing = super::polled::polled_outgoing_data_stream_manager(registry); + let incoming = super::polled::polled_incoming_data_stream_manager(None); + + let options = StreamTextOptions { + topic: "chat".to_string(), + destination_identities: vec!["bob".to_string()], + ..Default::default() + }; + outgoing.manager.send_text(text.to_string(), options).await.expect("send_text should succeed"); + + // send_text only resolves once every packet has been queued, so draining terminates rather + // than blocking. + let mut packet_count = 0; + while let Ok(Some(packets)) = + tokio::time::timeout(std::time::Duration::from_millis(200), outgoing.packets.next_packets()) + .await + { + for packet in packets { + packet_count += 1; + incoming.manager.handle_packet_received(packet, EncryptionType::None); + } + } + + let opened = incoming.streams.next_opened_stream().await.expect("a stream should open"); + let reader = opened.text_reader.expect("expected a text stream"); + (packet_count, reader.read_all().await.unwrap()) +} + +#[test] +fn polled_inlines_for_v2_recipients() { + crate::runtime::runtime().block_on(async { + let (packets, text) = + polled_roundtrip(Arc::new(AllV2Registry), "hello hello compressible world").await; + assert_eq!(packets, 1, "a v2 recipient should get a single inline packet"); + assert_eq!(text, "hello hello compressible world"); + }); +} + +#[test] +fn polled_falls_back_to_legacy_framing_for_pre_v2_recipients() { + crate::runtime::runtime().block_on(async { + let (packets, text) = polled_roundtrip(Arc::new(PreV2Registry), "hello world").await; + assert_eq!(packets, 3, "a pre-v2 recipient should get header + chunk + trailer"); + assert_eq!(text, "hello world"); + }); +} diff --git a/livekit-uniffi/src/lib.rs b/livekit-uniffi/src/lib.rs index 94f2cc0d0..d682f4ffe 100644 --- a/livekit-uniffi/src/lib.rs +++ b/livekit-uniffi/src/lib.rs @@ -15,6 +15,9 @@ /// Data tracks core from [`livekit-datatrack`]. pub mod data_track; +/// Data streams v2 core from [`livekit-data-stream`]. +pub mod data_stream; + /// Access token generation and verification from [`livekit-api::access_token`]. pub mod access_token; diff --git a/livekit-uniffi/uniffi.toml b/livekit-uniffi/uniffi.toml index 2ed2769f6..fbc7e1adc 100644 --- a/livekit-uniffi/uniffi.toml +++ b/livekit-uniffi/uniffi.toml @@ -6,6 +6,48 @@ android = true package_name = "io.livekit.uniffi" cdylib_name = "livekit_uniffi" # the name of the so file to be loaded +# The Kotlin checksum test is off because it fails on ARM devices. +# +# UniFFI gives each checksum function a return type in the Kotlin bindings. The bindings then +# compare the result with the expected checksum. Both available forms of this comparison are +# defective: +# +# - uniffi 0.31.1 uses the type Short. On 32-bit ARM, in release mode, the upper bits of a +# checksum above 32767 are incorrect. The test fails. See mozilla/uniffi-rs#2740. +# +# - uniffi 0.31.2 and 0.32.0 use the type Int, but they do not mask the upper 16 bits. On 64-bit +# ARM, Rust extends the sign of the 16-bit value. A checksum of 0x8000 or more then has 0xFFFF +# in its upper bits. The test fails. See https://github.com/mozilla/uniffi-rs/pull/2897. +# +# mozilla/uniffi-rs#2935 adds the mask, but no release contains that change. +# +# The AAR contains code for arm64-v8a and for armeabi-v7a. For this reason, one of the two defects +# applies to every uniffi release that this crate can use. +# +# It is safe to omit this test. The test finds a shared library that does not agree with the +# bindings. `cargo make android-package-local` builds the library and the bindings together, thus +# they always agree. Swift, Python, Dart and Node keep their checksum tests. +# +# Remove this line when a uniffi release contains mozilla/uniffi-rs#2935. +omit_checksums = true + +# Two names that are perfectly fine in Rust but do not compile in Kotlin. Renamed for Kotlin only, +# so the Rust source and the Swift/Python/Node bindings keep the original names. +[bindings.kotlin.rename] +# UniFFI gives every object a non-suspend `close()` to satisfy AutoCloseable. An exported Rust +# method also called `close` differs from it only by `suspend`, which Kotlin rejects as conflicting +# overloads -- and the whole generated file then fails to compile. Renaming the stream's own close +# leaves AutoCloseable's untouched. +"ByteStreamWriter.close" = "close_stream" +"TextStreamWriter.close" = "close_stream" +"IncomingStreamQueue.close" = "close_stream" +"OutgoingPacketQueue.close" = "close_stream" + +# DataStreamError's variant fields cannot be renamed from here: uniffi keys the rename table by +# crate name but looks up enum and record members by the item's full module path, so a rename for +# anything declared in a submodule is silently ignored (methods use the crate name and do work). +# Those fields are named in Rust instead -- see data_stream/common.rs. + [bindings.dart] # Dart package name; must match the published pub package so the Native Assets # asset id resolves to `package:livekit_uniffi/uniffi:livekit_uniffi`. diff --git a/livekit/src/room/mod.rs b/livekit/src/room/mod.rs index ad132771b..dfba72faa 100644 --- a/livekit/src/room/mod.rs +++ b/livekit/src/room/mod.rs @@ -722,10 +722,7 @@ impl Room { dt::remote::Manager::new(remote_dt_options); let (incoming_stream_manager, incoming_data_stream_input, incoming_output) = - ds::incoming::Manager::new( - INTERNAL_DATA_STREAM_TOPICS.into(), - options.data_stream.max_payload_byte_length, - ); + ds::incoming::Manager::new(options.data_stream.max_payload_byte_length); let (outgoing_stream_manager, packet_rx) = ds::outgoing::Manager::new(); let room_info = join_response.room.unwrap(); @@ -2432,12 +2429,21 @@ async fn incoming_data_stream_task( } } }, - ds::incoming::OutputEvent::ChunkReceived(ds::incoming::ChunkReceived { chunk, participant_identity }) => { - dispatcher.dispatch(&RoomEvent::StreamChunkReceived { chunk: chunk.into(), participant_identity: participant_identity.into() }); + // Chunk/trailer packets carry no topic of their own, so the manager reports the + // topic of the stream they belong to for the internal check below. + ds::incoming::OutputEvent::ChunkReceived(ds::incoming::ChunkReceived { chunk, participant_identity, topic }) => { + if !topic.as_deref().is_some_and(is_internal_topic) { + dispatcher.dispatch(&RoomEvent::StreamChunkReceived { chunk: chunk.into(), participant_identity: participant_identity.into() }); + } } - ds::incoming::OutputEvent::TrailerReceived(ds::incoming::TrailerReceived { trailer, participant_identity }) => { - dispatcher.dispatch(&RoomEvent::StreamTrailerReceived { trailer: trailer.into(), participant_identity: participant_identity.into() }); + ds::incoming::OutputEvent::TrailerReceived(ds::incoming::TrailerReceived { trailer, participant_identity, topic }) => { + if !topic.as_deref().is_some_and(is_internal_topic) { + dispatcher.dispatch(&RoomEvent::StreamTrailerReceived { trailer: trailer.into(), participant_identity: participant_identity.into() }); + } } + // The Rust SDK observes completion through the reader itself; the explicit + // closed signal exists for FFI hosts sequencing handlers on ordered topics. + ds::incoming::OutputEvent::StreamClosed(_) => {} }, _ = close_rx.recv() => { _ = session.incoming_data_stream_input.send(ds::incoming::InputEvent::Shutdown); @@ -2448,33 +2454,40 @@ async fn incoming_data_stream_task( } /// Data stream topics reserved for internal SDK use (e.g. RPC). Events for these topics are -/// handled within the `livekit` crate and never surfaced through `RoomEvent`; the list is also -/// passed to `IncomingStreamManager` so it can flag internal streams. +/// handled within the `livekit` crate and never surfaced through `RoomEvent`. const INTERNAL_DATA_STREAM_TOPICS: &[&str] = &[rpc::RPC_REQUEST_TOPIC, rpc::RPC_RESPONSE_TOPIC]; fn is_internal_topic(topic: &str) -> bool { INTERNAL_DATA_STREAM_TOPICS.contains(&topic) } -/// Receives packets from the outgoing stream manager and send them. +/// Receives packet batches from the outgoing stream manager and send them. async fn outgoing_data_stream_task( - mut packet_rx: UnboundedRequestReceiver>, + mut packet_rx: UnboundedRequestReceiver, Result<(), SendError>>, engine: Arc, mut close_rx: broadcast::Receiver<()>, ) { loop { tokio::select! { - Ok((packet, responder)) = packet_rx.recv() => { - // A packet stamped with an explicit sender identity (impersonation, e.g. an - // agent attributing a stream to another participant) must be sent raw so the - // session doesn't overwrite the identity with the local participant's. - let is_raw_packet = !packet.participant_identity.is_empty(); - // Bridge the engine error into the data-stream crate's opaque `SendError` - // (the crate only needs to know whether the send failed). - let result = engine - .publish_data(packet, DataPacketKind::Reliable, is_raw_packet) - .await - .map_err(|_| SendError); + Ok((packets, responder)) = packet_rx.recv() => { + // The batch is acknowledged as a whole; the first failure fails the request. + let mut result = Ok(()); + for packet in packets { + // A packet stamped with an explicit sender identity (impersonation, e.g. an + // agent attributing a stream to another participant) must be sent raw so the + // session doesn't overwrite the identity with the local participant's. + let is_raw_packet = !packet.participant_identity.is_empty(); + // Bridge the engine error into the data-stream crate's opaque `SendError` + // (the crate only needs to know whether the send failed). + if engine + .publish_data(packet, DataPacketKind::Reliable, is_raw_packet) + .await + .is_err() + { + result = Err(SendError); + break; + } + } let _ = responder.respond(result); }, _ = close_rx.recv() => {