Skip to content

Adopt the LiveKitFFI data stream additions and drop the Swift workarounds - #1181

Draft
pblazej wants to merge 16 commits into
mainfrom
blaze/dsv2-ffi-followup
Draft

Adopt the LiveKitFFI data stream additions and drop the Swift workarounds#1181
pblazej wants to merge 16 commits into
mainfrom
blaze/dsv2-ffi-followup

Conversation

@pblazej

@pblazej pblazej commented Oct 28, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Draft until LiveKitFFI ships a release that includes livekit/rust-sdks#1558. CI stays red until Package.swift moves the LiveKitFFI pin from 0.13.1 to that release.

Added

  • Readers show trailer attributes in info after the stream ends.
  • Writers send attributes with close(reason:attributes:).
  • sendText, sendBytes, and sendFile take an onProgress callback, and readers expose progress as an AsyncStream.
  • StreamTextOptions.totalSize declares the size of an incremental text stream.
  • StreamProgress has bytesProcessed, bytesTotal, and fraction, with NSNumber accessors for Objective-C.

Removed

  • The Swift check of the declared length at the end of a stream, and the per-reader serial runner. Rust checks the length now.
  • The abandon-on-cancel wrapper around reads. A cancelled read closes the core reader, as in 2.18.0, and a later read throws .alreadyClosed.
  • 11 tests that covered only the removed code. Each one has a Rust test in #1558.

Kept

  • The file-name check, the Int.max conversion, the sender retirement after a reconnect, the performRpc cancellation fix, the zero receive limit, and the error mapping.
  • The abandon-on-cancel wrapper around sends and writes, because UniFFI does not cancel the Rust call.
  • Old signatures, as forwarding overloads. The API check reports additions only.

pblazej and others added 16 commits October 28, 2026 11:10
@pblazej @claude omnigent
Let the core check a stream's declared length
7c00d6c 
The Rust core now fails an inline stream whose payload does not match the
length in its header (Incomplete or LengthExceeded), as it already did for
chunked streams. The readers no longer count bytes, check the length at the
end, or run their reads one at a time to keep that count right after a
cancel.

Removes the tests of that Swift check. The Rust tests now cover it:
v2_inline_payload_is_held_to_its_declared_length,
v2_compressed_errors_when_too_few_bytes / _too_many_bytes and
v1_errors_when_too_few_bytes / _too_many_bytes (livekit-data-stream), and
outgoing_text_stream_declares_total_length (livekit-ffi), whose receiver
fails a short stream with Incomplete.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Close the core reader when a read is cancelled
1ab1d87 
The FFI readers now have close(), which fails a pending read at once and
ends the stream in the incoming manager. A cancelled read calls it and
throws CancellationError, so the stream ends, as a v1 reader's
AsyncThrowingStream did when its reading task was cancelled. A later read
throws StreamError.alreadyClosed.

This replaces two workarounds: the read that a cancel abandoned in its own
task, and readAll() as a loop over next(), which existed only so a cancel
could stop it. readAll() calls the core's readAll() again.

Tests: cancelledReadEndsTheStream replaces cancellingWaitingReadReturns,
cancelledReadAllLeavesRestOfStream (the stream no longer stays readable)
and cancelledReadLeavesNoOpenStream. The Rust side of the close is covered
by incoming_reader_close_ends_its_stream_without_more_packets and
incoming_byte_reader_close_ends_its_stream (livekit-ffi).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Show trailer attributes in an incoming stream's info
3b93eed 
A reader's info now reads the core's info on each access, so once the stream
has ended its attributes also hold the attributes the sender set in the
trailer, with trailer keys replacing header keys, as in JS. The property
changes from a stored let to a computed var, which keeps the API and ABI.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Send attributes with the end of an outgoing stream
eaba4e9 
TextStreamWriter and ByteStreamWriter gain
close(reason:attributes:), which puts the attributes in the stream's
trailer through the FFI writer's close(reason:attributes:). Receivers see
them in the reader's info once the stream has ended. close(reason:) stays
as a plain overload with its signature and Objective-C selector, and
forwards to the new method.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Report the progress of sends and incoming streams
3935709 
StreamProgress holds the payload bytes sent or received so far, the total
when it is known, and their fraction. As in JS, a send counts bytes before
compression and ends with a report of the whole payload, and a reader counts
bytes as they arrive, with no total for a stream that declared no size.

- LocalParticipant: sendText(_:options:onProgress:),
  sendBytes(_:options:onProgress:) and sendFile(_:options:onProgress:)
  pass the closure to the FFI send as a StreamProgressDelegate. A one-shot
  text or byte send reports once, when it completes; a file reports as it
  goes out. The earlier signatures stay as plain overloads that forward.
- TextStreamReader and ByteStreamReader: progress, an AsyncStream over the
  FFI reader's nextProgress(). A cancel ends the iteration at once.
- Objective-C: the send overloads are exposed under their own selectors,
  and readProgress(onProgress:) wraps the sequence the way readChunks wraps
  the chunks.

