Implement RegionUrlProvider and Room.prepareConnection (#463)

* implement RegionUrlProvider and Room.prepareConnection * spotless * fix test * ensure signal client uses proper scheme when connecting * more scheme enforcement

Implement RegionUrlProvider and Room.prepareConnection (#463)
* implement RegionUrlProvider and Room.prepareConnection * spotless * fix test * ensure signal client uses proper scheme when connecting * more scheme enforcement
davidliu · GitHub
Commit d526b58e9db9a49290286dca547302ec2acc63fe d526b58e 1 parent c256a01d
gradle/libs.versions.toml
livekit-android-sdk/build.gradle
livekit-android-sdk/src/main/java/io/livekit/android/dagger/WebModule.kt
livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt
livekit-android-sdk/src/main/java/io/livekit/android/room/RegionUrlProvider.kt
livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt
livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt
livekit-android-sdk/src/main/java/io/livekit/android/room/util/ConnectionWarmer.kt
livekit-android-sdk/src/main/java/io/livekit/android/util/OkHttpCallExt.kt
livekit-android-sdk/src/main/java/io/livekit/android/util/UrlUtils.kt
livekit-android-test/build.gradle
livekit-android-test/src/main/java/io/livekit/android/test/mock/dagger/TestWebModule.kt
livekit-android-test/src/main/java/io/livekit/android/test/mock/room/util/MockConnectionWarmer.kt
livekit-android-test/src/test/java/io/livekit/android/room/RegionUrlProviderTest.kt
livekit-android-test/src/test/java/io/livekit/android/room/RoomTest.kt
--- a/gradle/libs.versions.toml
查看文件 @d526b58
+++ b/gradle/libs.versions.toml
查看文件 @d526b58
@@ -51,7 +51,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" }
-okhttp = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp" }
+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" }
 semver4j = { module = "com.vdurmont:semver4j", version.ref = "semver4j" }
 webrtc = { module = "io.github.webrtc-sdk:android-prefixed", version.ref = "webrtc" }
@@ -79,6 +80,7 @@ espresso = { module = "androidx.test.espresso:espresso-core", version = "3.5.1" 
 junit = { module = "junit:junit", version.ref = "junit-lib" }
 junitJupiterApi = { module = "org.junit.jupiter:junit-jupiter-api", version.ref = "junit-jupiter" }
 junitJupiterEngine = { module = "org.junit.jupiter:junit-jupiter-engine", version.ref = "junit-jupiter" }
+okhttp-mockwebserver = { module = "com.squareup.okhttp3:mockwebserver", version.ref = "okhttp" }
 # Mockito 5 requires JVM 11.
 #noinspection GradleDependency
--- a/livekit-android-sdk/build.gradle
查看文件 @d526b58
+++ b/livekit-android-sdk/build.gradle
查看文件 @d526b58
@@ -144,7 +144,8 @@ dependencies {
     implementation libs.coroutines.lib
     implementation libs.kotlinx.serialization.json
     api libs.webrtc
-    api libs.okhttp
+    api libs.okhttp.lib
+    implementation libs.okhttp.coroutines
     api libs.audioswitch
     implementation libs.androidx.annotation
     implementation libs.androidx.core
--- a/livekit-android-sdk/src/main/java/io/livekit/android/dagger/WebModule.kt
查看文件 @d526b58
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/dagger/WebModule.kt
查看文件 @d526b58
@@ -27,6 +27,8 @@ import io.livekit.android.room.network.NetworkCallbackManagerFactory
 import io.livekit.android.room.network.NetworkCallbackManagerImpl
 import io.livekit.android.room.network.NetworkCallbackRegistry
 import io.livekit.android.room.network.NetworkCallbackRegistryImpl
+import io.livekit.android.room.util.ConnectionWarmer
+import io.livekit.android.room.util.OkHttpConnectionWarmer
 import io.livekit.android.stats.AndroidNetworkInfo
 import io.livekit.android.stats.NetworkInfo
 import okhttp3.OkHttpClient
@@ -47,6 +49,9 @@ internal object WebModule {
     }
     @Provides
+    fun connectionWarmer(okHttpConnectionWarmer: OkHttpConnectionWarmer): ConnectionWarmer = okHttpConnectionWarmer
+
+    @Provides
     fun websocketFactory(okHttpClient: OkHttpClient): WebSocket.Factory {
         return okHttpClient
     }
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt
查看文件 @d526b58
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt
查看文件 @d526b58
@@ -135,6 +135,8 @@ internal constructor(
     private val pendingTrackResolvers: MutableMap<String, Continuation<LivekitModels.TrackInfo>> =
         mutableMapOf()
+
+    internal var regionUrlProvider: RegionUrlProvider? = null
     private var sessionUrl: String? = null
     private var sessionToken: String? = null
     private var connectOptions: ConnectOptions? = null
@@ -368,6 +370,7 @@ internal constructor(
         connectOptions = null
         lastRoomOptions = null
         participantSid = null
+        regionUrlProvider = null
         closeResources(reason)
         connectionState = ConnectionState.DISCONNECTED
     }
@@ -418,7 +421,7 @@ internal constructor(
             LKLog.d { "Skip reconnection - engine is closed" }
             return
         }
-        val url = sessionUrl
+        var url = sessionUrl
         val token = sessionToken
         if (url == null || token == null) {
             LKLog.w { "couldn't reconnect, no url or no token" }
@@ -432,6 +435,15 @@ internal constructor(
             val reconnectStartTime = SystemClock.elapsedRealtime()
             for (retries in 0 until MAX_RECONNECT_RETRIES) {
+                // First try use previously valid url.
+                if (retries != 0) {
+                    try {
+                        url = regionUrlProvider?.getNextBestRegionUrl() ?: url
+                    } catch (e: Exception) {
+                        LKLog.d(e) { "Exception while getting next best region url while reconnecting." }
+                    }
+                }
+
                 ensureActive()
                 if (retries != 0) {
                     yield()
@@ -469,7 +481,7 @@ internal constructor(
                     try {
                         closeResources("Full Reconnecting")
                         listener?.onFullReconnecting()
-                        joinImpl(url, token, connectOptions, lastRoomOptions ?: RoomOptions())
+                        joinImpl(url!!, token, connectOptions, lastRoomOptions ?: RoomOptions())
                     } catch (e: Exception) {
                         LKLog.w(e) { "Error during reconnection." }
                         // reconnect failed, retry.
@@ -484,7 +496,7 @@ internal constructor(
                     LKLog.v { "Attempting soft reconnect." }
                     subscriber?.prepareForIceRestart()
                     try {
-                        val response = client.reconnect(url, token, participantSid)
+                        val response = client.reconnect(url!!, token, participantSid)
                         if (response is Either.Left) {
                             val reconnectResponse = response.value
                             val rtcConfig = makeRTCConfig(Either.Right(reconnectResponse), connectOptions)
@@ -514,7 +526,7 @@ internal constructor(
                     break
                 }
-                // wait until ICE connected
+                // wait until publisher ICE connected
                 val endTime = SystemClock.elapsedRealtime() + MAX_ICE_CONNECT_TIMEOUT_MS
                 if (hasPublished) {
                     while (SystemClock.elapsedRealtime() < endTime) {
@@ -532,6 +544,7 @@ internal constructor(
                     break
                 }
+                // wait until subscriber ICE connected
                 while (SystemClock.elapsedRealtime() < endTime) {
                     if (subscriber?.isConnected() == true) {
                         LKLog.v { "reconnected to ICE" }
@@ -546,14 +559,18 @@ internal constructor(
                     LKLog.v { "RTCEngine closed, aborting reconnection" }
                     break
                 }
+
                 if (connectionState == ConnectionState.CONNECTED &&
                     (!hasPublished || publisher?.isConnected() == true)
                 ) {
+                    // Is connected, notify and return.
+                    regionUrlProvider?.clearAttemptedRegions()
                     client.onPCConnected()
                     listener?.onPostReconnect(isFullReconnect)
                     return@launch
                 }
+                // Didn't manage to reconnect, check if should continue to next attempt.
                 val curReconnectTime = SystemClock.elapsedRealtime() - reconnectStartTime
                 if (curReconnectTime > MAX_RECONNECT_TIMEOUT) {
                     break
@@ -954,6 +971,7 @@ internal constructor(
     override fun onRefreshToken(token: String) {
         sessionToken = token
+        regionUrlProvider?.token = token
     }
     override fun onLocalTrackUnpublished(trackUnpublished: LivekitRtc.TrackUnpublishedResponse) {
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/RegionUrlProvider.kt 0 → 100644
查看文件 @d526b58
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/RegionUrlProvider.kt 0 → 100644
查看文件 @d526b58
+/*
+ * Copyright 2024 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 android.os.SystemClock
+import androidx.annotation.VisibleForTesting
+import dagger.assisted.Assisted
+import dagger.assisted.AssistedFactory
+import dagger.assisted.AssistedInject
+import io.livekit.android.util.LKLog
+import io.livekit.android.util.executeAsync
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.coroutineScope
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.decodeFromString
+import kotlinx.serialization.json.Json
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import java.net.URI
+
+/**
+ * @suppress
+ */
+class RegionUrlProvider
+@AssistedInject
+constructor(
+    @Assisted val serverUrl: URI,
+    @Assisted var token: String,
+    private val okHttpClient: OkHttpClient,
+    private val json: Json,
+) {
+    private var regionSettings: RegionSettings? = null
+    private var lastUpdateAt: Long = 0L
+    private var settingsCacheTimeMs = 30000
+    private var attemptedRegions = mutableSetOf<RegionInfo>()
+
+    fun isLKCloud() = serverUrl.isLKCloud()
+
+    @Throws(RoomException.ConnectException::class)
+    suspend fun getNextBestRegionUrl() = coroutineScope {
+        if (!isLKCloud()) {
+            throw IllegalStateException("Region availability is only supported for LiveKit Cloud domains")
+        }
+        if (regionSettings == null || SystemClock.elapsedRealtime() - lastUpdateAt > settingsCacheTimeMs) {
+            fetchRegionSettings()
+        }
+        val regions = regionSettings?.regions ?: return@coroutineScope null
+        val regionsLeft = regions.filter { region ->
+            !attemptedRegions.any { attempted -> attempted.url == region.url }
+        }
+
+        if (regionsLeft.isEmpty()) {
+            return@coroutineScope null
+        }
+
+        val nextRegion = regionsLeft.first()
+        attemptedRegions.add(nextRegion)
+        LKLog.d { "next region: $nextRegion" }
+        return@coroutineScope nextRegion.url
+    }
+
+    @Throws(RoomException.ConnectException::class)
+    @OptIn(ExperimentalCoroutinesApi::class)
+    suspend fun fetchRegionSettings(): RegionSettings? {
+        val request = Request.Builder()
+            .url(serverUrl.getCloudConfigUrl("/regions").toString())
+            .header("Authorization", "Bearer $token")
+            .build()
+        val bodyString = okHttpClient.newCall(request)
+            .executeAsync()
+            .use { response ->
+                if (!response.isSuccessful) {
+                    throw RoomException.ConnectException("Could not fetch region settings: ${response.code} ${response.message}")
+                }
+                return@use response.body?.string() ?: return null
+            }
+
+        LKLog.e { bodyString }
+        return json.decodeFromString<RegionSettings>(bodyString).also {
+            regionSettings = it
+            lastUpdateAt = SystemClock.elapsedRealtime()
+        }
+    }
+
+    fun clearAttemptedRegions() {
+        attemptedRegions.clear()
+    }
+
+    @AssistedFactory
+    interface Factory {
+        fun create(serverUrl: URI, token: String): RegionUrlProvider
+    }
+}
+
+internal fun URI.isLKCloud() = regionUrlProviderTesting || host.endsWith(".livekit.cloud") || host.endsWith(".livekit.run")
+
+internal fun URI.getCloudConfigUrl(appendPath: String = ""): URI {
+    val scheme = if (this.scheme.startsWith("ws")) {
+        this.scheme.replaceFirst("ws", "http")
+    } else {
+        this.scheme
+    }
+    return URI(
+        scheme,
+        null,
+        this.host,
+        this.port,
+        "/settings$appendPath",
+        null,
+        null,
+    )
+}
+
+private var regionUrlProviderTesting = false
+
+@VisibleForTesting
+fun setRegionUrlProviderTesting(enable: Boolean) {
+    regionUrlProviderTesting = enable
+}
+
+/**
+ * @suppress
+ */
+@Serializable
+data class RegionSettings(val regions: List<RegionInfo>)
+
+/**
+ * @suppress
+ */
+@Serializable
+data class RegionInfo(val region: String, val url: String, val distance: Long)
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt
查看文件 @d526b58
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt
查看文件 @d526b58
@@ -43,6 +43,7 @@ import io.livekit.android.room.participant.*
 import io.livekit.android.room.provisions.LKObjects
 import io.livekit.android.room.track.*
 import io.livekit.android.room.types.toSDKType
+import io.livekit.android.room.util.ConnectionWarmer
 import io.livekit.android.util.FlowObservable
 import io.livekit.android.util.LKLog
 import io.livekit.android.util.flow
@@ -59,6 +60,7 @@ import livekit.LivekitModels
 import livekit.LivekitRtc
 import livekit.org.webrtc.*
 import livekit.org.webrtc.audio.AudioDeviceModule
+import java.net.URI
 import javax.inject.Named
 class Room
@@ -84,6 +86,8 @@ constructor(
     val lkObjects: LKObjects,
     networkCallbackManagerFactory: NetworkCallbackManagerFactory,
     private val audioDeviceModule: AudioDeviceModule,
+    private val regionUrlProviderFactory: RegionUrlProvider.Factory,
+    private val connectionWarmer: ConnectionWarmer,
 ) : RTCEngine.Listener, ParticipantListener {
     private lateinit var coroutineScope: CoroutineScope
@@ -263,6 +267,9 @@ constructor(
     private var stateLock = Mutex()
+    private var regionUrlProvider: RegionUrlProvider? = null
+    private var regionUrl: String? = null
+
     private fun getCurrentRoomOptions(): RoomOptions =
         RoomOptions(
             adaptiveStream = adaptiveStream,
@@ -275,6 +282,44 @@ constructor(
         )
     /**
+     * prepareConnection should be called as soon as the page is loaded, in order
+     * to speed up the connection attempt. This function will
+     * - perform DNS resolution and pre-warm the DNS cache
+     * - establish TLS connection and cache TLS keys
+     *
+     * With LiveKit Cloud, it will also determine the best edge data center for
+     * the current client to connect to if a token is provided.
+     */
+    suspend fun prepareConnection(url: String, token: String? = null) {
+        if (state != State.DISCONNECTED) {
+            LKLog.i { "Room is not in disconnected state, ignoring prepareConnection call." }
+            return
+        }
+        LKLog.d { "preparing connection to $url" }
+
+        try {
+            val urlActual = URI(url)
+            if (urlActual.isLKCloud() && token != null) {
+                val regionUrlProvider = regionUrlProviderFactory.create(urlActual, token)
+                this.regionUrlProvider = regionUrlProvider
+
+                val regionUrl = regionUrlProvider.getNextBestRegionUrl()
+                // we will not replace the regionUrl if an attempt had already started
+                // to avoid overriding regionUrl after a new connection attempt had started
+                if (regionUrl != null && state == State.DISCONNECTED) {
+                    this.regionUrl = regionUrl
+                    connectionWarmer.fetch(regionUrl)
+                    LKLog.d { "prepared connection to $regionUrl" }
+                }
+            } else {
+                connectionWarmer.fetch(url)
+            }
+        } catch (e: Exception) {
+            LKLog.e(e) { "Error while preparing connection:" }
+        }
+    }
+
+    /**
      * Connect to a LiveKit Room.
      *
      * @param url
@@ -309,60 +354,7 @@ constructor(
             // Setup local participant.
             localParticipant.reinitialize()
-            coroutineScope.launch {
-                localParticipant.events.collect {
-                    when (it) {
-                        is ParticipantEvent.TrackPublished -> emitWhenConnected(
-                            RoomEvent.TrackPublished(
-                                room = this@Room,
-                                publication = it.publication,
-                                participant = it.participant,
-                            ),
-                        )
-
-                        is ParticipantEvent.TrackUnpublished -> emitWhenConnected(
-                            RoomEvent.TrackUnpublished(
-                                room = this@Room,
-                                publication = it.publication,
-                                participant = it.participant,
-                            ),
-                        )
-
-                        is ParticipantEvent.ParticipantPermissionsChanged -> emitWhenConnected(
-                            RoomEvent.ParticipantPermissionsChanged(
-                                room = this@Room,
-                                participant = it.participant,
-                                newPermissions = it.newPermissions,
-                                oldPermissions = it.oldPermissions,
-                            ),
-                        )
-
-                        is ParticipantEvent.MetadataChanged -> {
-                            emitWhenConnected(
-                                RoomEvent.ParticipantMetadataChanged(
-                                    this@Room,
-                                    it.participant,
-                                    it.prevMetadata,
-                                ),
-                            )
-                        }
-
-                        is ParticipantEvent.NameChanged -> {
-                            emitWhenConnected(
-                                RoomEvent.ParticipantNameChanged(
-                                    this@Room,
-                                    it.participant,
-                                    it.name,
-                                ),
-                            )
-                        }
-
-                        else -> {
-                            // do nothing
-                        }
-                    }
-                }
-            }
+            setupLocalParticipantEventHandling()
             if (roomOptions.e2eeOptions != null) {
                 e2eeManager = e2EEManagerFactory.create(roomOptions.e2eeOptions.keyProvider).apply {
@@ -384,7 +376,56 @@ constructor(
             if (audioProcessingController is AuthedAudioProcessingController) {
                 audioProcessingController.authenticate(url, token)
             }
-            engine.join(url, token, options, roomOptions)
+
+            // Don't use URL equals.
+            if (regionUrlProvider?.serverUrl.toString() != url) {
+                regionUrl = null
+                regionUrlProvider = null
+            }
+
+            val urlObj = URI(url)
+            if (urlObj.isLKCloud()) {
+                if (regionUrlProvider == null) {
+                    regionUrlProvider = regionUrlProviderFactory.create(urlObj, token)
+                } else {
+                    regionUrlProvider?.token = token
+                }
+
+                // trigger the first fetch without waiting for a response
+                // if initial connection fails, this will speed up picking regional url
+                // on subsequent runs
+                launch {
+                    try {
+                        regionUrlProvider?.fetchRegionSettings()
+                    } catch (e: Exception) {
+                        LKLog.w(e) { "could not fetch region settings" }
+                    }
+                }
+            }
+
+            var nextUrl: String? = regionUrl ?: url
+            regionUrl = null
+
+            while (nextUrl != null) {
+                val connectUrl = nextUrl
+                nextUrl = null
+                try {
+                    engine.regionUrlProvider = regionUrlProvider
+                    engine.join(connectUrl, token, options, roomOptions)
+                } catch (e: Exception) {
+                    if (e is CancellationException) {
+                        throw e // rethrow to properly cancel.
+                    }
+
+                    nextUrl = regionUrlProvider?.getNextBestRegionUrl()
+                    if (nextUrl != null) {
+                        LKLog.d(e) { "Connection to $connectUrl failed, retrying with another region: $nextUrl" }
+                    } else {
+                        throw e // rethrow since no more regions to try.
+                    }
+                }
+            }
+
             ensureActive()
             networkCallbackManager.registerCallback()
             if (options.audio) {
@@ -495,6 +536,63 @@ constructor(
         }
     }
+    private fun setupLocalParticipantEventHandling() {
+        coroutineScope.launch {
+            localParticipant.events.collect {
+                when (it) {
+                    is ParticipantEvent.TrackPublished -> emitWhenConnected(
+                        RoomEvent.TrackPublished(
+                            room = this@Room,
+                            publication = it.publication,
+                            participant = it.participant,
+                        ),
+                    )
+
+                    is ParticipantEvent.TrackUnpublished -> emitWhenConnected(
+                        RoomEvent.TrackUnpublished(
+                            room = this@Room,
+                            publication = it.publication,
+                            participant = it.participant,
+                        ),
+                    )
+
+                    is ParticipantEvent.ParticipantPermissionsChanged -> emitWhenConnected(
+                        RoomEvent.ParticipantPermissionsChanged(
+                            room = this@Room,
+                            participant = it.participant,
+                            newPermissions = it.newPermissions,
+                            oldPermissions = it.oldPermissions,
+                        ),
+                    )
+
+                    is ParticipantEvent.MetadataChanged -> {
+                        emitWhenConnected(
+                            RoomEvent.ParticipantMetadataChanged(
+                                this@Room,
+                                it.participant,
+                                it.prevMetadata,
+                            ),
+                        )
+                    }
+
+                    is ParticipantEvent.NameChanged -> {
+                        emitWhenConnected(
+                            RoomEvent.ParticipantNameChanged(
+                                this@Room,
+                                it.participant,
+                                it.name,
+                            ),
+                        )
+                    }
+
+                    else -> {
+                        // do nothing
+                    }
+                }
+            }
+        }
+    }
+
     private fun handleParticipantDisconnect(identity: Participant.Identity) {
         val newParticipants = mutableRemoteParticipants.toMutableMap()
         val removedParticipant = newParticipants.remove(identity) ?: return
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt
查看文件 @d526b58
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt
查看文件 @d526b58
@@ -28,6 +28,8 @@ import io.livekit.android.stats.getClientInfo
 import io.livekit.android.util.CloseableCoroutineScope
 import io.livekit.android.util.Either
 import io.livekit.android.util.LKLog
+import io.livekit.android.util.toHttpUrl
+import io.livekit.android.util.toWebsocketUrl
 import io.livekit.android.webrtc.toProtoSessionDescription
 import kotlinx.coroutines.CancellableContinuation
 import kotlinx.coroutines.CoroutineDispatcher
@@ -122,6 +124,7 @@ constructor(
     /**
      * @throws Exception if fails to connect.
      */
+    @Throws(Exception::class)
     suspend fun join(
         url: String,
         token: String,
@@ -135,6 +138,7 @@ constructor(
     /**
      * @throws Exception if fails to connect.
      */
+    @Throws(Exception::class)
     @VisibleForTesting
     suspend fun reconnect(url: String, token: String, participantSid: String?): Either<ReconnectResponse, Unit> {
         val reconnectResponse = connect(
@@ -159,7 +163,7 @@ constructor(
         // Clean up any pre-existing connection.
         close(reason = "Starting new connection", shouldClearQueuedRequests = false)
-        val wsUrlString = "$url/rtc" + createConnectionParams(token, getClientInfo(), options, roomOptions)
+        val wsUrlString = "${url.toWebsocketUrl()}/rtc" + createConnectionParams(token, getClientInfo(), options, roomOptions)
         isReconnecting = options.reconnect
         LKLog.i { "connecting to $wsUrlString" }
@@ -305,7 +309,7 @@ constructor(
         var reason: String? = null
         try {
             lastUrl?.let {
-                val validationUrl = "http" + it.substring(2).replaceFirst("/rtc?", "/rtc/validate?")
+                val validationUrl = it.toHttpUrl().replaceFirst("/rtc?", "/rtc/validate?")
                 val request = Request.Builder().url(validationUrl).build()
                 val resp = okHttpClient.newCall(request).execute()
                 val body = resp.body
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/util/ConnectionWarmer.kt 0 → 100644
查看文件 @d526b58
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/util/ConnectionWarmer.kt 0 → 100644
查看文件 @d526b58
+/*
+ * Copyright 2024 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.util
+
+import io.livekit.android.util.executeAsync
+import io.livekit.android.util.toHttpUrl
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.Response
+import javax.inject.Inject
+
+interface ConnectionWarmer {
+    suspend fun fetch(url: String): Response
+}
+
+class OkHttpConnectionWarmer
+@Inject
+constructor(
+    private val okHttpClient: OkHttpClient,
+) : ConnectionWarmer {
+    @OptIn(ExperimentalCoroutinesApi::class)
+    override suspend fun fetch(url: String): Response {
+        val request = Request.Builder()
+            .url(url.toHttpUrl())
+            .method("HEAD", null)
+            .build()
+        return okHttpClient.newCall(request)
+            .executeAsync()
+    }
+}
--- a/livekit-android-sdk/src/main/java/io/livekit/android/util/OkHttpCallExt.kt 0 → 100644
查看文件 @d526b58
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/util/OkHttpCallExt.kt 0 → 100644
查看文件 @d526b58
+/*
+ * Copyright 2024 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.util
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.suspendCancellableCoroutine
+import okhttp3.Call
+import okhttp3.Callback
+import okhttp3.Response
+import okhttp3.internal.closeQuietly
+import okio.IOException
+import kotlin.coroutines.resumeWithException
+
+// TODO: Switch to official executeAsync when released.
+@ExperimentalCoroutinesApi // resume with a resource cleanup.
+suspend fun Call.executeAsync(): Response =
+    suspendCancellableCoroutine { continuation ->
+        continuation.invokeOnCancellation {
+            this.cancel()
+        }
+        this.enqueue(
+            object : Callback {
+                override fun onFailure(
+                    call: Call,
+                    e: IOException,
+                ) {
+                    continuation.resumeWithException(e)
+                }
+
+                override fun onResponse(
+                    call: Call,
+                    response: Response,
+                ) {
+                    continuation.resume(response) {
+                        response.closeQuietly()
+                    }
+                }
+            },
+        )
+    }
--- a/livekit-android-sdk/src/main/java/io/livekit/android/util/UrlUtils.kt 0 → 100644
查看文件 @d526b58
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/util/UrlUtils.kt 0 → 100644
查看文件 @d526b58
+/*
+ * Copyright 2024 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.util
+
+fun String.toWebsocketUrl(): String {
+    if (startsWith("http")) {
+        return replaceFirst("http", "ws")
+    }
+    return this
+}
+
+fun String.toHttpUrl(): String {
+    if (startsWith("ws")) {
+        return replaceFirst("ws", "http")
+    }
+    return this
+}
--- a/livekit-android-test/build.gradle
查看文件 @d526b58
+++ b/livekit-android-test/build.gradle
查看文件 @d526b58
@@ -78,7 +78,7 @@ dependencies {
     implementation libs.timber
     implementation libs.coroutines.lib
     implementation libs.kotlinx.serialization.json
-    api libs.okhttp
+    api libs.okhttp.lib
     api libs.audioswitch
     implementation libs.androidx.annotation
     api libs.protobuf.javalite
@@ -95,6 +95,7 @@ dependencies {
     testImplementation libs.junit
     testImplementation libs.robolectric
+    testImplementation libs.okhttp.mockwebserver
     kaptTest libs.dagger.compiler
     androidTestImplementation libs.androidx.test.junit
--- a/livekit-android-test/src/main/java/io/livekit/android/test/mock/dagger/TestWebModule.kt
查看文件 @d526b58
+++ b/livekit-android-test/src/main/java/io/livekit/android/test/mock/dagger/TestWebModule.kt
查看文件 @d526b58
@@ -23,6 +23,7 @@ import dagger.Reusable
 import io.livekit.android.memory.CloseableManager
 import io.livekit.android.room.network.NetworkCallbackManagerFactory
 import io.livekit.android.room.network.NetworkCallbackManagerImpl
+import io.livekit.android.room.util.ConnectionWarmer
 import io.livekit.android.stats.NetworkInfo
 import io.livekit.android.stats.NetworkType
 import io.livekit.android.test.mock.MockNetworkCallbackRegistry
@@ -49,6 +50,15 @@ object TestWebModule {
     }
     @Provides
+    fun connectionWarmer(): ConnectionWarmer {
+        return object : ConnectionWarmer {
+            override suspend fun fetch(url: String): Response {
+                return Response.Builder().code(200).build()
+            }
+        }
+    }
+
+    @Provides
     @Singleton
     fun websocketFactory(webSocketFactory: MockWebSocketFactory): WebSocket.Factory {
         return webSocketFactory
--- a/livekit-android-test/src/main/java/io/livekit/android/test/mock/room/util/MockConnectionWarmer.kt 0 → 100644
查看文件 @d526b58
+++ b/livekit-android-test/src/main/java/io/livekit/android/test/mock/room/util/MockConnectionWarmer.kt 0 → 100644
查看文件 @d526b58
+/*
+ * Copyright 2024 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.test.mock.room.util
+
+import io.livekit.android.room.util.ConnectionWarmer
+import okhttp3.Response
+
+class MockConnectionWarmer : ConnectionWarmer {
+    override suspend fun fetch(url: String): Response {
+        return Response.Builder().code(200).build()
+    }
+}
--- a/livekit-android-test/src/test/java/io/livekit/android/room/RegionUrlProviderTest.kt 0 → 100644
查看文件 @d526b58
+++ b/livekit-android-test/src/test/java/io/livekit/android/room/RegionUrlProviderTest.kt 0 → 100644
查看文件 @d526b58
+/*
+ * Copyright 2024 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 io.livekit.android.test.BaseTest
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.serialization.json.Json
+import okhttp3.OkHttpClient
+import okhttp3.mockwebserver.MockResponse
+import okhttp3.mockwebserver.MockWebServer
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertNull
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+
+@OptIn(ExperimentalCoroutinesApi::class)
+@RunWith(RobolectricTestRunner::class)
+class RegionUrlProviderTest : BaseTest() {
+
+    lateinit var server: MockWebServer
+
+    @Before
+    fun setup() {
+        setRegionUrlProviderTesting(true)
+    }
+
+    @After
+    fun tearDown() {
+        setRegionUrlProviderTesting(false)
+    }
+
+    @Test
+    fun fetchRegionSettings() = runTest {
+        server = MockWebServer()
+        server.enqueue(MockResponse().setBody(regionResponse))
+        val regionUrlProvider = RegionUrlProvider(server.url("").toUri(), "token", OkHttpClient.Builder().build(), Json { ignoreUnknownKeys = true })
+
+        val settings = regionUrlProvider.fetchRegionSettings()
+        assertNotNull(settings)
+
+        val regions = settings!!.regions
+        assertNotNull(regions)
+        assertEquals(3, regions.size)
+        val regionA = regions[0]
+        assertEquals("a", regionA.region)
+        assertEquals("https://regiona.livekit.cloud", regionA.url)
+        assertEquals(100L, regionA.distance)
+    }
+
+    @Test
+    fun getNextRegionUrl() = runTest {
+        server = MockWebServer()
+        server.enqueue(MockResponse().setBody(regionResponse))
+        val regionUrlProvider = RegionUrlProvider(server.url("").toUri(), "token", OkHttpClient.Builder().build(), Json { ignoreUnknownKeys = true })
+
+        assertEquals("https://regiona.livekit.cloud", regionUrlProvider.getNextBestRegionUrl())
+        assertEquals("https://regionb.livekit.cloud", regionUrlProvider.getNextBestRegionUrl())
+        assertEquals("https://regionc.livekit.cloud", regionUrlProvider.getNextBestRegionUrl())
+        assertNull(regionUrlProvider.getNextBestRegionUrl())
+
+        // Check that only one request was needed.
+        assertEquals(1, server.requestCount)
+    }
+}
+
+private val regionResponse = """{
+    "regions": [
+        {
+            "region": "a",
+            "url": "https://regiona.livekit.cloud",
+            "distance": "100"
+        },
+        {
+            "region": "b",
+            "url": "https://regionb.livekit.cloud",
+            "distance": "1000"
+        },
+        {
+            "region": "c",
+            "url": "https://regionc.livekit.cloud",
+            "distance": "10000"
+        }
+    ]
+}"""
--- a/livekit-android-test/src/test/java/io/livekit/android/room/RoomTest.kt
查看文件 @d526b58
+++ b/livekit-android-test/src/test/java/io/livekit/android/room/RoomTest.kt
查看文件 @d526b58
@@ -39,6 +39,7 @@ import io.livekit.android.test.mock.MockEglBase
 import io.livekit.android.test.mock.MockLKObjects
 import io.livekit.android.test.mock.MockNetworkCallbackRegistry
 import io.livekit.android.test.mock.TestData
+import io.livekit.android.test.mock.room.util.MockConnectionWarmer
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.async
 import kotlinx.coroutines.flow.MutableSharedFlow
@@ -84,6 +85,9 @@ class RoomTest {
     @Mock
     lateinit var e2EEManagerFactory: E2EEManager.Factory
+    @Mock
+    lateinit var regionUrlProviderFactory: RegionUrlProvider.Factory
+
     lateinit var networkCallbackRegistry: MockNetworkCallbackRegistry
     var eglBase: EglBase = MockEglBase()
@@ -125,6 +129,8 @@ class RoomTest {
                 NetworkCallbackManagerImpl(networkCallback, networkCallbackRegistry)
             },
             audioDeviceModule = MockAudioDeviceModule(),
+            regionUrlProviderFactory = regionUrlProviderFactory,
+            connectionWarmer = MockConnectionWarmer(),
         )
     }