Add support for RPC V2 (#946)

* feat: add rpc tester to sample-app

* feat: advertise client protocol to other participants

* feat: add initial rpc v2 implementation

* feat: swap client protocol logic to work like regular sfu protocol

* feat: add missing tests from rpc spec

* fix: add missing changeset

* fix: bump protocol version

* feat: add bits to sdk which require bumped protocol version

* feat: try to fix build in ci

* fix: run spotless apply

* fix: add unused protocol fields to whitelist

* fix: run spotless apply

* fix: commit detekt file

* cleanup and merge fixes

* Set clientProtocol on local participant when connecting

* reset client protocol version for local participant when resetting

* cleanup

* update baseline

* clean up protos

* switch protobuf proguard to official recommended rule

* changesets

* switch to allowlist and cut down on the generated protos

---------

Co-authored-by: davidliu <dl@livekit.io>
Co-authored-by: davidliu <davidliu@deviange.net>

Author: 1egoman

.changeset/heavy-parrots-join.md

@@ -0,0 +1,5 @@
+---
+"client-sdk-android": patch
+---
+
+Change proguard rule for protobufs to official recommended rule, allowing unused protobuf classes to be removed with minification

.changeset/wacky-turtles-thank.md

@@ -0,0 +1,5 @@
+---
+"client-sdk-android": minor
+---
+
+Add support for RPC V2

gradle/livekit-protobuf.gradle

@@ -0,0 +1,72 @@
+// Shared protobuf setup for LiveKit client protos.
+//
+// Configure before applying:
+//   ext.livekitProtoIncludes = ['livekit_models.proto', ...]
+// Optional (for modules that compile a subset but import others):
+//   ext.livekitProtoImportSrc = true
+
+if (!project.ext.has('livekitProtoIncludes')) {
+    throw new GradleException("ext.livekitProtoIncludes must be set before applying livekit-protobuf.gradle")
+}
+
+def stagedProtoSrcDir = layout.buildDirectory.dir("staged-proto-src")
+def stageProtoSources = tasks.register("stageProtoSources", Copy) {
+    from(generated.protoSrc) {
+        include project.ext.livekitProtoIncludes as String[]
+    }
+    into stagedProtoSrcDir
+}
+
+android.sourceSets.main.proto {
+    srcDir stagedProtoSrcDir
+}
+
+android.sourceSets.main.java {
+    srcDir "${protobuf.generatedFilesBaseDir}/main/javalite"
+}
+
+configurations {
+    descriptorProtoSource
+}
+
+dependencies {
+    descriptorProtoSource "com.google.protobuf:protobuf-java:${libs.versions.protobuf.get()}"
+}
+
+def extractedImportProtosDir = layout.buildDirectory.dir("extracted-import-protos")
+def extractProtoImports = tasks.register("extractProtoImports", Copy) {
+    into extractedImportProtosDir
+    from { zipTree(configurations.descriptorProtoSource.singleFile) } {
+        include "google/protobuf/descriptor.proto"
+    }
+    from(generated.protoSrc) {
+        include "logger/**"
+    }
+}
+
+protobuf {
+    protoc {
+        // for apple m1, please add protoc_platform=osx-x86_64 in $HOME/.gradle/gradle.properties
+        if (project.hasProperty('protoc_platform')) {
+            artifact = "com.google.protobuf:protoc:${libs.versions.protobuf.get()}:${protoc_platform}"
+        } else {
+            artifact = "com.google.protobuf:protoc:${libs.versions.protobuf.get()}"
+        }
+    }
+
+    generateProtoTasks {
+        all().each { task ->
+            task.dependsOn stageProtoSources
+            task.dependsOn extractProtoImports
+            task.addIncludeDir files(extractedImportProtosDir)
+            if (project.ext.has('livekitProtoImportSrc') && project.ext.livekitProtoImportSrc) {
+                task.addIncludeDir files(generated.protoSrc)
+            }
+            task.builtins {
+                java {
+                    option "lite"
+                }
+            }
+        }
+    }
+}

livekit-android-sdk/build.gradle

@@ -30,18 +30,6 @@ android {
         }
     }
 
