Improve Coordinator Socket reconnection with TokenRepository (#1580)

* feat: add video moderation

* feat: add video moderation

* feat: add video moderation

* feat: add video moderation

* Make ModerationWarningUiContainer as public api

* remove dependency of stream-video-android-filters-video from android-ui-compose

* remove all public-apis

* revert BlurIntensity enum

* revert BlurIntensity enum

* revert changes in strings.xml

* sdk will render moderation warning ui and will blur the video

* sdk will render moderation warning ui and will blur the video

* sdk will render moderation warning ui and will blur the video

* Add default moderation filter

* Spotless

* update

* Spotless

* refactor all moderation related configuration into ModerationConfig

* refactor all moderation related configuration into ModerationConfig

* refactor all moderation related configuration into ModerationConfig

* refactor

* add ut and comments

* add ut and comments

* minor changes

* Make blurring a part of core-sdk

* rename methods

* add token repository

* add expiry-time in demo-app

* spotless and apidump

* single thread coordinator socket

* single thread coordinator socket

* single thread coordinator socket

* single thread coordinator socket

* refactor api files

* refactor

* 1. increase token expiry time
2. Refactor usage token

* 1. increase token expiry time
2. Refactor usage token

* 1. Update token provider

* chore: remove comments

* chore: improve token provider

* chore: update FakeTokenManager with PersistingTokenProvider

* chore: Improve socket disconnection logging and add TODO

The log message for socket disconnection now includes the specific error for better debugging.

Additionally, a TODO comment was added to the `sendData` method in `SocketActions.kt` to refactor it to return a boolean in the future.

* remove comments

* chore: remove logs and commented out code

* exclude StreamVideoBuilderTest test from standard test

* exclude StreamVideoBuilderTest test from standard test

* add unit tests

---------

Co-authored-by: Aleksandar Apostolov <[email protected]>

Author: rahul-lohra

demo-app/build.gradle.kts

@@ -292,6 +292,9 @@ dependencies {
 
     implementation(libs.audioswitch)
 
+    // Logging
+    implementation(libs.okhttp.logging)
+
     // Also Leak Canary added in the previous block
 
     // Instrumentation tests

demo-app/src/main/kotlin/io/getstream/video/android/data/services/stream/StreamService.kt

@@ -17,11 +17,14 @@
 package io.getstream.video.android.data.services.stream
 
 import com.jakewharton.retrofit2.converter.kotlinx.serialization.asConverterFactory
+import io.getstream.log.streamLog
 import io.getstream.video.android.model.User
 import io.getstream.video.android.models.UserCredentials
 import io.getstream.video.android.models.builtInCredentials
 import kotlinx.serialization.json.Json
 import okhttp3.MediaType.Companion.toMediaType
+import okhttp3.OkHttpClient
+import okhttp3.logging.HttpLoggingInterceptor
 import retrofit2.Retrofit
 import retrofit2.create
 import retrofit2.http.GET
@@ -32,23 +35,34 @@ fun interface StreamService {
     suspend fun getAuthData(
         @Query("environment") environment: String,
         @Query("user_id") userId: String?,
+        @Query("exp") exp: Int,
     ): GetAuthDataResponse
 
     companion object {
         private const val BASE_URL = "https://pronto.getstream.io/"
+        const val TOKEN_EXPIRY_TIME = 7 * 24 * 60 * 60 // 7d * 24 hrs * 60 mins * 60 sec
 
         private val json = Json { ignoreUnknownKeys = true }
-
+        private val okHttpClient = OkHttpClient.Builder()
+            .addInterceptor(
+                HttpLoggingInterceptor {
+                    streamLog(tag = "Video:Http") { it }
+                }.apply {
+                    level = HttpLoggingInterceptor.Level.BODY
+                },
+            )
+            .build()
         private val retrofit = Retrofit.Builder()
             .baseUrl(BASE_URL)
             .addConverterFactory(json.asConverterFactory("application/json".toMediaType()))
+            .client(okHttpClient)
             .build()
 
         private val serviceInstance = retrofit.create<StreamService>()
 
-        val instance = StreamService { environment, userId ->
+        val instance = StreamService { environment, userId, exp ->
             User.builtInCredentials[userId]?.toAuthDataResponse()
-                ?: serviceInstance.getAuthData(environment, userId)
+                ?: serviceInstance.getAuthData(environment, userId, exp)
         }
     }
 }

demo-app/src/main/kotlin/io/getstream/video/android/ui/login/LoginViewModel.kt

@@ -92,6 +92,7 @@ class LoginViewModel @Inject constructor(
                     val authData = StreamService.instance.getAuthData(
                         environment = it.env,
                         userId = userId,
+                        StreamService.TOKEN_EXPIRY_TIME,
                     )
                     val loggedInGoogleUser =
                         if (autoLogIn) null else googleAccountRepository.getCurrentUser()
@@ -128,6 +129,7 @@ class LoginViewModel @Inject constructor(
                     val authData = StreamService.instance.getAuthData(
                         environment = it.env,
                         userId = user.id,
+                        StreamService.TOKEN_EXPIRY_TIME,
                     )
                     // Store the data in the demo app
                     dataStore.updateUser(user)

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

@@ -61,6 +61,7 @@ import io.getstream.video.android.util.config.AppConfig
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.firstOrNull
+import kotlinx.coroutines.runBlocking
 
 public enum class InitializedState {
     NOT_STARTED, RUNNING, FINISHED, FAILED
@@ -124,6 +125,7 @@ object StreamVideoInitHelper {
                 authData = StreamService.instance.getAuthData(
                     environment = AppConfig.currentEnvironment.value!!.env,
                     userId = userId,
+                    StreamService.TOKEN_EXPIRY_TIME,
                 )
 
                 loggedInUser = User(id = authData.userId, role = "admin")
@@ -140,6 +142,7 @@ object StreamVideoInitHelper {
                     authData = StreamService.instance.getAuthData(
                         environment = AppConfig.currentEnvironment.value!!.env,
                         userId = loggedInUser.id,
+                        StreamService.TOKEN_EXPIRY_TIME,
                     )
                 }
 
@@ -197,7 +200,19 @@ object StreamVideoInitHelper {
 
         chatClient.connectUser(
             user = chatUser,
-            token = token,
+            tokenProvider = object : io.getstream.chat.android.client.token.TokenProvider {
+                override fun loadToken(): String {
+                    return runBlocking {
+                        val email = user.custom?.get("email")
+                        val authData = StreamService.instance.getAuthData(
+                            environment = AppConfig.currentEnvironment.value!!.env,
+                            userId = email,
+                            StreamService.TOKEN_EXPIRY_TIME,
+                        )
+                        authData.token
+                    }
+                }
+            },
         ).enqueue()
     }
 
@@ -303,10 +318,13 @@ object StreamVideoInitHelper {
             ),
             tokenProvider = object : TokenProvider {
                 override suspend fun loadToken(): String {
-                    val email = user.custom?.get("email")
+                    val userEmail = user.custom?.get("email")
+                    val userId = user.id
+                    val userIdForTokenRenewal = if (userEmail.isNullOrEmpty()) userId else userEmail
                     val authData = StreamService.instance.getAuthData(
                         environment = AppConfig.currentEnvironment.value!!.env,
-                        userId = email,
+                        userId = userIdForTokenRenewal,
+                        StreamService.TOKEN_EXPIRY_TIME,
                     )
                     return authData.token
                 }

stream-video-android-core/api/stream-video-android-core.api

@@ -13118,9 +13118,17 @@ public abstract interface class io/getstream/video/android/core/socket/common/to
 	public abstract fun loadToken (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
+public final class io/getstream/video/android/core/socket/common/token/TokenRepository {
+	public fun <init> ()V
+	public fun <init> (Ljava/lang/String;)V
+	public synthetic fun <init> (Ljava/lang/String;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public final fun getToken ()Ljava/lang/String;
+	public final fun updateToken (Ljava/lang/String;)V
+}
+
 public class io/getstream/video/android/core/socket/coordinator/CoordinatorSocketConnection : io/getstream/video/android/core/socket/common/SocketListener, io/getstream/video/android/core/socket/common/SocketActions {
-	public fun <init> (Ljava/lang/String;Ljava/lang/String;Lio/getstream/video/android/model/User;Ljava/lang/String;Lokhttp3/OkHttpClient;Lio/getstream/video/android/core/internal/network/NetworkStateProvider;Lkotlinx/coroutines/CoroutineScope;Landroidx/lifecycle/Lifecycle;Lio/getstream/video/android/core/socket/common/token/TokenProvider;)V
-	public synthetic fun <init> (Ljava/lang/String;Ljava/lang/String;Lio/getstream/video/android/model/User;Ljava/lang/String;Lokhttp3/OkHttpClient;Lio/getstream/video/android/core/internal/network/NetworkStateProvider;Lkotlinx/coroutines/CoroutineScope;Landroidx/lifecycle/Lifecycle;Lio/getstream/video/android/core/socket/common/token/TokenProvider;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun <init> (Ljava/lang/String;Ljava/lang/String;Lio/getstream/video/android/model/User;Ljava/lang/String;Lokhttp3/OkHttpClient;Lio/getstream/video/android/core/internal/network/NetworkStateProvider;Lkotlinx/coroutines/CoroutineScope;Landroidx/lifecycle/Lifecycle;Lio/getstream/video/android/core/socket/common/token/TokenProvider;Lio/getstream/video/android/core/socket/common/token/TokenRepository;)V
+	public synthetic fun <init> (Ljava/lang/String;Ljava/lang/String;Lio/getstream/video/android/model/User;Ljava/lang/String;Lokhttp3/OkHttpClient;Lio/getstream/video/android/core/internal/network/NetworkStateProvider;Lkotlinx/coroutines/CoroutineScope;Landroidx/lifecycle/Lifecycle;Lio/getstream/video/android/core/socket/common/token/TokenProvider;Lio/getstream/video/android/core/socket/common/token/TokenRepository;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public fun connect (Lio/getstream/video/android/model/User;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public synthetic fun connect (Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun connectionId ()Lkotlinx/coroutines/flow/StateFlow;
@@ -13327,8 +13335,8 @@ public final class io/getstream/video/android/core/socket/coordinator/state/Vide
 
 public final class io/getstream/video/android/core/socket/sfu/SfuSocketConnection : io/getstream/video/android/core/socket/common/SocketListener, io/getstream/video/android/core/socket/common/SocketActions {
 	public static final field Companion Lio/getstream/video/android/core/socket/sfu/SfuSocketConnection$Companion;
-	public fun <init> (Ljava/lang/String;Ljava/lang/String;Lokhttp3/OkHttpClient;Lio/getstream/video/android/core/internal/network/NetworkStateProvider;Lkotlinx/coroutines/CoroutineScope;Landroidx/lifecycle/Lifecycle;Lio/getstream/video/android/core/socket/common/token/TokenProvider;)V
-	public synthetic fun <init> (Ljava/lang/String;Ljava/lang/String;Lokhttp3/OkHttpClient;Lio/getstream/video/android/core/internal/network/NetworkStateProvider;Lkotlinx/coroutines/CoroutineScope;Landroidx/lifecycle/Lifecycle;Lio/getstream/video/android/core/socket/common/token/TokenProvider;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun <init> (Ljava/lang/String;Ljava/lang/String;Lokhttp3/OkHttpClient;Lio/getstream/video/android/core/internal/network/NetworkStateProvider;Lkotlinx/coroutines/CoroutineScope;Landroidx/lifecycle/Lifecycle;Lio/getstream/video/android/core/socket/common/token/TokenProvider;Lio/getstream/video/android/core/socket/common/token/TokenRepository;)V
+	public synthetic fun <init> (Ljava/lang/String;Ljava/lang/String;Lokhttp3/OkHttpClient;Lio/getstream/video/android/core/internal/network/NetworkStateProvider;Lkotlinx/coroutines/CoroutineScope;Landroidx/lifecycle/Lifecycle;Lio/getstream/video/android/core/socket/common/token/TokenProvider;Lio/getstream/video/android/core/socket/common/token/TokenRepository;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public synthetic fun connect (Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun connect (Lstream/video/sfu/event/JoinRequest;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun connectionId ()Lkotlinx/coroutines/flow/StateFlow;

stream-video-android-core/build.gradle.kts

@@ -240,3 +240,27 @@ mavenPublishing {
         ),
     )
 }
+
+afterEvaluate {
+    tasks.named("testDebugUnitTest") {
+        dependsOn("isolatedTest")
+    }
+}
+
+tasks.register<Test>("isolatedTest") {
+
+    description = "Runs StreamVideoBuilderTest in an isolation"
+    group = "verification"
+
+    // Only this class
+    include("**/StreamVideoBuilderTest.class")
+
+    // Force new JVM
+    forkEvery = 1
+
+    // Reuse Android's debug unit test configuration
+    val androidTestTask = tasks.named<Test>("testDebugUnitTest").get()
+
+    testClassesDirs = androidTestTask.testClassesDirs
+    classpath = androidTestTask.classpath
+}

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

@@ -37,8 +37,9 @@ import io.getstream.video.android.core.permission.android.DefaultStreamPermissio
 import io.getstream.video.android.core.permission.android.StreamPermissionCheck
 import io.getstream.video.android.core.socket.common.scope.ClientScope
 import io.getstream.video.android.core.socket.common.scope.UserScope
-import io.getstream.video.android.core.socket.common.token.ConstantTokenProvider
+import io.getstream.video.android.core.socket.common.token.RepositoryTokenProvider
 import io.getstream.video.android.core.socket.common.token.TokenProvider
+import io.getstream.video.android.core.socket.common.token.TokenRepository
 import io.getstream.video.android.core.sounds.RingingCallVibrationConfig
 import io.getstream.video.android.core.sounds.Sounds
 import io.getstream.video.android.core.sounds.defaultResourcesRingingConfig
@@ -95,6 +96,7 @@ import java.net.ConnectException
  * @see ClientState.connection
  *
  */
+
 public class StreamVideoBuilder @JvmOverloads constructor(
     context: Context,
     private val apiKey: ApiKey,
@@ -106,7 +108,7 @@ public class StreamVideoBuilder @JvmOverloads constructor(
         object : TokenProvider {
             override suspend fun loadToken(): String = legacy.invoke(null)
         }
-    } ?: ConstantTokenProvider(token),
+    } ?: RepositoryTokenProvider(tokenRepository),
     private val loggingLevel: LoggingLevel = LoggingLevel(),
     private val notificationConfig: NotificationConfig = NotificationConfig(),
     private val ringNotification: ((call: Call) -> Notification?)? = null,
@@ -214,7 +216,7 @@ public class StreamVideoBuilder @JvmOverloads constructor(
 
         // Android JSR-310 backport backport
         AndroidThreeTen.init(context)
-
+        tokenRepository.updateToken(token)
         // This connection module class exposes the connections to the various retrofit APIs.
         val coordinatorConnectionModule = CoordinatorConnectionModule(
             context = context,
@@ -225,9 +227,9 @@ public class StreamVideoBuilder @JvmOverloads constructor(
             loggingLevel = loggingLevel,
             user = user,
             apiKey = apiKey,
-            userToken = token,
             tokenProvider = tokenProvider,
             lifecycle = lifecycle,
+            tokenRepository = tokenRepository,
         )
 
         val deviceTokenStorage = DeviceTokenStorage(context)
@@ -272,6 +274,7 @@ public class StreamVideoBuilder @JvmOverloads constructor(
             vibrationConfig = vibrationConfig,
             enableStereoForSubscriber = enableStereoForSubscriber,
             telecomConfig = telecomConfig,
+            tokenRepository = tokenRepository,
         )
 
         if (user.type == UserType.Guest) {
@@ -348,6 +351,11 @@ public class StreamVideoBuilder @JvmOverloads constructor(
     }
 }
 
+/**
+ * Refactor Later
+ */
+internal val tokenRepository = TokenRepository("")
+
 sealed class GEO {
     /** Run calls over our global edge network, this is the default and right for most applications */
     object GlobalEdgeNetwork : GEO()

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

@@ -107,8 +107,9 @@ import io.getstream.video.android.core.permission.android.DefaultStreamPermissio
 import io.getstream.video.android.core.permission.android.StreamPermissionCheck
 import io.getstream.video.android.core.socket.ErrorResponse
 import io.getstream.video.android.core.socket.common.scope.ClientScope
-import io.getstream.video.android.core.socket.common.token.ConstantTokenProvider
+import io.getstream.video.android.core.socket.common.token.RepositoryTokenProvider
 import io.getstream.video.android.core.socket.common.token.TokenProvider
+import io.getstream.video.android.core.socket.common.token.TokenRepository
 import io.getstream.video.android.core.socket.coordinator.state.VideoSocketState
 import io.getstream.video.android.core.sounds.CallSoundAndVibrationPlayer
 import io.getstream.video.android.core.sounds.RingingCallVibrationConfig
@@ -161,7 +162,8 @@ internal class StreamVideoClient internal constructor(
     internal var token: String,
     private val lifecycle: Lifecycle,
     internal val coordinatorConnectionModule: CoordinatorConnectionModule,
-    internal val tokenProvider: TokenProvider = ConstantTokenProvider(token),
+    internal val tokenRepository: TokenRepository,
+    internal val tokenProvider: TokenProvider = RepositoryTokenProvider(tokenRepository),
     internal val streamNotificationManager: StreamNotificationManager,
     internal val enableCallNotificationUpdates: Boolean,
     internal val callServiceConfigRegistry: CallServiceConfigRegistry = CallServiceConfigRegistry(),
@@ -273,6 +275,7 @@ internal class StreamVideoClient internal constructor(
             // Retry once with a new token if the token is expired
             if (e.isAuthError()) {
                 val newToken = tokenProvider.loadToken()
+                tokenRepository.updateToken(newToken)
                 token = newToken
                 coordinatorConnectionModule.updateToken(newToken)
                 apiCall()

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt

@@ -78,6 +78,7 @@ import io.getstream.video.android.core.model.VideoTrack
 import io.getstream.video.android.core.model.toPeerType
 import io.getstream.video.android.core.socket.common.VideoParser
 import io.getstream.video.android.core.socket.common.parser2.MoshiVideoParser
+import io.getstream.video.android.core.socket.common.token.TokenRepository
 import io.getstream.video.android.core.socket.sfu.state.SfuSocketState
 import io.getstream.video.android.core.toJson
 import io.getstream.video.android.core.trace.PeerConnectionTraceKey
@@ -236,12 +237,12 @@ public class RtcSession internal constructor(
             apiUrl = sfuUrl,
             wssUrl = sfuWsUrl,
             connectionTimeoutInMs = 2000L,
-            userToken = sfuToken,
             lifecycle = lifecycle,
             onSignalingLost = { error ->
                 call.debug.fastReconnect()
             },
             tracer = sfuTracer,
+            tokenRepository = TokenRepository(sfuToken),
         )
     },
 ) {

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

@@ -76,11 +76,6 @@ internal interface ConnectionModuleDeclaration<Api, SocketConnection, Http : OkH
      */
     val loggingLevel: LoggingLevel get() = LoggingLevel()
 
-    /**
-     * The user token.
-     */
-    val userToken: Token
-
     /**
      * The lifecycle of the application.
      */