The progress closure is a trailing parameter rather than an options field:
the options classes are shared with streamText and streamBytes, where
progress does not apply.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Declare the size of an outgoing text stream
da49cc7 
StreamTextOptions gains totalLength, the size of the text in UTF-8 bytes,
named like StreamInfo.totalLength. streamText sends it in the header
through the FFI options' totalLength, so receivers see it in the info and
as the progress total, and the core fails a stream that does not match it.
A one-shot sendText keeps declaring the size of its text. A negative value
reads as nil, as StreamByteOptions.totalSize does.

The initializer without totalLength stays as a convenience overload with
its signature and Objective-C selector. Because Int? does not bridge,
Objective-C gets an initializer that takes totalLengthNumber, deprecated
for Swift callers like the totalSizeNumber initializer of
StreamByteOptions.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Add a changeset for the data stream additions
782b304 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Show stream progress and close attributes in the handler docs
01e94e1 
Adds an example to registerByteStreamHandler that follows a reader's
progress and reads the attributes the stream closed with, and documents
readProgress(onProgress:) for Objective-C callers.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Test that a close reason wins over a short declared length
22e9706 
The core now ends a stream whose trailer carries a reason with
AbnormalEnd(reason), even when the stream is shorter than its declared
length; before, it reported Incomplete. abnormalClosure checks both a
stream without a declared length and one that stops short of it. The core
side is trailer_reason_wins_over_a_short_declared_length
(livekit-data-stream).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Name the declared text size totalSize
db57c01 
Renames StreamTextOptions.totalLength to totalSize, with the initializer
label, the Objective-C totalSizeNumber: initializer, the docs, the tests and
the changeset, so both option types and JS and Flutter use the same name.
StreamInfo.totalLength keeps its name, and the option still maps to the FFI
totalLength.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Close the core reader when a stream reader is released
1899763 
A cancelled progress iteration abandons the core's nextProgress() wait,
because UniFFI does not cancel the Rust future. That pending call holds the
FFI reader, so the core's own release, which ends the stream, never ran:
after a cancelled progress observation and the release of the reader, a
stalled stream stayed open.

The readers now close the FFI reader in deinit. Rust then ends the pending
progress wait with nil, the call returns and releases the FFI reader, and the
manager drops the stream. Cancelling progress still does not end the stream
while the reader is alive, and the progress docs now say so.

releasingReaderAfterCancelledProgressEndsTheStream covers both readers. It
failed before this change with openAfterRelease == 1 for .byte and .text.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Test read cancellation for both stream readers
c159215 
cancelledReadEndsTheStream now runs for the text reader too, for a pending
next() and readAll(), cancelled while waiting in the core or before the read
starts: CancellationError, no stream left open, then .alreadyClosed. It uses
the both-kind stand-in readers of the progress test, so the byte-only
SignallingReader and the unused entered and totalLength parameters of
openReader are gone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Give Objective-C the progress total and fraction
def2e0a 
StreamProgress.bytesTotal (Int?) and fraction (Double?) do not bridge to
Objective-C. bytesTotalNumber and fractionNumber return them as NSNumber?,
like DataStreamOptions.maxPayloadByteLengthNumber. A bridge test pins the
mapping and the accessors, and testStreamAdditionsAreExposed checks both
selectors.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Use a case type in the progress bridge test
5b54f9d 
SwiftLint's large_tuple rule rejects the four-member tuple that 72b67fe7
passed to progressMapsCountsAndObjCAccessors.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Keep a reader alive while its progress is observed
c24c5d9 
The progress sequence held only the FFI reader, so the deinit close from
aa01ec06 ended the stream as soon as the app released its last reader
reference, even while it still iterated progress: the stream closed and the
iteration returned nil.

The sequence's producer now holds the reader. AsyncStream(unfolding:)
drops the producer when the sequence finishes or its iteration is cancelled,
so an observation keeps the reader only for its own lifetime. Each
nextProgress() wait still captures only the FFI reader, so a cancelled
observation lets the reader deinit and wake the orphaned wait, which
releasingReaderAfterCancelledProgressEndsTheStream keeps checking.

observedProgressKeepsTheStreamOpen covers both readers: it consumes one
update, releases the reader, sends the rest of the stream, and expects the
stream to stay open and the iteration to reach the end. Before this change
it failed for .byte and .text with openAfterRelease == 0 (expected 1) and a
last update of 0 bytes (expected 5).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
@pblazej @claude omnigent
Pass the send progress delegate in the FFI options record
31f8480 
rust-sdks 7d0b8d90 (PR #1558) moved the progress delegate of sendText,
sendBytes and sendFile from a trailing parameter into an optional progress
field of the FFI StreamTextOptions and StreamByteOptions records. The sends
now set that field on the record. sendFile builds its record from
options.ffi too, which keeps the file under SwiftLint's length limit. No
code relied on the records being Equatable or Hashable, which they no longer
are, and the public API does not change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: omnigent <noreply@omnigent.ai>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@hiroshihorie hiroshihorie Awaiting requested review from hiroshihorie hiroshihorie will be requested when the pull request is marked ready for review hiroshihorie is a code owner
@xianshijing-lk xianshijing-lk Awaiting requested review from xianshijing-lk xianshijing-lk will be requested when the pull request is marked ready for review xianshijing-lk is a code owner

At least 1 approving review is required to merge this pull request.

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

1 participant

@pblazej