Skip to content

Commit d653ec2

Browse files
PratimMallickcursoragentaleksandar-apostolovrahul-lohra
authored
Pin coordinator join to a specific SFU via sfu_id (#1819)
* feat(core): pin coordinator join to an SFU via sfu_id Co-authored-by: Cursor <cursoragent@cursor.com> * fix(core): pin sfu_id via interceptor to keep joinCall binary-compatible Adding a Retrofit @query replaced the published ProductvideoApi.joinCall signature. Append the coordinator pin on the HTTP request instead. Co-authored-by: Cursor <cursoragent@cursor.com> * style(core): apply Spotless license header on SFU pin interceptor Co-authored-by: Cursor <cursoragent@cursor.com> * refactor(core): move SFU pin to forceSfuId builder function Replace the StreamVideoBuilder sfuId constructor parameter with an @InternalStreamVideoApi forceSfuId() function, matching the existing forceApiUrl / forceWssUrl pattern. The public constructor signature is untouched, so the API dump returns to its develop state. Drop StreamVideoClient.pinnedSfuId, which was written but never read; the interceptor receives the pin through CoordinatorConnectionModule. * refactor(core): only install the SFU pin interceptor when pinned Guard the interceptor at the install site so it is absent from the coordinator OkHttp chain unless a pin is configured, instead of adding a no-op interceptor to every client. The blank check moves to the guard, so the builder passes the configured value through unchanged. --------- Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: Aleksandar Apostolov <apostolov.alexandar@gmail.com> Co-authored-by: Rahul Kumar Lohra <tgunix@gmail.com>
1 parent 693f242 commit d653ec2

5 files changed

Lines changed: 185 additions & 2 deletions

File tree

‎demo-app/src/main/kotlin/io/getstream/video/android/util/StreamVideoInitHelper.kt‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ import io.getstream.video.android.core.StreamVideo
3333
import io.getstream.video.android.core.StreamVideoBuilder
3434
import io.getstream.video.android.core.call.CallType
3535
import io.getstream.video.android.core.internal.ExperimentalStreamVideoApi
36+
import io.getstream.video.android.core.internal.InternalStreamVideoApi
3637
import io.getstream.video.android.core.logging.LoggingLevel
3738
import io.getstream.video.android.core.moderations.ModerationConfig
3839
import io.getstream.video.android.core.moderations.ModerationWarningConfig
@@ -81,13 +82,16 @@ public enum class InitializedState {
8182
* @property secret API secret from the local coordinator — used to sign a JWT for [userId].
8283
* @property userId User ID to connect as.
8384
* @property token Pre-generated JWT token. If null, one is generated from [secret].
85+
* @property sfuId Coordinator edge pin (`?sfu_id=` / `WithPinToSFUID`). When set,
86+
* every join / rejoin / migrate asks the coordinator for this SFU.
8487
*/
8588
data class LocalDevConfig(
8689
val coordinatorAddress: String,
8790
val apiKey: String,
8891
val userId: String,
8992
val secret: String? = null,
9093
val token: String? = null,
94+
val sfuId: String? = null,
9195
) {
9296
init {
9397
require(secret != null || token != null) {
@@ -196,6 +200,7 @@ object StreamVideoInitHelper {
196200
token = localCfg.resolveToken(),
197201
loggingLevel = LoggingLevel(priority = Priority.VERBOSE),
198202
localCoordinatorAddress = localCfg.coordinatorAddress,
203+
sfuId = localCfg.sfuId,
199204
localTokenProvider = object : TokenProvider {
200205
override suspend fun loadToken(): String = localCfg.resolveToken()
201206
},
@@ -367,14 +372,15 @@ object StreamVideoInitHelper {
367372
}
368373

369374
/** Sets up and returns the [StreamVideo] required to connect to the API. */
370-
@OptIn(ExperimentalStreamVideoApi::class)
375+
@OptIn(ExperimentalStreamVideoApi::class, InternalStreamVideoApi::class)
371376
private fun initializeStreamVideo(
372377
context: Context,
373378
apiKey: ApiKey,
374379
user: User,
375380
token: String,
376381
loggingLevel: LoggingLevel,
377382
localCoordinatorAddress: String? = null,
383+
sfuId: String? = null,
378384
localTokenProvider: TokenProvider? = null,
379385
): StreamVideo {
380386
val callServiceConfigRegistry = CallServiceConfigRegistry()
@@ -506,6 +512,8 @@ object StreamVideoInitHelper {
506512
telecomConfig = TelecomConfig(context.packageName),
507513
connectOnInit = false,
508514
rejectCallWhenBusy = false,
509-
).build()
515+
).apply {
516+
sfuId?.takeIf { it.isNotBlank() }?.let(::forceSfuId)
517+
}.build()
510518
}
511519
}

‎stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoBuilder.kt‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -172,6 +172,7 @@ public class StreamVideoBuilder @JvmOverloads constructor(
172172

173173
private var apiUrl: String? = null
174174
private var wssUrl: String? = null
175+
private var sfuId: String? = null
175176

176177
/**
177178
* Set the API URL to be used for the video client.
@@ -193,6 +194,16 @@ public class StreamVideoBuilder @JvmOverloads constructor(
193194
wssUrl = value
194195
}
195196

197+
/**
198+
* Set the SFU id the coordinator should pin every join, rejoin and migrate to.
199+
*
200+
* For testing purposes only.
201+
*/
202+
@InternalStreamVideoApi
203+
public fun forceSfuId(value: String): StreamVideoBuilder = apply {
204+
sfuId = value
205+
}
206+
196207
/**
197208
* Builds the [StreamVideo] client.
198209
*
@@ -261,6 +272,7 @@ public class StreamVideoBuilder @JvmOverloads constructor(
261272
tokenProvider = tokenProvider,
262273
lifecycle = lifecycle,
263274
tokenRepository = tokenRepository,
275+
pinnedSfuId = sfuId,
264276
)
265277

266278
val deviceTokenStorage = DeviceTokenStorage(context)

‎stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/internal/module/CoordinatorConnectionModule.kt‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ internal class CoordinatorConnectionModule(
6060
override val apiKey: ApiKey,
6161
override val lifecycle: Lifecycle,
6262
override val tracer: Tracer = Tracer("coordinator"),
63+
pinnedSfuId: String? = null,
6364
) : ConnectionModuleDeclaration<ProductvideoApi, CoordinatorSocketConnection, OkHttpClient, UserToken> {
6465
// Internals
6566
private val authInterceptor = CoordinatorAuthInterceptor(apiKey, tokenRepository)
@@ -75,6 +76,11 @@ internal class CoordinatorConnectionModule(
7576
override val http: OkHttpClient = OkHttpClient.Builder().addInterceptor(
7677
HeadersInterceptor(HeadersUtil()),
7778
)
79+
.apply {
80+
if (!pinnedSfuId.isNullOrBlank()) {
81+
addInterceptor(CoordinatorSfuPinInterceptor(pinnedSfuId))
82+
}
83+
}
7884
.addInterceptor(authInterceptor).addInterceptor(
7985
HttpLoggingInterceptor {
8086
streamLog(tag = "Video:Http") { it }

‎stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/internal/module/CoordinatorSfuPinInterceptor.kt‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/*
2+
* Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
3+
*
4+
* Licensed under the Stream License;
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://github.com/GetStream/stream-video-android/blob/main/LICENSE
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.getstream.video.android.core.internal.module
18+
19+
import okhttp3.Interceptor
20+
import okhttp3.Response
21+
22+
/**
23+
* Adds `?sfu_id=` on coordinator join requests when a local-dev pin is configured.
24+
*
25+
* The coordinator reads this query via `WithPinToSFUID`. It is not part of the
26+
* published OpenAPI join body, so this interceptor keeps
27+
* [io.getstream.android.video.generated.apis.ProductvideoApi.joinCall]
28+
* binary-compatible instead of adding a Retrofit `@Query`.
29+
*/
30+
internal class CoordinatorSfuPinInterceptor(
31+
private val pinnedSfuId: String?,
32+
) : Interceptor {
33+
companion object {
34+
const val QUERY_SFU_ID = "sfu_id"
35+
const val JOIN_PATH_SUFFIX = "/join"
36+
}
37+
38+
override fun intercept(chain: Interceptor.Chain): Response {
39+
val request = chain.request()
40+
val pin = pinnedSfuId?.takeIf { it.isNotBlank() }
41+
if (pin == null ||
42+
!request.url.encodedPath.endsWith(JOIN_PATH_SUFFIX) ||
43+
request.url.queryParameter(QUERY_SFU_ID) != null
44+
) {
45+
return chain.proceed(request)
46+
}
47+
val url = request.url.newBuilder()
48+
.addQueryParameter(QUERY_SFU_ID, pin)
49+
.build()
50+
return chain.proceed(request.newBuilder().url(url).build())
51+
}
52+
}

‎stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/internal/module/CoordinatorSfuPinInterceptorTest.kt‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
/*
2+
* Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
3+
*
4+
* Licensed under the Stream License;
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://github.com/GetStream/stream-video-android/blob/main/LICENSE
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.getstream.video.android.core.internal.module
18+
19+
import com.google.common.truth.Truth.assertThat
20+
import okhttp3.MediaType.Companion.toMediaType
21+
import okhttp3.OkHttpClient
22+
import okhttp3.Protocol
23+
import okhttp3.Request
24+
import okhttp3.RequestBody.Companion.toRequestBody
25+
import okhttp3.Response
26+
import okhttp3.ResponseBody.Companion.toResponseBody
27+
import org.junit.Test
28+
29+
class CoordinatorSfuPinInterceptorTest {
30+
31+
@Test
32+
fun `adds sfu_id on coordinator join when a pin is configured`() {
33+
val proceeded = intercept(
34+
pinnedSfuId = "SFU-1",
35+
url = "https://video.stream-io-api.com/video/call/default/abc/join?connection_id=c1",
36+
)
37+
38+
assertThat(proceeded.url.queryParameter("sfu_id")).isEqualTo("SFU-1")
39+
assertThat(proceeded.url.queryParameter("connection_id")).isEqualTo("c1")
40+
}
41+
42+
@Test
43+
fun `leaves non-join requests unchanged`() {
44+
val proceeded = intercept(
45+
pinnedSfuId = "SFU-1",
46+
url = "https://video.stream-io-api.com/video/call/default/abc",
47+
)
48+
49+
assertThat(proceeded.url.queryParameter("sfu_id")).isNull()
50+
}
51+
52+
@Test
53+
fun `leaves join unchanged when no pin is configured`() {
54+
val proceeded = intercept(
55+
pinnedSfuId = null,
56+
url = "https://video.stream-io-api.com/video/call/default/abc/join",
57+
)
58+
59+
assertThat(proceeded.url.queryParameter("sfu_id")).isNull()
60+
}
61+
62+
@Test
63+
fun `leaves join unchanged when the pin is blank`() {
64+
val proceeded = intercept(
65+
pinnedSfuId = " ",
66+
url = "https://video.stream-io-api.com/video/call/default/abc/join",
67+
)
68+
69+
assertThat(proceeded.url.queryParameter("sfu_id")).isNull()
70+
}
71+
72+
@Test
73+
fun `does not duplicate an existing sfu_id`() {
74+
val proceeded = intercept(
75+
pinnedSfuId = "SFU-2",
76+
url = "https://video.stream-io-api.com/video/call/default/abc/join?sfu_id=SFU-1",
77+
)
78+
79+
assertThat(proceeded.url.queryParameterValues("sfu_id")).containsExactly("SFU-1")
80+
}
81+
82+
private fun intercept(pinnedSfuId: String?, url: String): Request {
83+
lateinit var proceeded: Request
84+
val client = OkHttpClient.Builder()
85+
.addInterceptor(CoordinatorSfuPinInterceptor(pinnedSfuId))
86+
.addInterceptor { chain ->
87+
proceeded = chain.request()
88+
Response.Builder()
89+
.request(proceeded)
90+
.protocol(Protocol.HTTP_1_1)
91+
.code(200)
92+
.message("OK")
93+
.body("".toResponseBody())
94+
.build()
95+
}
96+
.build()
97+
client.newCall(
98+
Request.Builder()
99+
.url(url)
100+
.post("{}".toRequestBody("application/json".toMediaType()))
101+
.build(),
102+
).execute().close()
103+
return proceeded
104+
}
105+
}

0 commit comments

Comments
 (0)