-    sourceSets {
-        main {
-            proto {
-                srcDir generated.protoSrc
-                exclude '*/*.proto' // only use top-level protos.
-            }
-            java {
-                srcDir "${protobuf.generatedFilesBaseDir}/main/javalite"
-            }
-        }
-    }
-
     testOptions {
         unitTests {
             includeAndroidResources = true
@@ -74,26 +62,12 @@ android {
 
 }
 
-protobuf {
-    protoc {
-        // for apple m1, please add protoc_platform=osx-x86_64 in $HOME/.gradle/gradle.properties
-        if (project.hasProperty('protoc_platform')) {
-            artifact = "com.google.protobuf:protoc:${libs.versions.protobuf.get()}:${protoc_platform}"
-        } else {
-            artifact = "com.google.protobuf:protoc:${libs.versions.protobuf.get()}"
-        }
-    }
-
-    generateProtoTasks {
-        all().each { task ->
-            task.builtins {
-                java {
-                    option "lite"
-                }
-            }
-        }
-    }
-}
+ext.livekitProtoIncludes = [
+        'livekit_models.proto',
+        'livekit_rtc.proto',
+        'livekit_metrics.proto',
+]
+apply from: rootProject.file('gradle/livekit-protobuf.gradle')
 
 jacoco {
     toolVersion = "0.8.14"

livekit-android-sdk/consumer-rules.pro

@@ -42,4 +42,6 @@
 
 # Protobuf
 #########################################
--keep class * extends com.google.protobuf.GeneratedMessageLite { *; }
+-keepclassmembers class * extends com.google.protobuf.GeneratedMessageLite {
+  <fields>;
+}

livekit-android-sdk/detekt-baseline-release.xml

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2024 LiveKit, Inc.
+ * Copyright 2023-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.
@@ -16,6 +16,7 @@
 
 package io.livekit.android
 
+import io.livekit.android.room.ClientProtocolVersion
 import io.livekit.android.room.ProtocolVersion
 import io.livekit.android.room.Room
 import livekit.org.webrtc.PeerConnection
@@ -53,6 +54,13 @@ data class ConnectOptions(
      * the protocol version to use with the server.
      */
     val protocolVersion: ProtocolVersion = ProtocolVersion.v13,
+
+    /**
+     * The client protocol version to advertise to other participants in the room
+     * 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,
 ) {
     internal var reconnect: Boolean = false
     internal var participantSid: String? = null

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2025 LiveKit, Inc.
+ * Copyright 2023-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.
@@ -328,6 +328,7 @@ fun LivekitModels.DisconnectReason?.convert(): DisconnectReason {
         LivekitModels.DisconnectReason.SIP_TRUNK_FAILURE -> DisconnectReason.SIP_TRUNK_FAILURE
         LivekitModels.DisconnectReason.CONNECTION_TIMEOUT -> DisconnectReason.CONNECTION_TIMEOUT
         LivekitModels.DisconnectReason.MEDIA_FAILURE -> DisconnectReason.MEDIA_FAILURE
+        LivekitModels.DisconnectReason.AGENT_ERROR,
         LivekitModels.DisconnectReason.UNKNOWN_REASON,
         LivekitModels.DisconnectReason.UNRECOGNIZED,
         null,

livekit-android-sdk/src/main/java/io/livekit/android/events/RoomEvent.kt

@@ -61,7 +61,11 @@ import io.livekit.android.room.participant.RpcHandler
 import io.livekit.android.room.participant.VideoTrackPublishDefaults
 import io.livekit.android.room.participant.publishTracksInfo
 import io.livekit.android.room.provisions.LKObjects
+import io.livekit.android.room.rpc.RPC_REQUEST_DATA_STREAM_TOPIC
+import io.livekit.android.room.rpc.RPC_RESPONSE_DATA_STREAM_TOPIC
+import io.livekit.android.room.rpc.RpcClientManager
 import io.livekit.android.room.rpc.RpcManager
+import io.livekit.android.room.rpc.RpcServerManager
 import io.livekit.android.room.track.LocalAudioTrackOptions
 import io.livekit.android.room.track.LocalTrackPublication
 import io.livekit.android.room.track.LocalVideoTrackOptions
@@ -146,6 +150,8 @@ constructor(
     private val connectionWarmer: ConnectionWarmer,
     private val audioRecordPrewarmer: AudioRecordPrewarmer,
     private val incomingDataStreamManager: IncomingDataStreamManager,
+    private val rpcClientManager: RpcClientManager,
+    private val rpcServerManager: RpcServerManager,
     private val remoteParticipantFactory: RemoteParticipant.Factory,
 ) : RTCEngine.Listener, ParticipantListener, RpcManager, IncomingDataStreamManager by incomingDataStreamManager {
 
@@ -155,6 +161,27 @@ constructor(
 
     init {
         engine.listener = this
+
+        // Register SDK-internal text-stream handlers for the RPC v2 transport. These reserve
+        // the topics `lk.rpc_request` and `lk.rpc_response` from user-level handler registration.
+        incomingDataStreamManager.registerTextStreamHandler(RPC_REQUEST_DATA_STREAM_TOPIC) { receiver, fromIdentity ->
+            coroutineScope.launch {
+                rpcServerManager.handleIncomingDataStream(receiver, fromIdentity)
+            }
+        }
+        incomingDataStreamManager.registerTextStreamHandler(RPC_RESPONSE_DATA_STREAM_TOPIC) { receiver, fromIdentity ->
+            coroutineScope.launch {
+                rpcClientManager.handleIncomingDataStreamResponse(receiver, fromIdentity)
+            }
+        }
+
+        // Wire each manager's clientProtocol lookup via the remote-participants store.
+        val getRemoteClientProtocol: (Participant.Identity) -> Int = { id ->
+            remoteParticipants[id]?.clientProtocol
+                ?: ClientProtocolVersion.DEFAULT.value
+        }
+        rpcClientManager.getRemoteClientProtocol = getRemoteClientProtocol
+        rpcServerManager.getRemoteClientProtocol = getRemoteClientProtocol
     }
 
     enum class State {
@@ -454,7 +481,7 @@ constructor(
             roomOptions = getCurrentRoomOptions()
 
             // Setup local participant.
-            localParticipant.reinitialize()
+            localParticipant.reinitialize(options)
             setupLocalParticipantEventHandling()
 
             if (roomOptions.e2eeOptions != null) {

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

@@ -173,7 +173,7 @@ constructor(
         // Clean up any pre-existing connection.
         close(reason = "Starting new connection", shouldClearQueuedRequests = false)
 
-        val wsUrlString = "${url.toWebsocketUrl()}/rtc${createConnectionParams(getClientInfo(), options, roomOptions)}"
+        val wsUrlString = "${url.toWebsocketUrl()}/rtc${createConnectionParams(getClientInfo(options.clientProtocol), options, roomOptions)}"
         isReconnecting = options.reconnect
 
         LKLog.i { "connecting to $wsUrlString" }
@@ -240,6 +240,7 @@ constructor(
         addParam(CONNECT_QUERY_OS, clientInfo.os)
         addParam(CONNECT_QUERY_OS_VERSION, clientInfo.osVersion)
         addParam(CONNECT_QUERY_NETWORK_TYPE, networkInfo.getNetworkType().protoName)
+        addParam(CONNECT_QUERY_CLIENT_PROTOCOL, options.clientProtocol.value.toString())
 
         return queryBuilder.toString()
     }
@@ -856,6 +857,18 @@ constructor(
             LivekitRtc.SignalResponse.MessageCase.SUBSCRIBED_AUDIO_CODEC_UPDATE -> {
                 // TODO
             }
+
+            LivekitRtc.SignalResponse.MessageCase.PUBLISH_DATA_TRACK_RESPONSE -> {
+                // TODO
+            }
+
+            LivekitRtc.SignalResponse.MessageCase.UNPUBLISH_DATA_TRACK_RESPONSE -> {
+                // TODO
+            }
+
+            LivekitRtc.SignalResponse.MessageCase.DATA_TRACK_SUBSCRIBER_HANDLES -> {
+                // TODO
+            }
         }
     }
 
@@ -959,6 +972,7 @@ constructor(
         const val CONNECT_QUERY_OS_VERSION = "os_version"
         const val CONNECT_QUERY_NETWORK_TYPE = "network"
         const val CONNECT_QUERY_PARTICIPANT_SID = "sid"
+        const val CONNECT_QUERY_CLIENT_PROTOCOL = "client_protocol"
 
         const val SD_TYPE_ANSWER = "answer"
         const val SD_TYPE_OFFER = "offer"
@@ -1012,6 +1026,27 @@ enum class ProtocolVersion(val value: Int) {
     v13(13),
 }
 
+/**
+ * The protocol version this SDK advertises to **peers** (other participants) for
+ * client-to-client feature negotiation (RPC v2, etc.). Distinct from [ProtocolVersion],
+ * which tracks the signaling protocol between client and server.
+ *
+ * Sent to the server during the join handshake via the `client_protocol` connection
+ * query parameter and `ClientInfo.client_protocol`; the server then populates
+ * `ParticipantInfo.client_protocol` for other peers in the room to read.
+ */
+@Suppress("unused")
+enum class ClientProtocolVersion(val value: Int) {
+    /** Initial client protocol. RPC v1 only (15 KB packet payload limit). */
+    DEFAULT(0),
+
+    /**
+     * RPC v2: request and success-response payloads are carried over text data streams
+     * instead of inline packets, lifting the 15 KB payload limit.
+     */
+    DATA_STREAM_RPC(1),
+}
+
 class ServerInfo(
     val edition: Edition,
     val version: Semver?,