feat(data-stream): abort_all_streams / abort_streams_from on incoming FFI

Add an AbortAllStreams input event to the incoming data-stream backend and
expose abort_all_streams() / abort_streams_from(identity) on the UniFFI
IncomingDataStreamManager, so the host can fail open readers on disconnect or
participant-leave instead of letting them hang (which would also stall an
ordered topic's queue).

Also fix the constructor to take Option<u64> (UniFFI can't lift usize) and drop
the stale reserved_topics argument to Manager::new.

Author: 1egoman

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,
 }

livekit-data-stream/src/incoming/manager.rs

@@ -234,6 +234,7 @@ impl Manager {
                     }
                 }
                 InputEvent::AbortStreamsFrom(identity) => self.handle_abort(identity),
+                InputEvent::AbortAllStreams => self.handle_abort_all(),
                 InputEvent::Shutdown => break,
             }
         }
@@ -518,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 {

livekit-uniffi/src/data_stream/incoming.rs

@@ -52,10 +52,13 @@ pub trait IncomingDataStreamManagerDelegate: Send + Sync {
 #[uniffi::export]
 impl IncomingDataStreamManager {
     #[uniffi::constructor]
-    pub fn new(delegate: Arc<dyn IncomingDataStreamManagerDelegate>, max_payload_byte_length: Option<usize>) -> Arc<Self> {
+    pub fn new(
+        delegate: Arc<dyn IncomingDataStreamManagerDelegate>,
+        max_payload_byte_length: Option<u64>,
+    ) -> Arc<Self> {
         let token = CancellationToken::new();
-        // No reserved topics: RPC routing is a concern of the `livekit` crate, not this FFI layer.
-        let (manager, input, output) = ds::incoming::Manager::new(vec![], max_payload_byte_length);
+        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()));
@@ -75,6 +78,19 @@ impl IncomingDataStreamManager {
             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.