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/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..37a9cdf16 100644 --- a/livekit-data-stream/src/incoming/events.rs +++ b/livekit-data-stream/src/incoming/events.rs @@ -39,6 +39,9 @@ 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, /// Stop the run loop. Shutdown, } @@ -55,6 +58,10 @@ pub struct StreamOpened { 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 +69,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 diff --git a/livekit-data-stream/src/incoming/manager.rs b/livekit-data-stream/src/incoming/manager.rs index af257b0e7..fb979b255 100644 --- a/livekit-data-stream/src/incoming/manager.rs +++ b/livekit-data-stream/src/incoming/manager.rs @@ -46,7 +46,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`. @@ -184,10 +186,6 @@ pub struct Manager { 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, } @@ -199,7 +197,6 @@ struct ManagerInner { 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 @@ -211,7 +208,6 @@ impl Manager { input_rx, output_tx, - reserved_topics, max_payload_byte_length: max_payload_byte_length .unwrap_or(DEFAULT_MAX_PAYLOAD_BYTE_LENGTH), }; @@ -238,6 +234,7 @@ impl Manager { } } InputEvent::AbortStreamsFrom(identity) => self.handle_abort(identity), + InputEvent::AbortAllStreams => self.handle_abort_all(), InputEvent::Shutdown => break, } } @@ -250,7 +247,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, @@ -341,7 +338,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 +348,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,12 +364,11 @@ 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.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 { @@ -476,11 +466,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.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 +519,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 { @@ -681,16 +684,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 } } @@ -732,7 +731,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 +756,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 +772,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 +802,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 +818,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 +835,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 +853,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 +865,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 +881,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, @@ -897,7 +896,7 @@ mod tests { #[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 +933,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 +953,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 +964,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 +984,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 +1004,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 +1025,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 +1036,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 +1059,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 +1089,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 +1106,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 +1130,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 +1164,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 +1191,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 +1219,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 +1244,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 +1268,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 +1307,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 +1339,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 +1378,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 +1446,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 +1473,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 +1507,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 { @@ -1532,7 +1530,7 @@ mod tests { #[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 +1558,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 +1580,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 +1617,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; + } + OutputEvent::StreamOpened(_) => 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/raw_stream.rs b/livekit-data-stream/src/outgoing/raw_stream.rs index 8f9055fe5..6688eccf1 100644 --- a/livekit-data-stream/src/outgoing/raw_stream.rs +++ b/livekit-data-stream/src/outgoing/raw_stream.rs @@ -63,12 +63,21 @@ 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(()) @@ -155,8 +164,9 @@ 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(()) } 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-uniffi/Cargo.toml b/livekit-uniffi/Cargo.toml index f8a915ee4..ffefb17c3 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"] } diff --git a/livekit-uniffi/src/data_stream/common.rs b/livekit-uniffi/src/data_stream/common.rs new file mode 100644 index 000000000..d185f231e --- /dev/null +++ b/livekit-uniffi/src/data_stream/common.rs @@ -0,0 +1,370 @@ +// 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, + } + } +} + +/// 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")] + EncryptionTypeMismatch, + + #[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, +} + +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 => Self::EncryptionTypeMismatch, + 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. +/// +/// Encryption is defaulted to `None`: end-to-end encryption for data streams over this FFI is a +/// follow-up (the foreign side is expected to hand us already-decrypted packets). +pub(crate) fn decode_data_packet(bytes: &[u8]) -> 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: common::EncryptionType::None, + }, + proto::data_packet::Value::StreamChunk(chunk) => { + ds::Packet::Chunk { chunk: chunk.into(), encryption_type: common::EncryptionType::None } + } + 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..e722dadf6 --- /dev/null +++ b/livekit-uniffi/src/data_stream/incoming.rs @@ -0,0 +1,216 @@ +// 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, 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-open events 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); +} + +#[uniffi::export] +impl IncomingDataStreamManager { + #[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. + pub fn handle_packet_received(&self, packet: Bytes) { + if let Some(event) = decode_data_packet(&packet) { + 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())); + } +} + +/// 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); + } + } + } + // 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..72b0aa1ed --- /dev/null +++ b/livekit-uniffi/src/data_stream/mod.rs @@ -0,0 +1,30 @@ +// 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. + +pub mod common; +pub mod incoming; +pub mod outgoing; + +#[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..cb9e2fd07 --- /dev/null +++ b/livekit-uniffi/src/data_stream/outgoing.rs @@ -0,0 +1,221 @@ +// 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, 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. + fn on_packets_available(&self, packets: Vec); +} + +/// 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 to the transport delegate and acknowledge the send. Wire + // send-failures are not propagated back to the originating `send_*` call for now (matches + // the data-track delegate); can be upgraded to a Result-returning delegate later. + 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((packet, responder)) => { + delegate.on_packets_available(vec![Bytes::from(packet.encode_to_vec())]); + let _ = responder.respond(Ok(())); + } + 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/tests.rs b/livekit-uniffi/src/data_stream/tests.rs new file mode 100644 index 000000000..45fccd519 --- /dev/null +++ b/livekit-uniffi/src/data_stream/tests.rs @@ -0,0 +1,140 @@ +// 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, 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()) +} + +/// 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)); + } + } +} + +#[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")); + + 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"); + }); +} + +/// Collects every outbound packet the manager emits. +struct PacketCapture(Mutex>); + +impl OutgoingDataStreamManagerDelegate for PacketCapture { + fn on_packets_available(&self, packets: Vec) { + self.0.lock().unwrap().extend(packets); + } +} + +/// 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"); + }); +} 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..71b6d20fe 100644 --- a/livekit-uniffi/uniffi.toml +++ b/livekit-uniffi/uniffi.toml @@ -6,6 +6,21 @@ android = true package_name = "io.livekit.uniffi" cdylib_name = "livekit_uniffi" # the name of the so file to be loaded +# 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" + +# 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..09f028912 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,11 +2429,17 @@ 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() }); + } } }, _ = close_rx.recv() => { @@ -2448,8 +2451,7 @@ 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 {