Data streams v2

gradle/libs.versions.toml

@@ -32,6 +32,15 @@ noise = "2.0.0"
 lifecycleProcess = "2.8.7"
 agp = "8.7.2"
 kotlin = "1.9.25"
+# Data streams v2 core. Unreleased: build it from rust-sdks with
+# `cargo make android-package-local` and it resolves from mavenLocal(). Switch to a published
+# version once livekit-uniffi ships this API.
+livekit-uniffi = "0.0.1"
+# livekit-uniffi's AAR pins jna 5.16.0, whose libjnidispatch.so fails to dlopen on 16 KB
+# page-size devices ("program alignment (8192) cannot be smaller than system page size").
+# 5.19.1 loads correctly there, so the SDK overrides the transitive pin. Also used for the
+# desktop JNA dispatch library on the unit-test classpath; see gradle/uniffi-native-lib.gradle.
+jna = "5.19.1"
 
 [libraries]
 android-jain-sip-ri = { module = "javax.sip:android-jain-sip-ri", version.ref = "androidJainSipRi" }
@@ -59,6 +68,8 @@ androidx-lifecycle-process = { module = "androidx.lifecycle:lifecycle-process",
 androidx-lifecycle-viewmodel-ktx = { module = "androidx.lifecycle:lifecycle-viewmodel-ktx", version.ref = "androidx-lifecycle" }
 kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json", version.ref = "kotlinx-serialization" }
 leakcanary-android = { module = "com.squareup.leakcanary:leakcanary-android", version.ref = "leakcanaryAndroid" }
+livekit-uniffi = { module = "io.livekit:livekit-uniffi-android", version.ref = "livekit-uniffi" }
+jna-desktop = { module = "net.java.dev.jna:jna", version.ref = "jna" }
 okhttp-lib = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp" }
 okhttp-coroutines = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp" }
 protobuf-javalite = { module = "com.google.protobuf:protobuf-javalite", version.ref = "protobufJavalite" }

gradle/uniffi-native-lib.gradle

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2026 LiveKit, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Points JNA at a *host* build of liblivekit_uniffi for local unit tests.
+//
+// Data streams are implemented in the Rust core, so any test that exercises them has to load
+// the native library. Unit tests here run on the host JVM (Robolectric), but the
+// livekit-uniffi AAR only ships Android .so files -- so the host build has to be supplied
+// separately. This mirrors how client-sdk-swift's tests link the real xcframework rather than
+// faking the core.
+//
+// Resolution order:
+//   1. LIVEKIT_UNIFFI_LIB_DIR environment variable
+//   2. livekitUniffiLibDir Gradle property
+//   3. auto-discovery in a sibling rust-sdks checkout, which is the layout the LiveKit SDKs
+//      already assume for local uniffi development
+//
+// To produce the host library:
+//   cd ../rust-sdks/livekit-uniffi
+//   CARGO_PROFILE_RELEASE_STRIP=false cargo build --release --target <host triple>
+
+ext.resolveUniffiNativeLibDirs = {
+    def libNames = ["liblivekit_uniffi.dylib", "liblivekit_uniffi.so"]
+    def hasLib = { File dir -> dir.isDirectory() && libNames.any { new File(dir, it).isFile() } }
+
+    def explicit = System.getenv("LIVEKIT_UNIFFI_LIB_DIR") ?: project.findProperty("livekitUniffiLibDir")
+    if (explicit) {
+        def dir = file(explicit)
+        if (!hasLib(dir)) {
+            throw new GradleException(
+                "LIVEKIT_UNIFFI_LIB_DIR/livekitUniffiLibDir is set to '${dir}' but no " +
+                    "liblivekit_uniffi.{dylib,so} was found there.",
+            )
+        }
+        return [dir]
+    }
+
+    // Sibling rust-sdks checkout. `target/<triple>/release` is where a --target build lands
+    // (the uniffi Makefile always passes --target); `target/release` covers a plain build.
+    def targetRoot = new File(rootDir, "../rust-sdks/target")
+    if (!targetRoot.isDirectory()) {
+        return []
+    }
+    def candidates = []
+    def plain = new File(targetRoot, "release")
+    if (hasLib(plain)) {
+        candidates << plain
+    }
+    targetRoot.listFiles({ File f -> f.isDirectory() } as FileFilter)?.sort { it.name }?.each { triple ->
+        def dir = new File(triple, "release")
+        if (hasLib(dir)) {
+            candidates << dir
+        }
+    }
+    return candidates
+}
+
+ext.configureUniffiForUnitTests = { Test task ->
+    // The generated bindings pick their cleaner by API level, and at 34+ use
+    // android.system.SystemCleaner. Robolectric's implementation of that delegates to
+    // jdk.internal.ref.CleanerFactory, which the module system hides by default -- so merely
+    // constructing an FFI object throws IllegalAccessError. Opening the package up lets the
+    // Android code path run as written, rather than forcing every test that touches the FFI down
+    // to an older Robolectric SDK.
+    task.jvmArgs += ["--add-exports", "java.base/jdk.internal.ref=ALL-UNNAMED"]
+
+    def dirs = resolveUniffiNativeLibDirs()
+    if (dirs.isEmpty()) {
+        task.doFirst {
+            logger.warn(
+                "[livekit] No host build of liblivekit_uniffi found. Tests that exercise data " +
+                    "streams will fail to load the native library.\n" +
+                    "  Build it:  cd ../rust-sdks/livekit-uniffi && " +
+                    "CARGO_PROFILE_RELEASE_STRIP=false cargo build --release --target \$(rustc -vV | " +
+                    "sed -n 's/^host: //p')\n" +
+                    "  Or point at one:  -PlivekitUniffiLibDir=/path/to/dir",
+            )
+        }
+        return
+    }
+    task.systemProperty("jna.library.path", dirs.collect { it.absolutePath }.join(File.pathSeparator))
+}

livekit-android-sdk/build.gradle

@@ -118,6 +118,11 @@ dependencies {
     implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:$kotlin_version"
     implementation libs.coroutines.lib
     implementation libs.kotlinx.serialization.json
+    implementation libs.livekit.uniffi
+    // livekit-uniffi depends on jna 5.16.0, whose native dispatch library cannot be loaded on
+    // 16 KB page-size devices -- which Android now requires apps targeting API 35+ to support.
+    // Overriding to a version that loads there; remove once the AAR itself moves up.
+    implementation "net.java.dev.jna:jna:${libs.versions.jna.get()}@aar"
     api libs.webrtc
     api libs.okhttp.lib
     implementation libs.okhttp.coroutines

livekit-android-sdk/src/main/AndroidManifest.xml

@@ -14,7 +14,16 @@
  limitations under the License.
 -->
 
-<manifest xmlns:android="http://schemas.android.com/apk/res/android">
+<manifest xmlns:android="http://schemas.android.com/apk/res/android"
+    xmlns:tools="http://schemas.android.com/tools">
+
+    <!--
+      livekit-uniffi's Gradle module declares minSdk 24 while this SDK supports 21. The native
+      library itself is built against platform 21 (cargo-ndk's default), so the declaration is
+      the only thing in the way. Overriding keeps this SDK's minSdk at 21; remove once
+      livekit-uniffi lowers its own declared minSdk to match.
+    -->
+    <uses-sdk tools:overrideLibrary="io.livekit.uniffi" />
 
     <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
     <uses-permission android:name="android.permission.INTERNET" />

livekit-android-sdk/src/main/java/io/livekit/android/ConnectOptions.kt

@@ -60,7 +60,7 @@ data class ConnectOptions(
      * for peer-to-peer feature negotiation (RPC v2, etc.). Defaults to the latest
      * version supported by this SDK build.
      */
-    val clientProtocol: ClientProtocolVersion = ClientProtocolVersion.DATA_STREAM_RPC,
+    val clientProtocol: ClientProtocolVersion = ClientProtocolVersion.DATA_STREAM_V2,
 ) {
     internal var reconnect: Boolean = false
     internal var participantSid: String? = null

livekit-android-sdk/src/main/java/io/livekit/android/RoomOptions.kt

@@ -18,6 +18,7 @@ package io.livekit.android
 
 import io.livekit.android.e2ee.E2EEOptions
 import io.livekit.android.room.Room
+import io.livekit.android.room.datastream.DataStreamOptions
 import io.livekit.android.room.network.ReconnectPolicy
 import io.livekit.android.room.participant.AudioTrackPublishDefaults
 import io.livekit.android.room.participant.VideoTrackPublishDefaults
@@ -51,4 +52,9 @@ data class RoomOptions(
      * @see [Room.reconnectPolicy]
      */
     val reconnectPolicy: ReconnectPolicy? = null,
+
+    /**
+     * Room-wide data stream settings.
+     */
+    val dataStreamOptions: DataStreamOptions = DataStreamOptions(),
 )

livekit-android-sdk/src/main/java/io/livekit/android/room/ClientCapability.kt

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2026 LiveKit, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.livekit.android.room
+
+import livekit.LivekitModels
+
+/**
+ * An optional feature capability a client advertises at connect time.
+ *
+ * Capabilities are independent feature flags, in contrast to [ClientProtocolVersion], which is a
+ * single monotonic version number. A client may speak a given client protocol yet still lack an
+ * individual capability, so the two are negotiated separately.
+ *
+ * Advertised by this SDK during the join handshake, mirrored by the server onto every
+ * `ParticipantInfo`, and readable per-peer via [io.livekit.android.room.participant.Participant.capabilities].
+ */
+enum class ClientCapability(val value: Int) {
+    /**
+     * The client can accept RTP packet trailers passed through by the SFU instead of having
+     * them stripped.
+     */
+    PACKET_TRAILER(1),
+
+    /**
+     * The client can decompress a `deflate-raw` compressed data stream payload.
+     */
+    COMPRESSION_DEFLATE_RAW(2),
+    ;
+
+    fun toProto(): LivekitModels.ClientInfo.Capability {
+        return when (this) {
+            PACKET_TRAILER -> LivekitModels.ClientInfo.Capability.CAP_PACKET_TRAILER
+            COMPRESSION_DEFLATE_RAW -> LivekitModels.ClientInfo.Capability.CAP_COMPRESSION_DEFLATE_RAW
+        }
+    }
+
+    companion object {
+        /**
+         * Converts from the protobuf enum, returning null for values this SDK build does not
+         * recognize.
+         *
+         * Unlike most `fromProto` helpers in this SDK this never throws: capabilities are an
+         * open, forward-extensible set, so a peer or server advertising a newer capability must
+         * be ignored rather than crash us.
+         */
+        fun fromProto(capability: LivekitModels.ClientInfo.Capability): ClientCapability? {
+            return when (capability) {
+                LivekitModels.ClientInfo.Capability.CAP_PACKET_TRAILER -> PACKET_TRAILER
+                LivekitModels.ClientInfo.Capability.CAP_COMPRESSION_DEFLATE_RAW -> COMPRESSION_DEFLATE_RAW
+                LivekitModels.ClientInfo.Capability.CAP_UNUSED,
+                LivekitModels.ClientInfo.Capability.UNRECOGNIZED,
+                -> null
+            }
+        }
+    }
+}
+
+/**
+ * The capabilities this SDK advertises to the server and, through it, to peers.
+ *
+ * Declared once so that every place which announces capabilities agrees. Data streams v2
+ * compression is advertised unconditionally: deflate-raw is compressed and decompressed by the
+ * Rust data stream core, so support does not vary by device, API level, or room options.
+ *
+ * @suppress
+ */
+internal val ADVERTISED_CLIENT_CAPABILITIES = listOf(
+    ClientCapability.COMPRESSION_DEFLATE_RAW,
+)

livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt

@@ -47,6 +47,8 @@ import io.livekit.android.events.RoomEvent
 import io.livekit.android.events.collect
 import io.livekit.android.memory.CloseableManager
 import io.livekit.android.renderer.TextureViewRenderer
+import io.livekit.android.room.datastream.DataStreamOptions
+import io.livekit.android.room.datastream.DataStreams
 import io.livekit.android.room.datastream.incoming.IncomingDataStreamManager
 import io.livekit.android.room.metrics.collectMetrics
 import io.livekit.android.room.network.NetworkCallbackManagerFactory
@@ -150,6 +152,7 @@ constructor(
     private val connectionWarmer: ConnectionWarmer,
     private val audioRecordPrewarmer: AudioRecordPrewarmer,
     private val incomingDataStreamManager: IncomingDataStreamManager,
+    private val dataStreams: DataStreams,
     private val rpcClientManager: RpcClientManager,
     private val rpcServerManager: RpcServerManager,
     private val remoteParticipantFactory: RemoteParticipant.Factory,
@@ -182,6 +185,13 @@ constructor(
         }
         rpcClientManager.getRemoteClientProtocol = getRemoteClientProtocol
         rpcServerManager.getRemoteClientProtocol = getRemoteClientProtocol
+
+        // The data stream core decides per-recipient whether a v2 framing is safe, so give it a
+        // live view of the room. Lambdas rather than a Room reference, which would be a DI cycle.
+        dataStreams.remoteClientProtocol = getRemoteClientProtocol
+        dataStreams.remoteIdentities = { remoteParticipants.keys.toList() }
+        dataStreams.remoteCapabilities = { id -> remoteParticipants[id]?.capabilities ?: emptyList() }
+        dataStreams.maxPayloadSize = { dataStreamOptions.maxPayloadSize }
     }
 
     enum class State {
@@ -316,6 +326,11 @@ constructor(
      */
     var e2eeOptions: E2EEOptions? = null
 
+    /**
+     * Room-wide data stream settings. Must be set up prior to [connect].
+     */
+    var dataStreamOptions: DataStreamOptions = DataStreamOptions()
+
     /**
      * Default options to use when creating an audio track.
      */
@@ -406,6 +421,7 @@ constructor(
             videoTrackPublishDefaults = videoTrackPublishDefaults,
             screenShareTrackCaptureDefaults = screenShareTrackCaptureDefaults,
             screenShareTrackPublishDefaults = screenShareTrackPublishDefaults,
+            dataStreamOptions = dataStreamOptions,
         )
 
     /**
@@ -647,6 +663,7 @@ constructor(
         adaptiveStream = options.adaptiveStream
         dynacast = options.dynacast
         e2eeOptions = options.e2eeOptions
+        dataStreamOptions = options.dataStreamOptions
     }
 
     /**
@@ -794,6 +811,10 @@ constructor(
         eventBus.postEvent(RoomEvent.ParticipantDisconnected(this, removedParticipant), coroutineScope)
 
         localParticipant.handleParticipantDisconnect(identity)
+
+        // Fail any stream they were midway through sending, so its reader raises instead of waiting
+        // for chunks that will never arrive.
+        dataStreams.abortStreamsFrom(identity)
     }
 
     fun getParticipantBySid(sid: String): Participant? {
@@ -1341,22 +1362,14 @@ constructor(
      * @suppress
      */
     override fun onDataStreamPacket(dp: LivekitModels.DataPacket, encryptionType: LivekitModels.Encryption.Type) {
-        when (dp.valueCase) {
-            LivekitModels.DataPacket.ValueCase.STREAM_HEADER -> {
-                incomingDataStreamManager.handleStreamHeader(dp.streamHeader, Participant.Identity(dp.participantIdentity), encryptionType)
-            }
-
-            LivekitModels.DataPacket.ValueCase.STREAM_CHUNK -> {
-                incomingDataStreamManager.handleDataChunk(dp.streamChunk, encryptionType)
-            }
-
-            LivekitModels.DataPacket.ValueCase.STREAM_TRAILER -> {
-                incomingDataStreamManager.handleStreamTrailer(dp.streamTrailer, encryptionType)
-            }
-
-            // Ignore other cases.
-            else -> {}
-        }
+        // Forwarded whole rather than destructured: the core re-decodes the packet itself, and v2
+        // headers carry fields (inline content, compression) that only it interprets. The packet has
+        // already been decrypted by the engine, which is what the core expects.
+        //
+        // The encryption goes with it: how the packet arrived is invisible to the core, which sees
+        // only the decrypted bytes, so DataStreams holds every stream to the encryption its header
+        // came in under.
+        dataStreams.handleIncoming(dp, encryptionType)
     }
 
     /**