Adaptive stream connection param

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

@@ -3,6 +3,7 @@ package io.livekit.android.room
 import android.os.SystemClock
 import com.google.protobuf.ByteString
 import io.livekit.android.ConnectOptions
+import io.livekit.android.RoomOptions
 import io.livekit.android.dagger.InjectionNames
 import io.livekit.android.room.participant.ParticipantTrackPermission
 import io.livekit.android.room.track.TrackException
@@ -80,6 +81,7 @@ internal constructor(
     private var sessionUrl: String? = null
     private var sessionToken: String? = null
     private var connectOptions: ConnectOptions? = null
+    private var lastRoomOptions: RoomOptions? = null
 
     private val publisherObserver = PublisherTransportObserver(this, client)
     private val subscriberObserver = SubscriberTransportObserver(this, client)
@@ -113,16 +115,28 @@ internal constructor(
         client.listener = this
     }
 
-    suspend fun join(url: String, token: String, options: ConnectOptions): LivekitRtc.JoinResponse {
+    suspend fun join(
+        url: String,
+        token: String,
+        options: ConnectOptions,
+        roomOptions: RoomOptions
+    ): LivekitRtc.JoinResponse {
         coroutineScope.close()
         coroutineScope = CloseableCoroutineScope(SupervisorJob() + ioDispatcher)
         sessionUrl = url
         sessionToken = token
-        return joinImpl(url, token, options)
-    }
-
-    suspend fun joinImpl(url: String, token: String, options: ConnectOptions): LivekitRtc.JoinResponse {
-        val joinResponse = client.join(url, token, options)
+        connectOptions = options
+        lastRoomOptions = roomOptions
+        return joinImpl(url, token, options, roomOptions)
+    }
+
+    suspend fun joinImpl(
+        url: String,
+        token: String,
+        options: ConnectOptions,
+        roomOptions: RoomOptions
+    ): LivekitRtc.JoinResponse {
+        val joinResponse = client.join(url, token, options, roomOptions)
         listener?.onJoinResponse(joinResponse)
         isClosed = false
         listener?.onSignalConnected(false)
@@ -281,6 +295,7 @@ internal constructor(
         sessionUrl = null
         sessionToken = null
         connectOptions = null
+        lastRoomOptions = null
         reconnectingJob?.cancel()
         reconnectingJob = null
         coroutineScope.close()
@@ -343,7 +358,7 @@ internal constructor(
                     try {
                         closeResources()
                         listener?.onFullReconnecting()
-                        joinImpl(url, token, connectOptions ?: ConnectOptions())
+                        joinImpl(url, token, connectOptions ?: ConnectOptions(), lastRoomOptions ?: RoomOptions())
                     } catch (e: Exception) {
                         LKLog.w(e) { "Error during reconnection." }
                         // reconnect failed, retry.

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

@@ -11,6 +11,7 @@ import dagger.assisted.Assisted
 import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
 import io.livekit.android.ConnectOptions
+import io.livekit.android.RoomOptions
 import io.livekit.android.Version
 import io.livekit.android.dagger.InjectionNames
 import io.livekit.android.events.BroadcastEventBus
@@ -148,6 +149,15 @@ constructor(
     private var hasLostConnectivity: Boolean = false
     private var connectOptions: ConnectOptions = ConnectOptions()
 
+    private fun getCurrentRoomOptions(): RoomOptions =
+        RoomOptions(
+            adaptiveStream = adaptiveStream,
+            dynacast = dynacast,
+            audioTrackCaptureDefaults = audioTrackCaptureDefaults,
+            videoTrackCaptureDefaults = videoTrackCaptureDefaults,
+            audioTrackPublishDefaults = audioTrackPublishDefaults,
+            videoTrackPublishDefaults = videoTrackPublishDefaults,
+        )
 
     suspend fun connect(url: String, token: String, options: ConnectOptions = ConnectOptions()) {
         if (this::coroutineScope.isInitialized) {
@@ -156,7 +166,7 @@ constructor(
         coroutineScope = CoroutineScope(defaultDispatcher + SupervisorJob())
         state = State.CONNECTING
         connectOptions = options
-        engine.join(url, token, options)
+        engine.join(url, token, options, getCurrentRoomOptions())
 
         val cm = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
         val networkRequest = NetworkRequest.Builder()

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

@@ -2,6 +2,7 @@ package io.livekit.android.room
 
 import com.vdurmont.semver4j.Semver
 import io.livekit.android.ConnectOptions
+import io.livekit.android.RoomOptions
 import io.livekit.android.dagger.InjectionNames
 import io.livekit.android.room.participant.ParticipantTrackPermission
 import io.livekit.android.room.track.Track
@@ -49,6 +50,8 @@ constructor(
     var listener: Listener? = null
     private var serverVersion: Semver? = null
     private var lastUrl: String? = null
+    private var lastOptions: ConnectOptions? = null
+    private var lastRoomOptions: RoomOptions? = null
 
     private var joinContinuation: CancellableContinuation<Either<LivekitRtc.JoinResponse, Unit>>? = null
     private lateinit var coroutineScope: CloseableCoroutineScope
@@ -59,15 +62,17 @@ constructor(
 
     private val responseFlow = MutableSharedFlow<LivekitRtc.SignalResponse>(Int.MAX_VALUE)
 
+
     /**
      * @throws Exception if fails to connect.
      */
     suspend fun join(
         url: String,
         token: String,
         options: ConnectOptions = ConnectOptions(),
+        roomOptions: RoomOptions = RoomOptions(),
     ): LivekitRtc.JoinResponse {
-        val joinResponse = connect(url, token, options)
+        val joinResponse = connect(url, token, options, roomOptions)
         return (joinResponse as Either.Left).value
     }
 
@@ -78,26 +83,30 @@ constructor(
         connect(
             url,
             token,
-            ConnectOptions()
-                .apply { reconnect = true }
+            (lastOptions ?: ConnectOptions()).copy()
+                .apply { reconnect = true },
+            lastRoomOptions ?: RoomOptions()
         )
     }
 
     suspend fun connect(
         url: String,
         token: String,
-        options: ConnectOptions
+        options: ConnectOptions,
+        roomOptions: RoomOptions
     ): Either<LivekitRtc.JoinResponse, Unit> {
         // Clean up any pre-existing connection.
         close()
 
-        val wsUrlString = "$url/rtc" + createConnectionParams(token, getClientInfo(), options)
+        val wsUrlString = "$url/rtc" + createConnectionParams(token, getClientInfo(), options, roomOptions)
         isReconnecting = options.reconnect
 
         LKLog.i { "connecting to $wsUrlString" }
 
         coroutineScope = CloseableCoroutineScope(SupervisorJob() + ioDispatcher)
         lastUrl = wsUrlString
+        lastOptions = options
+        lastRoomOptions = roomOptions
 
         val request = Request.Builder()
             .url(wsUrlString)
@@ -113,7 +122,8 @@ constructor(
     private fun createConnectionParams(
         token: String,
         clientInfo: LivekitModels.ClientInfo,
-        options: ConnectOptions
+        options: ConnectOptions,
+        roomOptions: RoomOptions
     ): String {
 
         val queryParams = mutableListOf<Pair<String, String>>()
@@ -123,9 +133,12 @@ constructor(
             queryParams.add(CONNECT_QUERY_RECONNECT to 1.toString())
         }
 
-        val autoSubscribe = if(options.autoSubscribe) 1 else 0
+        val autoSubscribe = if (options.autoSubscribe) 1 else 0
         queryParams.add(CONNECT_QUERY_AUTOSUBSCRIBE to autoSubscribe.toString())
 
+        val adaptiveStream = if (roomOptions.adaptiveStream) 1 else 0
+        queryParams.add(CONNECT_QUERY_ADAPTIVE_STREAM to adaptiveStream.toString())
+
         // Client info
         queryParams.add(CONNECT_QUERY_SDK to "android")
         queryParams.add(CONNECT_QUERY_VERSION to clientInfo.version)
@@ -551,6 +564,9 @@ constructor(
         currentWs = null
         joinContinuation?.cancel()
         joinContinuation = null
+        lastUrl = null
+        lastOptions = null
+        lastRoomOptions = null
     }
 
     interface Listener {
@@ -577,6 +593,7 @@ constructor(
         const val CONNECT_QUERY_TOKEN = "access_token"
         const val CONNECT_QUERY_RECONNECT = "reconnect"
         const val CONNECT_QUERY_AUTOSUBSCRIBE = "auto_subscribe"
+        const val CONNECT_QUERY_ADAPTIVE_STREAM = "adaptive_stream"
         const val CONNECT_QUERY_SDK = "sdk"
         const val CONNECT_QUERY_VERSION = "version"
         const val CONNECT_QUERY_PROTOCOL = "protocol"

livekit-android-sdk/src/test/java/io/livekit/android/room/RoomTest.kt

@@ -22,7 +22,6 @@ import org.mockito.junit.MockitoJUnit
 import org.mockito.kotlin.*
 import org.robolectric.RobolectricTestRunner
 import org.webrtc.EglBase
-import java.lang.Exception
 
 @ExperimentalCoroutinesApi
 @RunWith(RobolectricTestRunner::class)
@@ -65,7 +64,7 @@ class RoomTest {
 
     suspend fun connect() {
         rtcEngine.stub {
-            onBlocking { rtcEngine.join(any(), any(), anyOrNull()) }
+            onBlocking { rtcEngine.join(any(), any(), anyOrNull(), anyOrNull()) }
                 .doSuspendableAnswer {
                     room.onJoinResponse(SignalClientTest.JOIN.join)
                     SignalClientTest.JOIN.join