Data streams v2

[1egoman]

Initial port of data streams v2 to the android sdk. The corresponding swift change (which this is fairly heavily patterned off of ) can be found here: livekit/client-sdk-swift#1075

Previously, the android sdk had its own data streams implementation. With data streams v2, the rust implementation will both be quite a bit more stable and gain some new features that make it significantly more performant (single packet data streams and DEFLATE compression when it makes the payload smaller). This roughly doubles data stream throughput in local testing.

So, port the android sdk to use the rust sdk data streams v2 implementation, and completely remove the pre-existing kotlin implementation. This is a substantial change which needs thorough testing.

New behaviors worth being aware of

All of these are either data streams v2 related changes, or bug fixes.

1. compress option

When sending a data stream, there is a new compress option. Just like how this works on web / rust, this defaults to true. If set to false, then compression will be disabled (useful if you know the data you are sending isn't compressible / you are doing your own compression, which is not uncommon in robotics use cases). The vast majority of users should leave this set to true.

2. Max data stream size

In a rust data streams v2 pull request review comment, we decided that for security reasons it made sense to introduce a maximum data stream size as a DOS protection. This limit is by default 5gb - any data stream that is larger will now read up until that point, and if the stream keeps going, a "payload too large" error will be raised on the stream and exposed to a user on the subsequent .read() call.

If a user is sending a large file, they can override this by setting a new maxPayloadByteLength option on the room.connect call:

room.connect(
    url = wsUrl,
    token = token,
    options = ConnectOptions(
        autoSubscribe = true
    ),
    roomOptions = RoomOptions(
        dynacast = false,
        dataStream = DataStreamOptions(maxPayloadByteLength = 1000)
    )
)

3. Throwing new data stream errors types

The old kotlin specific data streams implementation wasn't quite as strict and didn't surface as many error cases to the caller which were encountered while reading the stream as the rust implementation now does. This needs some testing in some of these edge cases to make sure I didn't inadvertently make this backwards incompatible in a non-aceptable way.

Adding demo to sample-app

To exercise these changes, I've added a new section to the sample-app example which renders a data stream testing interface. This has been heavily patterned off of the similar interface I added to rust-dev-client here:

Uniffi integration

In addition to the data streams v2 features, I've started integrating the uniffi kotlin bindings into this sdk based off of DL's in progress branch. For the most part, this has gone fairly smoothly. I've kept everything wired up with maven local for the time being and will leave it as a cleanup item for an android expert (likely DL) to this working properly prior to merging.

As part of this, I have also fixed two issues in the livekit-uniffi kotlin bindgen:

1. Any uniffi objects which have a method with the name close does not build on uniffi-rs 0.31. I made a ticket here: Kotlin bindings cannot have a method named close, if it does the generated code isn't valid mozilla/uniffi-rs#2955. I have worked around this by renaming close -> close_stream as a kotlin specific override here: livekit/rust-sdks@1fe0eba

2. Any uniffi enums which have tagged cases that contain fields names message fail to build:

UniFFI emits, for a structured error variant carrying a `message` field:

        class AbnormalEnd(val `message`: kotlin.String) : DataStreamException() {
            override val message get() = "message=${ `message` }"
        }

    The constructor property and the overridden `Throwable.message` are both called
    `message` in one class body, which does not compile (and their types differ --
    String vs String? -- so they cannot be merged into one override either).

    The upstream fix would be for uniffi to avoid emitting a property that collides with
    Throwable.message (or for livekit-uniffi to not name the field `message`).

Unfortunately, this one can't be fixed in the same way, so I've opted to rename all error cases that contain message to instead be reason, also in here: livekit/rust-sdks@1fe0eba. The uniffi-rs maintainers seem to be unable to figure out a good way to fix this - the latest related issue was closed: mozilla/uniffi-rs#2938

---

Warning

This pull request was LLM generated and has only been reviewed by a human who isn't a domain expert in android development. I have tested this and confirms it works in the happy path, but no other validation has been done.

A more thorough review of this needs to occur before it could be merged.

Todo

• Get Data streams v2 uniffi rust-sdks#1286 merged, the kotlin bindings published, and this ported over to use the kotlin bindings from maven central (instead of maven local)
• Add changeset

[changeset-bot]

⚠️ No Changeset found

Latest commit: ec3ade1

Merging this PR will not cause a version bump for any packages. If these changes should not result in a new version, you're good to go. If these changes should result in a version bump, you need to add a changeset.

This PR includes no changesets

When changesets are added to this PR, you'll see the packages that this PR includes changesets for and the associated semver types

Click here to learn what changesets are, and how to add one.

Click here if you're a maintainer who wants to add a changeset to this PR

[devin-ai-integration]

Devin Review found 5 potential issues.

View 5 additional findings in Devin Review.

[devin-ai-integration]

🟡 Large outgoing data streams can be buffered entirely in memory

Outgoing stream packets are placed on a queue with no size limit (Channel<ByteArray>(Channel.UNLIMITED) at livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt:144) while the sender is told the write succeeded immediately, so a large send can pile up in memory faster than the network drains it.

Impact: Sending a large file or payload can grow memory without bound and risk an out-of-memory crash.

Why the previous backpressure no longer reaches the producer

Previously ManagerStreamDestination.write chunked the payload itself and, for every chunk, awaited engine.waitForBufferStatusLow(...) followed by engine.sendData(...) before returning — so a slow data channel suspended the caller.

Now WriterDestination.write (livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt:504-512) returns as soon as the core accepts the write, and the core pushes encoded packets synchronously through OutgoingDelegate.onPacketsAvailable (.../DataStreams.kt:344-353) using trySend onto the unlimited channel. Only the single draining coroutine at .../DataStreams.kt:286-301 waits on waitForBufferStatusLow; nothing throttles the producer.

This is most visible for sendFile, where the core reads the whole file rather than the caller streaming it chunk by chunk. The comment at .../DataStreams.kt:137-143 claims this "restor[es] the backpressure the previous implementation had", which an unbounded channel does not do — it only preserves ordering.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[1egoman]

I left a comment covering this in the swift version: livekit/client-sdk-swift#975 (comment). IMO it's worth reading. Lukas + I (mostly lukas) ended up addressing this same issue on the web via a new "low water mark / high water mark" approach on the data channel which I think if android were to also adopt, it could fix this: livekit/client-sdk-js#2014

IMO fixing this should be out of scope and be a follow up task though, as this pull request is already quite large and this would easily add many hundreds more lines.

[github-actions]

Dependency diff:

++--- io.livekit:livekit-uniffi-android:0.0.1 FAILED
+\--- net.java.dev.jna:jna:5.19.1

[1egoman]

Admittedly, I'm not 100% sure this is correct - I would think the final statement in this function should be return false since packets which are not end to end encrypted should be dropped?. According to a LLM it matches what a few other sdks do currently but I think this needs a second pair of eyes to confirm more familiar with how e2ee works.

[devin-ai-integration]

Devin Review found 3 new potential issues.

View 7 additional findings in Devin Review.

[devin-ai-integration]

🟡 Sending data now reports success even when the data never leaves the device

Outgoing data stream packets are dropped with only a log line when the transport rejects them (LKLog.w at livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt:405) after the send call has already told the caller it succeeded, so an app is told a message was delivered when nothing was sent.

Impact: Apps can silently lose messages and files with no way to detect it or retry.

How the failure is swallowed on the new asynchronous send path

Previously OutgoingDataStreamManagerImpl called engine.sendData(...) inline on the caller's coroutine, so a failure (publisher not connected, oversized packet, closed data channel — see RTCEngine.sendData at livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt:760-805) propagated out of sendText/sendBytes/write as a Result.failure.

Now the core hands encoded packets to OutgoingDelegate.onPacketsAvailable (.../DataStreams.kt:466), which queues them on outboundPackets. A single drain coroutine started in init (.../DataStreams.kt:170-173) calls sendPacket, which is the only place the Result from engine.sendData is observed — and it just logs. By then outgoing.sendText(...)/writer.write(...) has already returned, so OutgoingDataStreamManagerImpl.runCatchingStream wraps it in Result.success.

The @CheckResult Result<TextStreamInfo>/Result<ByteStreamInfo> contract on OutgoingDataStreamManager (and BaseStreamSender.write's Result<Unit>) therefore no longer reflects whether the data reached the wire.

A fix would need the transport failure to be reported back into the core (so it can fail the stream, surfacing as DataStreamException.SendFailed), or at minimum be surfaced to the app through a room/stream-level error rather than only a log.

Prompt for agents

In livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt, sendPacket() is the only observer of the Result returned by RTCEngine.sendData, and it merely logs failures. Because the outbound path is now asynchronous (core -> OutgoingDelegate.onPacketsAvailable -> outboundPackets channel -> drain coroutine), the originating sendText/sendBytes/sendFile/write call has already returned Result.success by the time the transport rejects the packet. This is a regression from the previous implementation, where engine.sendData ran inline on the caller's coroutine and its failure became the caller's Result.failure.

Investigate whether livekit-uniffi exposes a way to report a transport send failure back into the core for the affected stream (which would let the core fail the stream and surface DataStreamException.SendFailed / StreamException.TerminatedException(SEND_FAILED) to readers/writers). If no such hook exists, consider at minimum surfacing the failure via a room-level event or by failing subsequent writes on the affected stream, so applications are not told a message was delivered when it was not.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[1egoman]

I will leave this up to @davidliu, this code isn't called as part of the call stack of sendText / etc - it's forwarded along from the outboundPackets channel:

        coroutineScope.launch {
            for (packet in outboundPackets) {
                sendPacket(packet)
            }
        }

This is IMO a fairly nice decoupling but not having access to the error is sort of the natural consequence of doing this. Maybe it's worth passing some sort of oneshot channel / callback type interface where the caller can wait for sendPacket to signal back to the caller a success / error state? I'm unsure and would prefer to defer to DL on this.