diff --git a/stream-video-android-core/build.gradle.kts b/stream-video-android-core/build.gradle.kts index d3998fa6462..a614927485a 100644 --- a/stream-video-android-core/build.gradle.kts +++ b/stream-video-android-core/build.gradle.kts @@ -86,6 +86,12 @@ android { unitTests { isIncludeAndroidResources = true isReturnDefaultValues = true + all { test -> + // Forked test-executor JVMs do not inherit org.gradle.jvmargs and default to + // 512m, which the suite (MockK inline instrumentation + Robolectric + mock + // web servers) now exceeds on CI runners. + test.maxHeapSize = "2g" + } } managedDevices { diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt index 692eac46e3b..ec272362ea5 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt @@ -563,8 +563,9 @@ public class Call( "You can re-define your permissions and their expected state by overriding the [permissionCheck] in [StreamVideoBuilder]\n" } } - // if we are a guest user, make sure we wait for the token before running the join flow - clientImpl.guestUserJob?.await() + // Guest-token acquisition is now single-flighted through core's StreamTokenManager + // via the StreamTokenProvider wired at builder time; downstream REST/WS calls that + // require the token will suspend transparently through the auth interceptor. // Ensure factory is created with the current audioBitrateProfile before joining ensureFactoryMatchesAudioProfile() diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/ClientState.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/ClientState.kt index 3f56f463da7..3c499c4003d 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/ClientState.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/ClientState.kt @@ -76,6 +76,14 @@ class ClientState(private val client: StreamVideo) { /** Current user for the client. */ public val user: StateFlow = _user + /** + * Adopts [user] as the SDK's active user. Used by the guest flow to publish the + * server-issued identity returned by `createGuest` (iOS analog: `state.user`). + */ + internal fun updateUser(user: User) { + _user.value = user + } + /** Coordinator connection state */ public val connection: StateFlow = _connection diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoBuilder.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoBuilder.kt index 56f7df4434d..43a1c8a3f77 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoBuilder.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoBuilder.kt @@ -18,8 +18,22 @@ package io.getstream.video.android.core import android.app.Notification import android.content.Context +import android.os.Build import androidx.lifecycle.ProcessLifecycleOwner import com.jakewharton.threetenabp.AndroidThreeTen +import io.getstream.android.core.api.StreamClient +import io.getstream.android.core.api.authentication.StreamTokenProvider +import io.getstream.android.core.api.log.StreamLoggerProvider +import io.getstream.android.core.api.model.StreamUser +import io.getstream.android.core.api.model.config.StreamClientSerializationConfig +import io.getstream.android.core.api.model.config.StreamComponentProvider +import io.getstream.android.core.api.model.config.StreamSocketConfig +import io.getstream.android.core.api.model.value.StreamApiKey +import io.getstream.android.core.api.model.value.StreamHttpClientInfoHeader +import io.getstream.android.core.api.model.value.StreamUserId +import io.getstream.android.core.api.model.value.StreamWsUrl +import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleMonitor +import io.getstream.android.core.api.subscribe.StreamSubscriptionManager import io.getstream.log.AndroidStreamLogger import io.getstream.log.StreamLog import io.getstream.log.streamLog @@ -40,6 +54,11 @@ import io.getstream.video.android.core.socket.common.scope.UserScope 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.v2.AnonymousStreamTokenProvider +import io.getstream.video.android.core.socket.coordinator.v2.GuestStreamTokenProvider +import io.getstream.video.android.core.socket.coordinator.v2.IntegrationStreamTokenProvider +import io.getstream.video.android.core.socket.coordinator.v2.VideoEventSerialization +import io.getstream.video.android.core.socket.coordinator.v2.WritableUserRepository import io.getstream.video.android.core.sounds.RingingCallVibrationConfig import io.getstream.video.android.core.sounds.Sounds import io.getstream.video.android.core.sounds.defaultResourcesRingingConfig @@ -52,6 +71,7 @@ import io.getstream.video.android.model.UserType import io.getstream.webrtc.ManagedAudioProcessingFactory import kotlinx.coroutines.launch import java.net.ConnectException +import java.util.concurrent.atomic.AtomicReference /** * The [StreamVideoBuilder] is used to create a new instance of the [StreamVideo] client. This is the @@ -169,6 +189,26 @@ public class StreamVideoBuilder @JvmOverloads constructor( private var apiUrl: String? = null private var wssUrl: String? = null + /** + * Seam that constructs the core [StreamClient] instance. Tests substitute this lambda so they + * can assert factory arguments without invoking core's Android-service-dependent + * initialisation (RESEARCH Pitfall 2). + */ + @InternalStreamVideoApi + internal var streamClientFactory: (StreamClientFactoryArgs) -> StreamClient = + ::defaultStreamClientFactory + + /** + * Substitute the [StreamClient] factory used by [build]. Intended for preview/snapshot + * harnesses that instantiate the video client under Paparazzi's layoutlib bridge, where + * core's default factory reaches Android system services (e.g. WifiManager) that are + * unavailable in the test environment. + */ + @InternalStreamVideoApi + public fun streamClientFactory( + factory: (StreamClientFactoryArgs) -> StreamClient, + ): StreamVideoBuilder = apply { streamClientFactory = factory } + /** * Set the API URL to be used for the video client. * @@ -255,6 +295,53 @@ public class StreamVideoBuilder @JvmOverloads constructor( tokenRepository = tokenRepository, ) + // Sink that adopts the server-issued guest identity into ClientState once the client + // exists (bound below, after construction). iOS analog: `state.user = guestInfo.user`. + val userRepository = BindableWritableUserRepository() + + // Select the core token provider based on user type — mirrors iOS's + // `initialConnectIfRequired` dispatch. + // - Authenticated: adapt the integration-supplied TokenProvider. + // - Guest: SDK obtains the JWT via createGuest REST call (D-06). Until the token + // arrives, REST auth runs as "anonymous" (same as iOS's AnonymousAuth transport for + // createGuest); GuestStreamTokenProvider flips it to "jwt" and syncs the token into + // the legacy Retrofit path the moment the server issues it. + // - Anonymous: REST-only (D-07). Construction succeeds; connect attempts fail. + val streamTokenProvider: StreamTokenProvider = when (user.type) { + UserType.Authenticated -> IntegrationStreamTokenProvider(tokenProvider) + UserType.Guest -> { + coordinatorConnectionModule.updateAuthType("anonymous") + GuestStreamTokenProvider( + api = coordinatorConnectionModule.api, + initialUser = user, + userRepository = userRepository, + onTokenIssued = { issuedToken -> + coordinatorConnectionModule.updateAuthType("jwt") + coordinatorConnectionModule.updateToken(issuedToken) + }, + ) + } + UserType.Anonymous -> { + coordinatorConnectionModule.updateAuthType("anonymous") + AnonymousStreamTokenProvider() + } + } + + // Build the core client and eagerly inject it into StreamVideoClient (D-03). + val streamClient = streamClientFactory.invoke( + StreamClientFactoryArgs( + scope = scope, + context = context, + user = user, + apiKey = apiKey, + resolvedWssUrl = resolvedWssUrl, + appName = appName, + appVersion = null, + lifecycle = lifecycle, + tokenProvider = streamTokenProvider, + ), + ) + val deviceTokenStorage = DeviceTokenStorage(context) // Install the StreamNotificationManager to configure push notifications. @@ -281,6 +368,7 @@ public class StreamVideoBuilder @JvmOverloads constructor( token = token, lifecycle = lifecycle, coordinatorConnectionModule = coordinatorConnectionModule, + streamClient = streamClient, tokenProvider = tokenProvider, streamNotificationManager = streamNotificationManager, enableCallNotificationUpdates = notificationConfig.enableCallNotificationUpdates, @@ -301,18 +389,20 @@ public class StreamVideoBuilder @JvmOverloads constructor( rejectCallWhenBusy = rejectCallWhenBusy, ) - if (user.type == UserType.Guest) { - coordinatorConnectionModule.updateAuthType("anonymous") - client.setupGuestUser(user) - } else if (user.type == UserType.Anonymous) { - coordinatorConnectionModule.updateAuthType("anonymous") - } + // Route guest identity adoption into ClientState now that the client exists. Bound + // before the connect launch below, so no token load can race the binding. + userRepository.bind { adoptedUser -> client.state.updateUser(adoptedUser) } - // Establish a WS connection with the coordinator (we don't support this for anonymous users) - if (user.type == UserType.Authenticated) { + // Establish a WS connection with the coordinator for authenticated AND guest users — + // iOS parity (`initialConnectIfRequired`): only anonymous users never connect. For + // guests, core's token manager drives the createGuest flow inside connect(). + // Push-device registration stays authenticated-only (guests have no durable identity). + if (user.type != UserType.Anonymous) { scope.launch { try { - if (notificationConfig.autoRegisterPushDevice) { + if (user.type == UserType.Authenticated && + notificationConfig.autoRegisterPushDevice + ) { client.registerPushDevice() } if (connectOnInit) { @@ -386,3 +476,103 @@ sealed class GEO { /** Run calls over our global edge network, this is the default and right for most applications */ object GlobalEdgeNetwork : GEO() } + +/** + * Aggregates the arguments handed to [StreamClient]'s factory function. Grouping them makes the + * `streamClientFactory` seam a single-parameter lambda that is easy to substitute in tests. + */ +@InternalStreamVideoApi +public data class StreamClientFactoryArgs( + val scope: kotlinx.coroutines.CoroutineScope, + val context: Context, + val user: User, + val apiKey: ApiKey, + val resolvedWssUrl: String, + val appName: String?, + val appVersion: String?, + val lifecycle: androidx.lifecycle.Lifecycle, + val tokenProvider: StreamTokenProvider, +) + +private fun defaultStreamClientFactory(args: StreamClientFactoryArgs): StreamClient { + val logProvider: StreamLoggerProvider = StreamLoggerProvider.defaultAndroidLogger() + + val clientInfoHeader = StreamHttpClientInfoHeader.create( + product = "video", + productVersion = BuildConfig.STREAM_VIDEO_VERSION, + os = "Android", + apiLevel = Build.VERSION.SDK_INT, + deviceModel = Build.MODEL, + app = args.appName ?: "unknown", + appVersion = args.appVersion ?: "unknown", + ) + + // D-06 / D-07: guest users authenticate via jwt (JWT obtained via createGuest); + // never use the anonymous factory for the coordinator socket. + val socketConfig = StreamSocketConfig.jwt( + url = StreamWsUrl.fromString(args.resolvedWssUrl), + apiKey = StreamApiKey.fromString(args.apiKey), + clientInfoHeader = clientInfoHeader, + ) + + val serializationConfig = StreamClientSerializationConfig.default( + productEvents = VideoEventSerialization(), + ) + + // Route the integration-supplied Lifecycle through core's LifecycleMonitor so that the + // legacy StreamLifecycleObserver is no longer needed on the coordinator path (D-10). + val streamLifecycleMonitor = StreamLifecycleMonitor( + logger = logProvider.taggedLogger("SVLifecycleMonitor"), + lifecycle = args.lifecycle, + subscriptionManager = StreamSubscriptionManager( + logger = logProvider.taggedLogger("SVLifecycleSubs"), + ), + ) + + return StreamClient( + scope = args.scope, + context = args.context.applicationContext, + user = StreamUser( + id = StreamUserId.fromString(args.user.id), + name = args.user.name, + imageURL = args.user.image, + customData = args.user.custom ?: emptyMap(), + ), + tokenProvider = args.tokenProvider, + products = listOf("video"), + socketConfig = socketConfig, + serializationConfig = serializationConfig, + // Phase 2: keep video's separate Retrofit OkHttpClient; HTTP unification is Phase 4+. + httpConfig = null, + components = StreamComponentProvider( + logProvider = logProvider, + lifecycleMonitor = streamLifecycleMonitor, + ), + ) +} + +/** + * [WritableUserRepository] whose sink is bound after the [StreamVideoClient] is constructed + * (the guest token provider is created before the client exists). A user written before + * [bind] is buffered and delivered on binding, so no adoption is lost even if a token load + * completes first. + */ +internal class BindableWritableUserRepository : WritableUserRepository { + @Volatile + private var sink: ((User) -> Unit)? = null + private val pending = AtomicReference(null) + + fun bind(sink: (User) -> Unit) { + this.sink = sink + pending.getAndSet(null)?.let(sink) + } + + override fun setUser(user: User) { + val target = sink + if (target != null) { + target(user) + } else { + pending.set(user) + } + } +} diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoClient.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoClient.kt index f6a1a7ee665..92f733a0663 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoClient.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoClient.kt @@ -20,6 +20,10 @@ import android.content.Context import android.media.AudioAttributes import androidx.collection.LruCache import androidx.lifecycle.Lifecycle +import io.getstream.android.core.api.StreamClient +import io.getstream.android.core.api.model.connection.StreamConnectionState +import io.getstream.android.core.api.socket.listeners.StreamClientListener +import io.getstream.android.core.api.subscribe.StreamSubscription import io.getstream.android.push.PushDevice import io.getstream.android.video.generated.models.AcceptCallResponse import io.getstream.android.video.generated.models.BlockUserRequest @@ -29,8 +33,6 @@ import io.getstream.android.video.generated.models.CallRequest import io.getstream.android.video.generated.models.CallRingEvent import io.getstream.android.video.generated.models.CallSettingsRequest import io.getstream.android.video.generated.models.CollectUserFeedbackRequest -import io.getstream.android.video.generated.models.CreateGuestRequest -import io.getstream.android.video.generated.models.CreateGuestResponse import io.getstream.android.video.generated.models.GetCallResponse import io.getstream.android.video.generated.models.GetOrCreateCallRequest import io.getstream.android.video.generated.models.GetOrCreateCallResponse @@ -75,7 +77,6 @@ import io.getstream.android.video.generated.models.UpdateCallMembersResponse import io.getstream.android.video.generated.models.UpdateCallRequest import io.getstream.android.video.generated.models.UpdateCallResponse import io.getstream.android.video.generated.models.UpdateUserPermissionsResponse -import io.getstream.android.video.generated.models.UserRequest import io.getstream.android.video.generated.models.VideoEvent import io.getstream.android.video.generated.models.WSCallEvent import io.getstream.log.taggedLogger @@ -114,7 +115,6 @@ import io.getstream.video.android.core.socket.common.scope.ClientScope 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 import io.getstream.video.android.core.sounds.Sounds @@ -129,6 +129,7 @@ import io.getstream.video.android.core.utils.toQueriedMembers import io.getstream.video.android.model.ApiKey import io.getstream.video.android.model.Device import io.getstream.video.android.model.User +import io.getstream.video.android.model.UserType import io.getstream.webrtc.ManagedAudioProcessingFactory import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineScope @@ -144,6 +145,7 @@ import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.map import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking import kotlinx.coroutines.suspendCancellableCoroutine import kotlinx.coroutines.withContext import kotlinx.coroutines.withTimeoutOrNull @@ -168,6 +170,7 @@ internal class StreamVideoClient internal constructor( internal var token: String, private val lifecycle: Lifecycle, internal val coordinatorConnectionModule: CoordinatorConnectionModule, + internal val streamClient: StreamClient, internal val tokenRepository: TokenRepository, internal val tokenProvider: TokenProvider = RepositoryTokenProvider(tokenRepository), internal val streamNotificationManager: StreamNotificationManager, @@ -203,8 +206,6 @@ internal class StreamVideoClient internal constructor( /** if true we fail fast on errors instead of logging them */ - internal var guestUserJob: Deferred? = null - public override val userId = user.id private val logger by taggedLogger("Call:StreamVideo") @@ -217,7 +218,27 @@ internal class StreamVideoClient internal constructor( internal fun getAudioContext(): AudioExecutionContext = audioExecutionContext - val socketImpl = coordinatorConnectionModule.socketConnection + private var streamClientSubscription: StreamSubscription? = null + + private val streamClientListener = object : StreamClientListener { + override fun onEvent(event: Any) { + if (event is VideoEvent) { + fireEvent(event) + } else { + logger.e { "[onEvent] non-VideoEvent received: ${event::class.simpleName}" } + } + } + + override fun onState(state: StreamConnectionState) { + // TODO(02-03): replace with this@StreamVideoClient.state.handleStreamState(state) + // once ClientState.handleStreamState lands in Plan 02-03. Held to validate the + // listener wiring at SDK init; connection-state routing is deferred to the next plan. + } + + override fun onError(err: Throwable) { + logger.e(err) { "[StreamClient] error" } + } + } fun onCallCleanUp(call: Call) { if (enableCallUpdatesAfterLeave) { @@ -232,6 +253,15 @@ internal class StreamVideoClient internal constructor( // remove all cached calls calls.clear() destroyedCalls.evictAll() + // cancel the StreamClient subscription before tearing down the socket + streamClientSubscription?.cancel() + streamClientSubscription = null + // disconnect the StreamClient socket before cancelling the scope; suspend bridged via + // runBlocking because cleanup() is non-suspending public API. + runBlocking { + runCatching { streamClient.disconnect() } + .onFailure { logger.e(it) { "[cleanup] StreamClient.disconnect failed" } } + } // stop all running coroutines scope.cancel() // call cleanup on the active call @@ -285,12 +315,14 @@ internal class StreamVideoClient internal constructor( try { apiCall() } catch (e: HttpException) { - // Retry once with a new token if the token is expired + // Retry once with a new token if the token is expired. + // REST-side: tokenRepository update stays for CoordinatorAuthInterceptor + // (Phase 4+ unification). WS-side: handled by core's StreamTokenManager + // via the StreamTokenProvider wired at builder time. if (e.isAuthError()) { val newToken = tokenProvider.loadToken() tokenRepository.updateToken(newToken) token = newToken - coordinatorConnectionModule.updateToken(newToken) apiCall() } else { throw e @@ -383,35 +415,24 @@ internal class StreamVideoClient internal constructor( } override suspend fun connectIfNotAlreadyConnected() = safeSuspendingCall { - coordinatorConnectionModule.socketConnection.connect(user) + // Anonymous users never open a coordinator socket (D-07; iOS parity). + if (user.type == UserType.Anonymous) return@safeSuspendingCall + if (streamClient.connectionState.value !is StreamConnectionState.Connected) { + streamClient.connect() + } } init { - // listen to socket events and errors - scope.launch(CoroutineName("init#coordinatorSocket.events.collect")) { - coordinatorConnectionModule.socketConnection.events().collect { - fireEvent(it) - } - } - scope.launch { - coordinatorConnectionModule.socketConnection.state().collect { - state.handleState(it) - } - } - - scope.launch(CoroutineName("init#coordinatorSocket.errors.collect")) { - coordinatorConnectionModule.socketConnection.errors().collect { error -> - state.handleError(error.streamError) - } - } - - scope.launch(CoroutineName("init#coordinatorSocket.connectionState.collect")) { - coordinatorConnectionModule.socketConnection.state().collect { it -> - // If the socket is reconnected then we have a new connection ID. - // We need to re-watch every watched call with the new connection ID - // (otherwise the WS events will stop) + // Subscribe once to the StreamClient's listener stream. Events cast to VideoEvent + // are forwarded to the existing fireEvent(...) dispatch pipeline; connection state + // routing is deferred to Plan 02-03 (see streamClientListener.onState). + streamClientSubscription = streamClient.subscribe(streamClientListener).getOrThrow() + + // Re-watch active calls whenever the coordinator socket (re)connects. + scope.launch(CoroutineName("init#streamClient.connectionState.collect")) { + streamClient.connectionState.collect { coreState -> val watchedCalls = calls - if (it is VideoSocketState.Connected && watchedCalls.isNotEmpty()) { + if (coreState is StreamConnectionState.Connected && watchedCalls.isNotEmpty()) { val filter = Filters.`in`("cid", watchedCalls.values.map { it.cid }).toMap() queryCalls(filters = filter, watch = true).also { if (it is Failure) { @@ -475,23 +496,20 @@ internal class StreamVideoClient internal constructor( override suspend fun connectAsync(): Deferred> { return scope.async { - // wait for the guest user setup if we're using guest users - guestUserJob?.await() - try { - val startTime = System.currentTimeMillis() - socketImpl.connect(user) - val duration = System.currentTimeMillis() - startTime - Success(duration) - } catch (e: ErrorResponse) { - if (e.code == VideoErrorCode.TOKEN_EXPIRED.code) { - refreshToken(e) - Failure(Error.GenericError("Initialize error. Token expired.")) - } else { - Failure(Error.ThrowableError("Error to connect user", e)) - } - } catch (e: Throwable) { - Failure(Error.ThrowableError("Error to connect user", e)) + // Anonymous users never open a coordinator socket (D-07). Fail fast without + // touching the network — iOS parity with `connectUser` throwing MissingPermissions. + if (user.type == UserType.Anonymous) { + return@async Failure( + Error.GenericError("Anonymous users cannot connect to the coordinator WS."), + ) } + val startTime = System.currentTimeMillis() + val result = streamClient.connect() + val duration = System.currentTimeMillis() - startTime + result.fold( + onSuccess = { Success(duration) }, + onFailure = { Failure(Error.ThrowableError("Error to connect user", it)) }, + ) } } @@ -499,17 +517,6 @@ internal class StreamVideoClient internal constructor( return connectAsync().await() } - private suspend fun refreshToken(error: Throwable) { - tokenProvider?.let { - val newToken = tokenProvider.loadToken() - coordinatorConnectionModule.updateToken(newToken) - - logger.d { "[refreshToken] Token has been refreshed with: $newToken" } - - socketImpl.reconnect(user, true) - } - } - /** * @see StreamVideo.deleteDevice */ @@ -517,35 +524,6 @@ internal class StreamVideoClient internal constructor( return streamNotificationManager.deleteDevice(device) } - fun setupGuestUser(user: User) { - coordinatorConnectionModule.updateToken("") - guestUserJob = scope.async { - val response = createGuestUser( - userRequest = UserRequest( - id = user.id, - image = user.image, - name = user.name, - custom = user.custom, - ), - ) - if (response.isFailure) { - throw IllegalStateException("Failed to create guest user") - } - response.onSuccess { - coordinatorConnectionModule.updateAuthType("jwt") - coordinatorConnectionModule.updateToken(it.accessToken) - } - } - } - - suspend fun createGuestUser(userRequest: UserRequest): Result { - return apiCall { - coordinatorConnectionModule.api.createGuest( - createGuestRequest = CreateGuestRequest(userRequest), - ) - } - } - internal suspend fun registerPushDevice() { streamNotificationManager.registerPushDevice() } @@ -755,9 +733,9 @@ internal class StreamVideoClient internal constructor( // We return null on timeout. The Coordinator WS will update the connectionId later // after it reconnects (it will call queryCalls) val connectionId = withTimeoutOrNull(timeMillis = WAIT_FOR_CONNECTION_ID_TIMEOUT) { - val value = - coordinatorConnectionModule.socketConnection.connectionId().first { it != null } - value + val connected = streamClient.connectionState + .first { it is StreamConnectionState.Connected } as StreamConnectionState.Connected + connected.connectionId }.also { logger.d { "[waitForConnectionId]: $it" } } diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/internal/module/CoordinatorConnectionModule.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/internal/module/CoordinatorConnectionModule.kt index 476857eb703..a9df59a9dba 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/internal/module/CoordinatorConnectionModule.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/internal/module/CoordinatorConnectionModule.kt @@ -25,16 +25,13 @@ import io.getstream.log.streamLog import io.getstream.video.android.core.header.HeadersUtil import io.getstream.video.android.core.internal.network.NetworkStateProvider import io.getstream.video.android.core.logging.LoggingLevel -import io.getstream.video.android.core.socket.common.scope.UserScope 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.CoordinatorSocketConnection import io.getstream.video.android.core.trace.Tracer import io.getstream.video.android.model.ApiKey import io.getstream.video.android.model.User import io.getstream.video.android.model.UserToken import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers import okhttp3.OkHttpClient import okhttp3.logging.HttpLoggingInterceptor import retrofit2.Retrofit @@ -60,7 +57,7 @@ internal class CoordinatorConnectionModule( override val apiKey: ApiKey, override val lifecycle: Lifecycle, override val tracer: Tracer = Tracer("coordinator"), -) : ConnectionModuleDeclaration { +) : ConnectionModuleDeclaration { // Internals private val authInterceptor = CoordinatorAuthInterceptor(apiKey, tokenRepository) private val retrofit: Retrofit by lazy { @@ -95,18 +92,11 @@ internal class CoordinatorConnectionModule( ) } override val api: ProductvideoApi by lazy { retrofit.create(ProductvideoApi::class.java) } - override val socketConnection: CoordinatorSocketConnection = CoordinatorSocketConnection( - apiKey = apiKey, - url = wssUrl, - user = user, - token = tokenRepository.getToken(), - httpClient = http, - networkStateProvider = networkStateProvider, - scope = UserScope(context = scope.coroutineContext + Dispatchers.IO.limitedParallelism(1)), - lifecycle = lifecycle, - tokenProvider = tokenProvider, - tokenRepository = tokenRepository, - ) + + // Phase 2 D-11: coordinator socket is owned by StreamClient (constructed in + // StreamVideoBuilder.build()); this field is a vestigial slot to satisfy the legacy + // ConnectionModuleDeclaration generic and is deleted in Phase 8 alongside the interface. + override val socketConnection: Unit = Unit override fun updateToken(token: UserToken) { tokenRepository.updateToken(token) diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/lifecycle/StreamLifecycleObserver.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/lifecycle/legacy/StreamLifecycleObserver.kt similarity index 95% rename from stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/lifecycle/StreamLifecycleObserver.kt rename to stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/lifecycle/legacy/StreamLifecycleObserver.kt index b1dd21fdbff..8fbe1271813 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/lifecycle/StreamLifecycleObserver.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/lifecycle/legacy/StreamLifecycleObserver.kt @@ -14,13 +14,14 @@ * limitations under the License. */ -package io.getstream.video.android.core.lifecycle +package io.getstream.video.android.core.lifecycle.legacy import androidx.lifecycle.DefaultLifecycleObserver import androidx.lifecycle.Lifecycle import androidx.lifecycle.LifecycleOwner import io.getstream.log.taggedLogger import io.getstream.video.android.core.dispatchers.DispatcherProvider +import io.getstream.video.android.core.lifecycle.LifecycleHandler import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.launch import kotlinx.coroutines.withContext diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/CoordinatorSocket.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/CoordinatorSocket.kt index 0e08e9aad65..57a13c94ec1 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/CoordinatorSocket.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/CoordinatorSocket.kt @@ -28,7 +28,7 @@ import io.getstream.video.android.core.errors.DisconnectCause import io.getstream.video.android.core.errors.VideoErrorCode import io.getstream.video.android.core.internal.network.NetworkStateProvider import io.getstream.video.android.core.lifecycle.ConnectionPolicyLifecycleHandler -import io.getstream.video.android.core.lifecycle.StreamLifecycleObserver +import io.getstream.video.android.core.lifecycle.legacy.StreamLifecycleObserver import io.getstream.video.android.core.socket.common.CallAwareConnectionPolicy import io.getstream.video.android.core.socket.common.ConnectionConf import io.getstream.video.android.core.socket.common.DISPOSE_SOCKET_REASON diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/CoordinatorSocketConnection.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/CoordinatorSocketConnection.kt index 2ddafa1ed23..fd5d88e7ec8 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/CoordinatorSocketConnection.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/CoordinatorSocketConnection.kt @@ -26,7 +26,7 @@ import io.getstream.android.video.generated.models.WSAuthMessageRequest import io.getstream.log.taggedLogger import io.getstream.video.android.core.errors.DisconnectCause import io.getstream.video.android.core.internal.network.NetworkStateProvider -import io.getstream.video.android.core.lifecycle.StreamLifecycleObserver +import io.getstream.video.android.core.lifecycle.legacy.StreamLifecycleObserver import io.getstream.video.android.core.socket.common.SocketActions import io.getstream.video.android.core.socket.common.SocketFactory import io.getstream.video.android.core.socket.common.SocketListener diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/v2/AnonymousStreamTokenProvider.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/v2/AnonymousStreamTokenProvider.kt new file mode 100644 index 00000000000..5f620a137f7 --- /dev/null +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/v2/AnonymousStreamTokenProvider.kt @@ -0,0 +1,38 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-video-android/blob/main/LICENSE + * + * 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.getstream.video.android.core.socket.coordinator.v2 + +import io.getstream.android.core.api.authentication.StreamTokenProvider +import io.getstream.android.core.api.model.value.StreamToken +import io.getstream.android.core.api.model.value.StreamUserId + +/** + * [StreamTokenProvider] for anonymous users, who never open a coordinator socket (D-07). + * + * Anonymous users are REST-only: the video client constructs normally, but any explicit + * connect attempt fails before touching the network. Mirrors the iOS SDK's behavior where + * `connectUser` throws `ClientError.MissingPermissions` for anonymous users while + * initialization itself succeeds. + */ +internal class AnonymousStreamTokenProvider : StreamTokenProvider { + + override suspend fun loadToken(userId: StreamUserId): StreamToken = + throw IllegalStateException( + "Anonymous users do not open a coordinator socket (D-07). " + + "REST-only operations remain available on the client.", + ) +} diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/v2/GuestStreamTokenProvider.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/v2/GuestStreamTokenProvider.kt index ac5c41bd20e..14775746b17 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/v2/GuestStreamTokenProvider.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/v2/GuestStreamTokenProvider.kt @@ -39,9 +39,18 @@ internal interface WritableUserRepository { /** * [StreamTokenProvider] implementation for guest users (COORD-09, D-06). * - * On invocation, calls the coordinator's `POST /video/guest` endpoint via [ProductvideoApi], - * adopts the server-issued user identity through [userRepository], and returns the resulting - * JWT wrapped in a [StreamToken]. + * Mirrors the iOS SDK's guest flow (`StreamVideo.loadGuestUserInfo`): on invocation, calls the + * coordinator's `POST /video/guest` endpoint via [ProductvideoApi], adopts the server-issued + * user identity through [userRepository], publishes the JWT to the SDK's REST auth path via + * [onTokenIssued], and returns the token wrapped in a [StreamToken] for core's WS auth path. + * + * The video SDK keeps two auth surfaces until Phase 4 unifies HTTP through core: the coordinator + * WS (core's `StreamTokenManager`, fed by this provider's return value) and the legacy Retrofit + * `CoordinatorAuthInterceptor` (fed by [onTokenIssued]). Both must observe the guest JWT — + * iOS expresses the same invariant through its single `tokenSubject`. + * + * Token refresh matches iOS: an expired guest token is renewed by calling `createGuest` again + * (`StreamTokenManager` re-invokes this provider). * * `StreamTokenManager` wraps this provider in a `StreamSingleFlightProcessor`, so concurrent * callers share the same in-flight request. No additional deduplication is needed here (unlike @@ -51,6 +60,7 @@ internal class GuestStreamTokenProvider( private val api: ProductvideoApi, private val initialUser: User, private val userRepository: WritableUserRepository, + private val onTokenIssued: (String) -> Unit, ) : StreamTokenProvider { override suspend fun loadToken(userId: StreamUserId): StreamToken { @@ -66,7 +76,25 @@ internal class GuestStreamTokenProvider( ) // Adopt the server-issued user identity so downstream SDK components observe the // canonical id/role instead of the placeholder used at builder time (AND-1202). - userRepository.setUser(response.user.toUser().copy(type = UserType.Guest)) + userRepository.setUser(adoptServerUser(response.user.toUser())) + // Sync the JWT into the legacy REST auth path before core proceeds to WS auth. + onTokenIssued(response.accessToken) return StreamToken.fromString(response.accessToken) } + + /** + * Ports iOS's name-preservation quirk (`StreamVideo.loadGuestUserInfo`): the server decorates + * guest display names (e.g. `guest--`); when the decorated name's last `-` component + * equals the locally supplied name, keep the local name. + */ + private fun adoptServerUser(serverUser: User): User { + val localName = initialUser.name + val lastNameComponent = serverUser.name?.split("-")?.lastOrNull() + val resolvedName = if (!localName.isNullOrBlank() && lastNameComponent == localName) { + localName + } else { + serverUser.name + } + return serverUser.copy(type = UserType.Guest, name = resolvedName) + } } diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/sfu/SfuSocket.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/sfu/SfuSocket.kt index afeb22012b7..23404148ff4 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/sfu/SfuSocket.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/sfu/SfuSocket.kt @@ -31,7 +31,7 @@ import io.getstream.video.android.core.events.SfuDataRequest import io.getstream.video.android.core.events.UnknownEvent import io.getstream.video.android.core.internal.network.NetworkStateProvider import io.getstream.video.android.core.lifecycle.NoOpLifecycleHandler -import io.getstream.video.android.core.lifecycle.StreamLifecycleObserver +import io.getstream.video.android.core.lifecycle.legacy.StreamLifecycleObserver import io.getstream.video.android.core.socket.common.ConnectionConf import io.getstream.video.android.core.socket.common.DISPOSE_SOCKET_REASON import io.getstream.video.android.core.socket.common.DISPOSE_SOCKET_RECONNECT diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/sfu/SfuSocketConnection.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/sfu/SfuSocketConnection.kt index 8b1316dac39..94f5295cbd6 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/sfu/SfuSocketConnection.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/sfu/SfuSocketConnection.kt @@ -23,7 +23,7 @@ import io.getstream.video.android.core.events.JoinCallResponseEvent import io.getstream.video.android.core.events.SfuDataEvent import io.getstream.video.android.core.events.SfuDataRequest import io.getstream.video.android.core.internal.network.NetworkStateProvider -import io.getstream.video.android.core.lifecycle.StreamLifecycleObserver +import io.getstream.video.android.core.lifecycle.legacy.StreamLifecycleObserver import io.getstream.video.android.core.socket.common.SfuParser import io.getstream.video.android.core.socket.common.SocketActions import io.getstream.video.android.core.socket.common.SocketFactory diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/BindableWritableUserRepositoryTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/BindableWritableUserRepositoryTest.kt new file mode 100644 index 00000000000..0f4810dbb62 --- /dev/null +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/BindableWritableUserRepositoryTest.kt @@ -0,0 +1,63 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-video-android/blob/main/LICENSE + * + * 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.getstream.video.android.core + +import io.getstream.video.android.model.User +import io.getstream.video.android.model.UserType +import org.junit.Test +import kotlin.test.assertEquals + +internal class BindableWritableUserRepositoryTest { + + private val guest = User(id = "guest-1", type = UserType.Guest) + + @Test + fun `setUser after bind delivers immediately`() { + val repo = BindableWritableUserRepository() + val delivered = mutableListOf() + repo.bind(delivered::add) + + repo.setUser(guest) + + assertEquals(listOf(guest), delivered) + } + + @Test + fun `setUser before bind is buffered and delivered on bind`() { + val repo = BindableWritableUserRepository() + val delivered = mutableListOf() + + repo.setUser(guest) + repo.bind(delivered::add) + + assertEquals(listOf(guest), delivered) + } + + @Test + fun `only the latest pre-bind user is delivered`() { + val repo = BindableWritableUserRepository() + val delivered = mutableListOf() + val stale = guest.copy(name = "stale") + val fresh = guest.copy(name = "fresh") + + repo.setUser(stale) + repo.setUser(fresh) + repo.bind(delivered::add) + + assertEquals(listOf(fresh), delivered) + } +} diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoBuilderDispatchTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoBuilderDispatchTest.kt new file mode 100644 index 00000000000..de30cef1517 --- /dev/null +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoBuilderDispatchTest.kt @@ -0,0 +1,132 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-video-android/blob/main/LICENSE + * + * 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.getstream.video.android.core + +import android.content.Context +import android.net.ConnectivityManager +import io.getstream.android.core.api.StreamClient +import io.getstream.video.android.core.socket.coordinator.v2.AnonymousStreamTokenProvider +import io.getstream.video.android.core.socket.coordinator.v2.GuestStreamTokenProvider +import io.getstream.video.android.core.socket.coordinator.v2.IntegrationStreamTokenProvider +import io.getstream.video.android.model.User +import io.getstream.video.android.model.UserType +import io.mockk.every +import io.mockk.mockk +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * Dispatch-level tests for [StreamVideoBuilder] focused on the token-provider selection wired in + * Plan 02-02 (D-05, D-06, D-07) and the `localCoordinatorAddress` -> `ws://` plumbing (COORD-08). + * + * Uses the [StreamVideoBuilder.streamClientFactory] seam to stub the [StreamClient] factory so + * the tests never call core's Android-service-dependent initialisation path (RESEARCH Pitfall 2). + */ +internal class StreamVideoBuilderDispatchTest { + + @Test + fun `Anonymous path selects AnonymousStreamTokenProvider and does not throw`() { + // iOS parity: anonymous users construct normally (REST-only, D-07); only explicit + // connect attempts fail. + val capturedProvider = captureFactoryArgsFor( + user = User(id = "anon-1", type = UserType.Anonymous), + token = "", + )?.tokenProvider + assertTrue( + capturedProvider is AnonymousStreamTokenProvider, + "Expected AnonymousStreamTokenProvider, got: ${capturedProvider?.javaClass}", + ) + } + + @Test + fun `Authenticated path selects IntegrationStreamTokenProvider`() { + val capturedProvider = captureFactoryArgsFor( + user = User(id = "auth-1", type = UserType.Authenticated), + token = "user-token", + )?.tokenProvider + assertTrue( + capturedProvider is IntegrationStreamTokenProvider, + "Expected IntegrationStreamTokenProvider, got: ${capturedProvider?.javaClass}", + ) + } + + @Test + fun `Guest path selects GuestStreamTokenProvider`() { + val capturedProvider = captureFactoryArgsFor( + user = User(id = "guest-1", type = UserType.Guest), + token = "", + )?.tokenProvider + assertTrue( + capturedProvider is GuestStreamTokenProvider, + "Expected GuestStreamTokenProvider, got: ${capturedProvider?.javaClass}", + ) + } + + @Test + fun `localCoordinatorAddress threads into resolvedWssUrl as ws`() { + val user = User(id = "auth-1", type = UserType.Authenticated) + val context = mockk(relaxed = true) + every { context.applicationContext } returns context + every { + context.getSystemService(Context.CONNECTIVITY_SERVICE) + } returns mockk(relaxed = true) + val builder = StreamVideoBuilder( + context = context, + apiKey = "apikey", + user = user, + token = "user-token", + localCoordinatorAddress = "10.0.2.2:3030", + ) + var capturedUrl: String? = null + builder.streamClientFactory = { args -> + capturedUrl = args.resolvedWssUrl + mockk(relaxed = true) + } + runCatching { builder.build() } + assertEquals("ws://10.0.2.2:3030/video/connect", capturedUrl) + } + + private fun builderFor(user: User, token: String = "token"): StreamVideoBuilder { + val context = mockk(relaxed = true) + every { context.applicationContext } returns context + every { + context.getSystemService(Context.CONNECTIVITY_SERVICE) + } returns mockk(relaxed = true) + return StreamVideoBuilder( + context = context, + apiKey = "apikey", + user = user, + token = token, + ) + } + + /** + * Substitutes the `streamClientFactory` seam so the factory args can be observed without + * invoking core's real factory. + */ + private fun captureFactoryArgsFor(user: User, token: String): StreamClientFactoryArgs? { + val builder = builderFor(user = user, token = token) + var captured: StreamClientFactoryArgs? = null + builder.streamClientFactory = { args -> + captured = args + mockk(relaxed = true) + } + runCatching { builder.build() } + return captured + } +} diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoClientTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoClientTest.kt index e44b1053d28..0c3a08dc8ae 100644 --- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoClientTest.kt +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoClientTest.kt @@ -18,10 +18,15 @@ package io.getstream.video.android.core import android.content.Context import androidx.lifecycle.Lifecycle +import io.getstream.android.core.api.StreamClient +import io.getstream.android.core.api.model.connection.StreamConnectionState +import io.getstream.android.core.api.subscribe.StreamSubscription import io.getstream.android.video.generated.models.CallAcceptedEvent import io.getstream.android.video.generated.models.CallRingEvent import io.getstream.android.video.generated.models.CallSessionStartedEvent import io.getstream.android.video.generated.models.VideoEvent +import io.getstream.result.Result.Failure +import io.getstream.result.Result.Success import io.getstream.video.android.core.call.CallBusyHandler import io.getstream.video.android.core.call.RtcSession import io.getstream.video.android.core.events.VideoEventListener @@ -30,12 +35,17 @@ import io.getstream.video.android.core.notifications.internal.StreamNotification 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.model.User +import io.getstream.video.android.model.UserType +import io.mockk.coEvery +import io.mockk.coVerify import io.mockk.every import io.mockk.mockk import io.mockk.spyk import io.mockk.unmockkAll import io.mockk.verify import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.test.runTest import org.junit.After import org.junit.Before import kotlin.test.Test @@ -45,11 +55,14 @@ import kotlin.test.assertTrue class StreamVideoClientTest { private lateinit var client: StreamVideoClient + private lateinit var streamClient: StreamClient private lateinit var state: ClientState @Before fun setup() { - client = prepareClient() + val prepared = prepareClient() + client = prepared.client + streamClient = prepared.streamClient state = mockk(relaxed = true) @@ -65,7 +78,13 @@ class StreamVideoClientTest { unmockkAll() } - private fun prepareClient(): StreamVideoClient { + private data class ClientHarness( + val client: StreamVideoClient, + val streamClient: StreamClient, + val tokenRepository: TokenRepository, + ) + + private fun prepareClient(user: User = mockk(relaxed = true)): ClientHarness { val context = mockk(relaxed = true) val lifecycle = mockk(relaxed = true) val coordinator = mockk(relaxed = true) @@ -73,15 +92,20 @@ class StreamVideoClientTest { val notificationManager = mockk(relaxed = true) val sounds = mockk(relaxed = true) val vibration = mockk(relaxed = true) + val streamClientMock = mockk(relaxed = true) + val subscription = mockk(relaxed = true) + every { streamClientMock.subscribe(any()) } returns Result.success(subscription) + every { streamClientMock.connectionState } returns MutableStateFlow(StreamConnectionState.Idle) - return spyk( + val client = spyk( StreamVideoClient( context = context, - user = mockk(relaxed = true), + user = user, apiKey = "apikey", token = "token", lifecycle = lifecycle, coordinatorConnectionModule = coordinator, + streamClient = streamClientMock, tokenRepository = tokenRepo, streamNotificationManager = notificationManager, enableCallNotificationUpdates = false, @@ -90,6 +114,7 @@ class StreamVideoClientTest { ), recordPrivateCalls = true, ) + return ClientHarness(client, streamClientMock, tokenRepo) } @Test @@ -224,7 +249,7 @@ class StreamVideoClientTest { val event = mockk(relaxed = true) every { event.callCid } returns "video:999" - val client = prepareClient() + val client = prepareClient().client val clientState = mockk(relaxed = true) client::class.java.getDeclaredField("state").apply { isAccessible = true @@ -248,7 +273,7 @@ class StreamVideoClientTest { val event = mockk(relaxed = true) every { event.callCid } returns "video:999" - val client = prepareClient() + val client = prepareClient().client val clientState = mockk(relaxed = true) client::class.java.getDeclaredField("state").apply { isAccessible = true @@ -266,4 +291,60 @@ class StreamVideoClientTest { verify(exactly = 1) { clientState.handleEvent(event) } unmockkAll() } + + @Test + fun `init subscribes to streamClient`() { + // The subscribe call is exercised inside prepareClient() via the ctor. Re-verify + // by asserting `streamClient.subscribe(...)` was invoked exactly once when the + // StreamVideoClient was constructed. + verify(exactly = 1) { streamClient.subscribe(any()) } + } + + @Test + fun `connectAsync delegates to streamClient connect`() = runTest { + coEvery { streamClient.connect() } returns Result.success(mockk(relaxed = true)) + + val result = client.connectAsync().await() + + coVerify(exactly = 1) { streamClient.connect() } + assertTrue(result is Success) + } + + @Test + fun `cleanup disconnects streamClient and does not touch tokenRepository`() { + val harness = prepareClient() + + harness.client.cleanup() + + coVerify(exactly = 1) { harness.streamClient.disconnect() } + // COORD-03 invariant: the 401 path no longer routes through tokenRepository.updateToken + // from inside cleanup(). (Cleanup itself never touched tokenRepository, but this + // assertion guards against a future regression that reintroduces the coupling.) + verify(exactly = 0) { harness.tokenRepository.updateToken(any()) } + } + + @Test + fun `connectAsync fails fast for anonymous users without touching streamClient`() = runTest { + // D-07 / iOS parity: anonymous users are REST-only; connect attempts fail before + // reaching the network. + val harness = prepareClient( + user = User(id = "anon-1", type = UserType.Anonymous), + ) + + val result = harness.client.connectAsync().await() + + assertTrue(result is Failure) + coVerify(exactly = 0) { harness.streamClient.connect() } + } + + @Test + fun `connectIfNotAlreadyConnected is a no-op for anonymous users`() = runTest { + val harness = prepareClient( + user = User(id = "anon-1", type = UserType.Anonymous), + ) + + harness.client.connectIfNotAlreadyConnected() + + coVerify(exactly = 0) { harness.streamClient.connect() } + } } diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/base/IntegrationTestBase.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/base/IntegrationTestBase.kt index 6909564ee57..723ca282280 100644 --- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/base/IntegrationTestBase.kt +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/base/IntegrationTestBase.kt @@ -111,7 +111,12 @@ open class IntegrationTestBase(val connectCoordinatorWS: Boolean = true) : TestB client = builder.build() clientImpl = client as StreamVideoClient clientImpl.testSessionId = runBlocking { - (client as StreamVideoClient).coordinatorConnectionModule.socketConnection.connectionId().value + // Derive the WS connectionId from core's StreamClient state (D-11 replaced the + // coordinator-socket connectionId() flow). Falls back to null before Connected. + ( + clientImpl.streamClient.connectionState.value + as? io.getstream.android.core.api.model.connection.StreamConnectionState.Connected + )?.connectionId } // always mock the peer connection factory, it can't work in unit tests Call.testInstanceProvider.mediaManagerCreator = { mockk(relaxed = true) } diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/socket/coordinator/v2/GuestStreamTokenProviderTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/socket/coordinator/v2/GuestStreamTokenProviderTest.kt index 4a2abb579ea..d3769bc7bc8 100644 --- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/socket/coordinator/v2/GuestStreamTokenProviderTest.kt +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/socket/coordinator/v2/GuestStreamTokenProviderTest.kt @@ -51,7 +51,7 @@ internal class GuestStreamTokenProviderTest { duration = "10ms", user = serverIssuedUser(), ) - val subject = GuestStreamTokenProvider(api, initialUser, repo) + val subject = GuestStreamTokenProvider(api, initialUser, repo, onTokenIssued = {}) val token = subject.loadToken(StreamUserId.fromString("guest-1")) @@ -66,6 +66,65 @@ internal class GuestStreamTokenProviderTest { } } + @Test + fun `loadToken publishes the issued token to onTokenIssued before returning`() = runTest { + val api = mockk() + val repo = mockk(relaxed = true) + coEvery { api.createGuest(any()) } returns CreateGuestResponse( + accessToken = "fake-jwt", + duration = "10ms", + user = serverIssuedUser(), + ) + val issuedTokens = mutableListOf() + val subject = GuestStreamTokenProvider( + api = api, + initialUser = User(id = "guest-local", type = UserType.Guest), + userRepository = repo, + onTokenIssued = issuedTokens::add, + ) + + val token = subject.loadToken(StreamUserId.fromString("guest-1")) + + // COORD-03 split-brain invariant: the REST auth path (legacy CoordinatorAuthInterceptor) + // must observe the same JWT that core's StreamTokenManager receives. + assertEquals(listOf("fake-jwt"), issuedTokens) + assertEquals("fake-jwt", token.rawValue) + } + + @Test + fun `adopted user keeps local display name when server decorates it (iOS parity)`() = runTest { + val api = mockk() + val repo = mockk(relaxed = true) + val initialUser = User(id = "guest-local", name = "Alex", type = UserType.Guest) + coEvery { api.createGuest(any()) } returns CreateGuestResponse( + accessToken = "fake-jwt", + duration = "10ms", + user = serverIssuedUser(name = "guest-abc123-Alex"), + ) + val subject = GuestStreamTokenProvider(api, initialUser, repo, onTokenIssued = {}) + + subject.loadToken(StreamUserId.fromString("guest-1")) + + verify(exactly = 1) { repo.setUser(match { it.name == "Alex" }) } + } + + @Test + fun `adopted user keeps server name when it does not derive from the local name`() = runTest { + val api = mockk() + val repo = mockk(relaxed = true) + val initialUser = User(id = "guest-local", name = "Alex", type = UserType.Guest) + coEvery { api.createGuest(any()) } returns CreateGuestResponse( + accessToken = "fake-jwt", + duration = "10ms", + user = serverIssuedUser(name = "Server Chosen Name"), + ) + val subject = GuestStreamTokenProvider(api, initialUser, repo, onTokenIssued = {}) + + subject.loadToken(StreamUserId.fromString("guest-1")) + + verify(exactly = 1) { repo.setUser(match { it.name == "Server Chosen Name" }) } + } + @Test fun `UserRequest carries initial user fields`() = runTest { val api = mockk() @@ -83,7 +142,7 @@ internal class GuestStreamTokenProviderTest { duration = "5ms", user = serverIssuedUser(), ) - val subject = GuestStreamTokenProvider(api, initialUser, repo) + val subject = GuestStreamTokenProvider(api, initialUser, repo, onTokenIssued = {}) subject.loadToken(StreamUserId.fromString("guest-1")) @@ -93,12 +152,13 @@ internal class GuestStreamTokenProviderTest { assertEquals("en_US", captured.captured.user.custom?.get("locale")) } - private fun serverIssuedUser(): UserResponse = UserResponse( + private fun serverIssuedUser(name: String? = null): UserResponse = UserResponse( createdAt = FIXED_TIME, id = "server-issued-id", language = "en", role = "user", updatedAt = FIXED_TIME, + name = name, ) private companion object { diff --git a/stream-video-android-previewdata/src/main/kotlin/io/getstream/video/android/mock/NoOpStreamClient.kt b/stream-video-android-previewdata/src/main/kotlin/io/getstream/video/android/mock/NoOpStreamClient.kt new file mode 100644 index 00000000000..5b75f6bd8c7 --- /dev/null +++ b/stream-video-android-previewdata/src/main/kotlin/io/getstream/video/android/mock/NoOpStreamClient.kt @@ -0,0 +1,51 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-video-android/blob/main/LICENSE + * + * 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.getstream.video.android.mock + +import io.getstream.android.core.api.StreamClient +import io.getstream.android.core.api.model.connection.StreamConnectionState +import io.getstream.android.core.api.socket.listeners.StreamClientListener +import io.getstream.android.core.api.subscribe.StreamSubscription +import io.getstream.android.core.api.subscribe.StreamSubscriptionManager +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow + +/** + * A no-op [StreamClient] used by preview/snapshot harnesses. Paparazzi runs under layoutlib's + * [android.content.Context] which does not implement system services like WifiManager, so the + * real client's factory would throw during initialisation. This fake side-steps that by holding + * only an [StreamConnectionState.Idle] state flow and returning inert subscriptions. + */ +internal object NoOpStreamClient : StreamClient { + + override val connectionState: StateFlow = + MutableStateFlow(StreamConnectionState.Idle) + + override fun subscribe( + listener: StreamClientListener, + options: StreamSubscriptionManager.Options, + ): Result = Result.success(NoOpSubscription) + + override suspend fun connect(): Result = + Result.failure(UnsupportedOperationException("NoOpStreamClient does not connect")) + + override suspend fun disconnect(): Result = Result.success(Unit) + + private object NoOpSubscription : StreamSubscription { + override fun cancel() = Unit + } +} diff --git a/stream-video-android-previewdata/src/main/kotlin/io/getstream/video/android/mock/StreamPreviewDataUtils.kt b/stream-video-android-previewdata/src/main/kotlin/io/getstream/video/android/mock/StreamPreviewDataUtils.kt index 0911f2fbc3c..9937be629a1 100644 --- a/stream-video-android-previewdata/src/main/kotlin/io/getstream/video/android/mock/StreamPreviewDataUtils.kt +++ b/stream-video-android-previewdata/src/main/kotlin/io/getstream/video/android/mock/StreamPreviewDataUtils.kt @@ -22,6 +22,7 @@ import io.getstream.video.android.core.MemberState import io.getstream.video.android.core.ParticipantState import io.getstream.video.android.core.StreamVideo import io.getstream.video.android.core.StreamVideoBuilder +import io.getstream.video.android.core.internal.InternalStreamVideoApi import io.getstream.video.android.core.model.MediaTrack import io.getstream.video.android.model.User import io.getstream.webrtc.VideoTrack @@ -35,6 +36,7 @@ public object StreamPreviewDataUtils { @PublishedApi internal lateinit var streamVideo: StreamVideo + @OptIn(InternalStreamVideoApi::class) public fun initializeStreamVideo(context: Context) { if (::streamVideo.isInitialized.not()) { streamVideo = StreamVideoBuilder( @@ -42,7 +44,12 @@ public object StreamPreviewDataUtils { apiKey = "stream-api-key", user = previewUsers.first(), token = "user-token", - ).build() + ) + // Paparazzi/preview harnesses run under layoutlib, which lacks the Android system + // services (WifiManager, etc.) that core's default StreamClient factory touches + // during eager construction. Swap in a no-op client so build() stays test-safe. + .streamClientFactory { NoOpStreamClient } + .build() } } }