Data streams v2 uniffi

.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)

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

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]' = <object object at 0x10089cc40>,
+        # id: 'typing.Optional[str]' = <object object at 0x10089cc40>,
+        # operation_type: 'typing.Optional[OperationType]' = <object object at 0x10089cc40>,
+        # version: 'typing.Optional[int]' = <object object at 0x10089cc40>,
+        # reply_to_stream_id: 'typing.Optional[str]' = <object object at 0x10089cc40>,
+        # attached_stream_ids: 'typing.List[str]' = <object object at 0x10089cc40>,
+        # generated: 'typing.Optional[bool]' = <object object at 0x10089cc40>,
+        # compress: 'typing.Optional[bool]' = <object object at 0x10089cc40>,
+        # sender_identity: 'typing.Optional[str]' = <object object at 0x10089cc40>
+    ))
+
+    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())

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<usize>),
     /// Stop the run loop.
     Shutdown,
 }
@@ -50,25 +58,50 @@ 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<String>,
 }
 
 /// A "raw trailer received" notification, which is used to trigger
 /// the deprecated [RoomEvent:::StreamTrailerReceived] event.
 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<String>,
 }
 
 /// An event emitted by [`IncomingStreamManager::run`] for the host crate to surface. The manager
 /// stays decoupled from `RoomEvent`; the host maps these onto its own event types.
 #[derive(FromVariants)]
 pub enum OutputEvent {
     StreamOpened(StreamOpened),
+    StreamClosed(StreamClosed),
     ChunkReceived(ChunkReceived),
     TrailerReceived(TrailerReceived),
 }