Repository navigation
Data streams: trailer attributes, progress, reader close, and declared text length over the FFI - #1558
Open
pblazej wants to merge 33 commits into
Open
Data streams: trailer attributes, progress, reader close, and declared text length over the FFI#1558pblazej wants to merge 33 commits into
pblazej wants to merge 33 commits into
Conversation
`cargo clippy -- -D warnings` failed on both crates before any data stream change. Mechanical fixes only, with no change in behavior: redundant closures and conversions, a needless return, a no-op struct update, and a type alias for a complex test type. `DataPacket::kind` is deprecated but still set, so its three uses allow the lint. `InputEvent` keeps its large variant, because boxing it changes a public type. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
The FFI readers converted the stream info once, when they were created, so attributes that the sender put in the trailer never reached a binding. The readers now keep the core info, whose attribute map the incoming manager updates, and convert it on each `info()` call. The core merges trailer attributes before it ends the stream, so the final attributes are visible when `next()` returns `None` or `readAll()` returns. This matches JS, where the trailer updates `reader.info` before the stream closes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
The core writers can send attributes in the stream trailer, but the FFI writers exported only `close()` and `close_with_reason()`. Both writers now export `close_with_options(reason, attributes)`, a thin wrapper over the core method. `attributes` is a plain map rather than an optional one: the trailer carries an empty map either way, so `None` and an empty map would be two spellings of the same packet. The name does not collide with the `close()` that UniFFI generates for Kotlin. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
Incoming: each FFI reader exports `next_progress()`, an async method that returns the current progress on the first call and then waits for the next change. It returns `None` once the stream ends. It wraps the core progress stream, so it counts bytes as they arrive, after decompression, and updates between two calls merge into the latest one. An async method rather than a callback keeps it usable from Dart, which cannot take a callback from a tokio thread (see `polled.rs`). For the FFI to hold that stream next to the reader, the core `StreamReader::progress()` now returns `impl Stream + Send + 'static`. The watch stream behind it already was; the old signature only tied it to `&self`. Outgoing: a new `StreamProgressDelegate` callback trait and three new methods, `send_text_with_progress`, `send_bytes_with_progress` and `send_file_with_progress`. The callback runs inside the send future, on the thread that polls it, like `RemoteParticipantRegistryDelegate`. - A file reports after each chunk is sent, counting file bytes before compression, through the new core `send_file_with_progress`. An empty file reports `0 / 0` once, so the last report always covers the file. - One-shot text and byte sends go to the transport as one request, so they report once, when the send completes. This follows JS, which counts pre-compression bytes and always fires a final 100% event. JS reports a 0..1 fraction; the FFI reports byte counts with an optional total, like the core `StreamProgress`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
`StreamTextOptions` had no total length, and `stream_text` passed `None` to `build_text_header`, so a sender that knew the final length of its text could not put it in the header. The core options and the FFI record now carry an optional `total_length`, which `stream_text` puts in the header. The receiver already checks a declared length (`Incomplete`, `LengthExceeded`) and reports progress against it. `send_text` ignores the field, because it sets the length from the text it sends. The room-apis protobuf options have no such field, so their conversion passes `None`. The FFI field goes last with `default = None`, so existing Swift and Kotlin call sites that build the record keep compiling. JS and Flutter call the same option `totalSize`; the FFI keeps the core name, which the byte options already use. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
A binding had no way to stop one reader: cancelling a pending `next()` or `readAll()` on the foreign side left the native read running, and the reader kept its receiver and buffered chunks until the stream ended or the room disconnected. A Rust caller avoids this by dropping the reader. Both readers now export a synchronous `close()`: - It cancels a per-reader token. A pending `next()` or `readAll()` wakes and fails with `DataStreamError::AlreadyClosed`, as does every later read, and `next_progress()` returns `None`. - The core reader is dropped, which frees its buffered chunks and closes its channel. The incoming manager then drops the stream at its next packet and reports it closed. Other streams are not affected. - `close()` never blocks, so a Swift `onCancel` handler can call it. A read in progress holds the reader lock until it sees the cancellation, so `close()` drops the reader from a short task that takes the lock after that read. There is no `Cancelled` error variant, and adding one would break exhaustive Kotlin `when` expressions over `DataStreamException`, so the read fails with the existing `AlreadyClosed`. JS errors the read with the abort reason; Python ends it like end-of-stream. Kotlin gets `closeStream()`, like the writers, because UniFFI already generates a non-suspend `close()` for `AutoCloseable`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
For a stream that arrives in one packet, the incoming manager checked the payload against the receive size limit but not against `total_length` in the header. A header that declared five bytes with three inline bytes ended with success. The chunked path already fails that stream at its trailer. The inline branch now compares the decoded byte count, after decompression for a compressed payload, with `total_length` when the header has one. A short payload fails with `StreamError::Incomplete` and a long one with `StreamError::LengthExceeded`, so no new error variant is needed. A header without a length is not checked, and matching payloads, compressed or not, read as before. JS rejects a short inline stream with `Incomplete` too. Unlike JS, a declared length of 0 is held to the payload, as the chunked path here already does. Every sender that inlines sets the real length, so this rejects no stream that a conforming sender produces. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
`cb688a63` changed the public trait method to return `impl Stream + Send + 'static`, so that the FFI could hold the progress stream next to the reader. That is a stricter contract on a public trait. The trait method is back to `impl Stream<Item = StreamProgress>`. The concrete readers get an inherent `progress_stream()` instead, which returns a `Send + Unpin + 'static` stream that does not borrow the reader. The trait implementations call it. The FFI readers use it. With the trait unchanged, the core part of the progress changeset is additive, so `livekit-data-stream` goes from `minor` to `patch` there. `livekit-ffi` stays `minor`. The `next_progress()` docs now say that progress counts the bytes received from the sender, not the bytes the app has read. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
`close()` on an FFI reader cancelled its reads and dropped the core reader, but the incoming manager learned about it only when a later chunk failed to deliver. If the sender sent nothing more, the stream's descriptor, decompressor and attributes stayed registered, and `on_stream_closed` never fired, which can block ordered handlers. `close()` now enqueues a targeted event: - Each reader the manager creates carries a `StreamCloser`: a weak handle to the manager's input, the stream id, and an instance number that the manager assigns per opened stream. `closer()` on the concrete core readers returns it. - `StreamCloser::close()` sends `InputEvent::CloseStream`. The manager removes the descriptor and emits `StreamClosed` only when the open stream with that id has the same instance number. So a repeated or late close, or a close that loses a race with the trailer, does nothing, and a later stream that reuses the id is never touched. - The stream ends with `StreamError::AlreadyClosed` in its chunk channel, so a Rust reader that is still reading fails instead of ending as if the stream were complete. - The FFI reader takes its closer when it is created, so `close()` enqueues synchronously even while a read holds the reader lock. The handle is weak, so a closer does not keep the manager running. Tests: closing with no further packets drops the open count and fires one close event while another stream keeps working, for text and byte readers. A repeated close fires one event. Close racing the final chunk and the trailer, in each order, fires one event. A late close spares a later stream with the same id. The FFI tests poll the read to `Pending` before closing instead of sleeping, and count close events against an inline marker stream, which closes last. The new `InputEvent` variant breaks exhaustive matches on the enum, and `livekit` re-exports the crate, so the changeset marks both `minor`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
- `incoming_progress_merges_updates_between_calls`: two chunks that arrive between two `next_progress()` calls yield only the latest value. - `incoming_progress_ends_when_the_stream_fails`: a stream that exceeds its declared length ends its progress with `None`, and the read fails with `LengthExceeded`. - `incoming_trailer_attributes_replace_header_attributes`: a trailer key replaces the header's value for the same key, and other header keys stay. - `v2_inline_non_empty_payload_with_declared_zero_length_fails`: the core holds a declared length of 0 to an inline payload, compressed or not. JS skips the check for 0. No sender inlines a non-empty payload with a length of 0, so this documents a deliberate difference from JS. These tests pin down behavior that the earlier commits already have. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
The new public `total_length` field breaks struct-literal construction of the core `StreamTextOptions`, which is not `#[non_exhaustive]`. `livekit` re-exports the type, and the public `RpcTransport::send_text` in `livekit-rpc` takes it, so both go from `patch` to `minor` in that changeset. `livekit-capture` exposes neither and stays `patch`. The writer changeset now says that the new data stream methods add requirements to the generated Swift protocols and Kotlin interfaces, so a hand-written conformer must implement them, which is why `livekit-ffi` takes a minor bump in every changeset that adds methods. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
Two regression tests for a closer used after the incoming manager's run loop stopped. Each test owns the run loop, waits for it to stop, and then calls `close()` twice. - `closer_after_shutdown_with_input_dropped_does_nothing`: the last `ManagerInput` is dropped, so the closer's weak handle cannot upgrade. - `closer_after_shutdown_with_input_retained_does_nothing`: the input is kept and the manager stops on `InputEvent::Shutdown`, so the send goes to a channel that nobody receives from. In both, `close()` returns without a panic, the only close event is the one from the shutdown abort, and the reader fails with `AbnormalEnd`. `close()` is synchronous, so returning shows that it does not block. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
`20523607` added a public `total_length` field to the core
`StreamTextOptions`. All its fields are public and it is not
`#[non_exhaustive]`, so the field broke exhaustive struct literals, and
`livekit` re-exports the type and `livekit-rpc` takes it in
`RpcTransport::send_text`.
The core `StreamTextOptions` is now exactly as on `524123ee`, and so is
the room-apis conversion that filled the field. The declared length goes
through a new method instead, which is purely additive:
Manager::stream_text_with_total_length(options, total_length)
`stream_text(options)` calls it with `None`. The `core-modules`
`StreamTextOptions` record keeps its trailing `total_length = None`
field, which is additive for Swift and Kotlin callers, and the FFI
`stream_text` passes that field to the new method. `send_text` still
uses the length of the text it sends.
The changeset now lists `livekit-rpc` as `patch`, since nothing it
exposes changes.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
`f367f24c` added a public `InputEvent::CloseStream` variant. `InputEvent` is public and not `#[non_exhaustive]`, so the variant broke exhaustive matches, and `livekit` re-exports the crate. `InputEvent` is now as on `524123ee`, apart from the clippy `allow` from `d8343986`. The manager's input queue carries a crate-private `ManagerMessage` instead: `Event(InputEvent)` or `CloseStream(StreamCloser)`. `ManagerInput::send` keeps its signature and wraps the event, and so does the drop guard. `StreamCloser` holds a weak handle to the same queue. Using one queue keeps a close in order with the packets of its stream, which a second channel that the run loop selects on would not. The instance-number check, the weak handle, the synchronous `close()`, the single close event and the shutdown behavior do not change, and the existing tests pass unchanged. `ManagerMessage` allows `clippy::large_enum_variant` like `InputEvent`, because boxing the event would allocate for every inbound packet. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
The branch no longer breaks public Rust API, so no changeset needs a breaking level. In this repo's knope setup, `major` on a 0.x crate bumps the middle number and is the breaking level: `livekit-ffi` 0.12.83 went to 0.13.0 on a `major` changeset. `minor` bumps only the last number: `livekit-ffi` 0.12.82 went to 0.12.83 and `livekit-data-stream` 0.1.0 went to 0.1.1 on `minor` changesets. Recent additive features (capture sources, frame metadata user data, data streams v2 for `livekit-data-stream`) use `minor`, and fixes and dependents use `patch`. So each changeset now uses: - `minor` for a crate that gains public API: `livekit-data-stream`, `livekit-ffi`, and `livekit`, whose root re-exports the data stream `api` module and whose `data_stream` module re-exports the crate. - `patch` for a fix (reader trailer attributes, the inline length check) and for a crate that only depends on the change (`livekit-capture`, `livekit-rpc`). The progress changeset moves `livekit-data-stream` and `livekit` to `minor` for that reason. The writer changeset keeps its sentence about new methods adding requirements to the generated Swift protocols and Kotlin interfaces, without tying it to a bump level. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
`livekit-ffi` is not published, so its Rust API may change, and UniFFI
0.31 renders argument defaults in every binding the repo generates.
The FFI therefore extends existing methods instead of adding siblings:
- `send_text`, `send_bytes` and `send_file` take a trailing
`progress: Option<Arc<dyn StreamProgressDelegate>>`, with
`#[uniffi::method(default(progress = None))]`. This replaces
`send_text_with_progress`, `send_bytes_with_progress` and
`send_file_with_progress`. The behavior is the same: one completion
report for the one-shot sends, a report per chunk for files.
- Both writers' `close` takes `reason: Option<String>` and
`attributes: HashMap<String, String>`, with
`default(reason = None, attributes)`. `default(attributes)` uses the
type's default, an empty map: `[:]` in Swift, `mapOf()` in Kotlin and
`const {}` in Dart. This replaces `close_with_options`. With no
arguments it sends the same trailer as before. `close_with_reason`
stays as on the base.
Generated Swift, Kotlin and Dart keep the old call forms working:
`sendText(text:options:)`, `close()`, `closeStream()` in Kotlin.
`stream_text` still reads `total_length` from the record field, which
already defaults to `None`. The reader `close()` and `next_progress()`
stay new methods, because there is nothing to default.
The core keeps its additive methods, since Rust has no default
arguments. The FFI tests pass `None` or an empty map where Rust needs
the full argument list. The progress and writer changesets describe
the parameters.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
A compressed `send_file_with_progress` reported after every chunk with the bytes read so far. When one read released several chunks, the value repeated, and when the last read released a chunk, the whole file was reported before the tail was sent. A 50 KB random file reported `[32768, 40960, 50000, 50000]`, and still reported 50000 when the last chunk failed. The compressed branch now reports once after each drain of the encoder, and only while the bytes read are below the file size. It sends the tail with `write_chunk` and reports the whole file once, after the last chunk was sent. The same file now reports `[32768, 40960, 50000]`. The uncompressed path does not change. Two new tests record what the transport and the callback saw, in order. One checks that the values rise strictly and that the total comes after the last chunk. The other fails the request that carries the last chunk and checks that the total is never reported. The existing compressed progress test expected one report per chunk, so it now asks for at least two strictly rising reports. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
A host that released a reader without calling `close()` left the stream registered in the incoming manager until a later packet failed to deliver, so with no further packets `on_stream_closed` never fired. Both FFI readers now implement `Drop`, which sends the same targeted close as `close()` unless `close()` already ran. The closer only sends on a channel, so this works on any thread, such as a Swift `deinit` or a Kotlin cleaner. After a normal end the manager ignores it, because no open stream has that instance number. The new test drops an unread reader with no further packets and sees the open count drop and one close event. Dropping a reader whose stream already ended adds no event. The FFI tests passed 30 of 30 repeated runs, including the older tests whose delegates drop readers at once. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
A second `close()` on an FFI reader sent another close message to the incoming manager and spawned another task to drop the core reader, both of which did nothing. `close_reader` now returns at once when the reader is already closed, and it sends the manager message itself, so both readers share the check. The repeated-close test still sees one close event. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
`on_progress` returns no error, so UniFFI turns an exception from a Kotlin or Dart implementation into a Rust panic, and the release profile, which sets `panic = "abort"`, ends the process. The docstring now says so; it reaches the generated Swift, Kotlin and Dart. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
A test inside `incoming.rs`, where the private `inner` field is visible. It starts a read on a stream with no data, waits until that read holds the lock on the core reader, calls `close()`, and then waits, with a timeout, until `inner` is `None`. The read ends with `AlreadyClosed`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
#1536 made every chunk and trailer carry the stream's destination identities. This branch adds three send paths: closing a writer with trailer attributes, compressed progress reporting in `write_file`, and the FFI writer `close(reason, attributes)`. All three build packets through `RawStream::write_chunk` or `RawStream::close`, which stamp the destinations. These tests pin that down next to the #1536 tests: - `writers_closed_with_trailer_attributes_target_every_packet`: a text writer to two recipients closed with attributes, and a byte writer closed with a reason and attributes. - `writer_dropped_without_close_targets_its_trailer`: the trailer that a dropped writer sends from its spawned task. - `compressed_send_file_with_progress_targets_every_packet`: a 50 KB compressed file, whose chunks go out both during the read and in the tail. - The FFI `outgoing_writers_send_trailer_attributes` now targets "bob", and checks that all five header, chunk and trailer packets carry it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
Contributor
Changeset ✓This PR includes a changeset covering all affected packages:
|
- Remove `incoming_inline_stream_shorter_than_declared_errors_reader`: the core inline length test proves the check, and `outgoing_text_stream_declares_total_length` proves the FFI mapping. - Remove `incoming_reader_info_shows_trailer_attributes`: `incoming_trailer_attributes_replace_header_attributes` covers it. - Remove `incoming_reader_closed_twice_reports_one_close`, and close the byte reader twice in `incoming_byte_reader_close_ends_its_stream`. - Remove the close-racing and id-reuse tests; the core `closer` tests own those cases. - Remove `incoming_progress_merges_updates_between_calls`, which tested the tokio watch channel. - Replace the `poll_once` helper with `FutureExt::now_or_never`. `incoming_progress_ends_when_the_stream_fails` stays: a progress consumer must not hang when its stream fails. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
`next_chunk` returns straight from its biased `select!` arms. It no longer drops the core reader itself when it sees the close; the task that `close_reader` spawns does that once the read releases the lock, which the test below proves. The test in `incoming.rs` now reuses `manager_with_events`, `open_text_stream` and `TIMEOUT` from the data stream tests, which become `pub(super)`. It polls the read once with `now_or_never()` to reach the pending state, instead of a hand-built header, a separate delegate, and a spawned read with a `try_lock` spin. It still waits until `inner` is `None`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
- The four inline length tests become one loop over compression that keeps every assertion: short is `Incomplete`, long and a declared 0 are `LengthExceeded`, and a matching or undeclared length reads. - The two shutdown tests become `closer_after_shutdown_does_nothing`, which covers both weak-handle branches in one manager: the send fails while the input is kept after `Shutdown`, and the upgrade fails after the input is dropped. `wait_for_stop`, `count_closes` and the open helper are inlined. - Remove `send_file_reports_progress_before_compression`. Its "at least two reports" assertion moves into `compressed_send_file_reports_rising_progress_and_the_total_last`. - Remove `writers_closed_with_trailer_attributes_target_every_packet`, since the FFI `outgoing_writers_send_trailer_attributes` covers the new close path, and `writer_dropped_without_close_targets_its_trailer`, which tested code this branch does not touch. `compressed_send_file_with_progress_targets_every_packet` stays: it is the only destination check on the rewritten compressed path. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
`StreamCloser::new` only assigned three fields. The manager and `detached()` now use struct literals, and `input` is `pub(crate)` like the other two fields. The public type and `close()` do not change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
`write_file_chunk` had two callers, both in the uncompressed branch of `write_file`, which now call `write_chunk` and report progress inline. The empty-file report moves into that branch, so the compressed branch no longer returns early. The reported values do not change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
pblazej
marked this pull request as ready for review
October 22, 2026 10:58
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
`handle_trailer` checked the declared total length before the trailer's reason. A sender that declared a length, sent part of it and closed with a reason, such as a cancelled `stream_text` with `total_length`, gave the reader `Incomplete` instead of `AbnormalEnd(reason)`. The order dates from before this branch and already affected byte streams with a declared length. A non-empty reason now ends the stream with `AbnormalEnd` first. This matches JS, where the trailer's reason errors the stream and the reader's length check runs only on a normal end. A short stream that closes without a reason is still `Incomplete`, as the existing `v1_errors_when_too_few_bytes` and `v2_compressed_errors_when_too_few_bytes` tests check. The new test closes a 10-byte stream after 5 bytes with a reason, uncompressed and compressed. It failed before with `Err(Incomplete)`. Inline streams have no trailer, so they are not affected. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
For a compressed file, a progress report counts the file bytes the deflate encoder consumed, and the encoder can still hold part of them, so a report can run ahead of the chunks sent. This is by design: - Progress counts pre-compression input, as in JS, whose progress tap sits before the compressor. - Knowing exactly which input a sent chunk carries would need a sync flush per chunk, which costs compression ratio. - The whole file is reported only after its last chunk was sent (736c1a1), so a failed send never reports completion. The core `send_file_with_progress` and the FFI `send_file` docs now say so, and the internal `write_file` doc no longer claims that every reported byte went out. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
The send_file docstrings said that progress is reported after each chunk. That is true only for an uncompressed file. A compressed file reports after each encoder drain. In both modes, the whole file is reported only after its last chunk was sent. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
The closer test now also closes a stream before its final chunk, and between the final chunk and the trailer. In both orders, the manager reports one close, the read fails with AlreadyClosed and no other error, and the stream ends. The FFI test that covered these orders was removed in an earlier trim, so this test is now their only coverage. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
trailer_with_reason_errors_abnormal_end now declares 10 bytes and sends 5, so it also checks that the trailer's reason wins over the short length. It fails with the old check order, so the separate trailer_reason_wins_over_a_short_declared_length test is deleted. Its compressed case tested no other branch, because handle_trailer has no compression path. The closer test now names its "trailer" step, and an unknown step panics, so a typo in the step table cannot pass as a trailer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
The crate-private write_file doc repeated the progress rules of Manager::send_file_with_progress. It now links to that method and keeps only the two facts that the method does not state: the values rise strictly, and an empty uncompressed file reports once. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: omnigent <noreply@omnigent.ai>
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR adds six data stream features to the FFI for the Swift and Android SDKs. Each change has a test that fails without it.
Changes
info()after the last read.next_progress(), andsend_text,send_bytes, andsend_filetake an optional progress callback.StreamTextOptions.total_length.totalSize). The receiver can show progress and detect a short stream.close(), which stops the native read and ends only that stream. Dropping a reader does the same.Compatibility
cargo semver-checkspasses forlivekit-data-stream,livekit,livekit-rpc, andlivekit-capture.-D warnings.Testing
cargo testpasses: 122livekit-data-streamtests, 24livekit-fficore-module tests, 26livekit-rpctests, and 71livekittests.data_stream_testandrpc_testsuites pass against a local server.After the next LiveKitFFI release, a follow-up to livekit/client-sdk-swift#1075 (merged) adopts these features in Swift.