Pass complete user data when connecting (#39)

Author: gpunto

.github/workflows/pr-quality.yml

@@ -10,6 +10,10 @@ permissions:
   pull-requests: write
   issues: write
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 jobs:
   pr-checklist:
     uses: GetStream/android-ci-actions/.github/workflows/pr-quality.yml@main

app/src/main/java/io/getstream/android/core/sample/SampleApp.kt

@@ -22,6 +22,7 @@ import android.os.Build
 import android.os.StrictMode
 import io.getstream.android.core.api.StreamClient
 import io.getstream.android.core.api.authentication.StreamTokenProvider
+import io.getstream.android.core.api.model.StreamUser
 import io.getstream.android.core.api.model.config.StreamClientSerializationConfig
 import io.getstream.android.core.api.model.value.StreamApiKey
 import io.getstream.android.core.api.model.value.StreamHttpClientInfoHeader
@@ -36,7 +37,7 @@ import kotlinx.coroutines.SupervisorJob
 class SampleApp : Application() {
 
     lateinit var streamClient: StreamClient
-    private val userId = StreamUserId.fromString("sample-user")
+    private val user = StreamUser(StreamUserId.fromString("sample-user"))
     private val token =
         StreamToken.fromString(
             "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoicGV0YXIifQ.mZFi4iSblaIoyo9JDdcxIkGkwI-tuApeSBawxpz42rs"
@@ -60,7 +61,7 @@ class SampleApp : Application() {
                 context = this.applicationContext,
                 scope = coroutinesScope,
                 apiKey = StreamApiKey.fromString("pd67s34fzpgw"),
-                userId = userId,
+                user = user,
                 products = listOf("feeds", "chat", "video"),
                 wsUrl =
                     StreamWsUrl.fromString(

stream-android-core/src/main/java/io/getstream/android/core/api/StreamClient.kt

@@ -24,6 +24,7 @@ import io.getstream.android.core.api.authentication.StreamTokenProvider
 import io.getstream.android.core.api.components.StreamAndroidComponentsProvider
 import io.getstream.android.core.api.http.StreamOkHttpInterceptors
 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.StreamHttpConfig
 import io.getstream.android.core.api.model.config.StreamSocketConfig
@@ -33,7 +34,6 @@ import io.getstream.android.core.api.model.connection.lifecycle.StreamLifecycleS
 import io.getstream.android.core.api.model.connection.network.StreamNetworkState
 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.observers.network.StreamNetworkMonitor
@@ -145,9 +145,9 @@ public interface StreamClient : StreamObservable<StreamClientListener> {
 /**
  * ### Overview
  *
- * Creates a [StreamClient] with the given [apiKey], [userId], [tokenProvider] and [scope]. The
- * client is created in a disconnected state. You must call `connect()` to establish a connection.
- * The client is automatically disconnected when the [scope] is cancelled.
+ * Creates a [StreamClient] with the given [apiKey], [user], [tokenProvider] and [scope]. The client
+ * is created in a disconnected state. You must call `connect()` to establish a connection. The
+ * client is automatically disconnected when the [scope] is cancelled.
  *
  * **Important**: The client instance **must be kept alive for the duration of the connection**. Do
  * not create a new client for every operation.
@@ -185,7 +185,7 @@ public interface StreamClient : StreamObservable<StreamClientListener> {
  * ```
  *
  * @param apiKey The API key.
- * @param userId The user ID.
+ * @param user The user ID.
  * @param wsUrl The WebSocket URL.
  * @param products Stream product codes (for feature gates / telemetry) negotiated with the socket.
  * @param clientInfoHeader The client info header.
@@ -215,7 +215,7 @@ public fun StreamClient(
 
     // Client config
     apiKey: StreamApiKey,
-    userId: StreamUserId,
+    user: StreamUser,
     wsUrl: StreamWsUrl,
     products: List<String>,
     clientInfoHeader: StreamHttpClientInfoHeader,
@@ -249,7 +249,7 @@ public fun StreamClient(
         StreamRetryProcessor(logger = logProvider.taggedLogger("SCRetryProcessor")),
 
     // Token
-    tokenManager: StreamTokenManager = StreamTokenManager(userId, tokenProvider, singleFlight),
+    tokenManager: StreamTokenManager = StreamTokenManager(user.id, tokenProvider, singleFlight),
 
     // Socket
     connectionIdHolder: StreamConnectionIdHolder = StreamConnectionIdHolder(),
@@ -344,7 +344,7 @@ public fun StreamClient(
 
     val mutableConnectionState = MutableStateFlow<StreamConnectionState>(StreamConnectionState.Idle)
     return StreamClientImpl(
-        userId = userId,
+        user = user,
         scope = clientScope,
         tokenManager = tokenManager,
         singleFlight = singleFlight,

stream-android-core/src/main/java/io/getstream/android/core/api/model/StreamUser.kt

@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) 2014-2025 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-core-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.android.core.api.model
+
+import io.getstream.android.core.annotations.StreamInternalApi
+import io.getstream.android.core.api.model.value.StreamUserId
+
+/**
+ * Represents a user in the Stream system.
+ *
+ * @property id The unique identifier for the user.
+ * @property name The name of the user (optional).
+ * @property imageURL The URL of the user's image (optional).
+ * @property customData Custom data associated with the user, represented as a map (default empty
+ *   map).
+ */
+@StreamInternalApi
+public data class StreamUser(
+    public val id: StreamUserId,
+    public val name: String? = null,
+    public val imageURL: String? = null,
+    public val customData: Map<String, Any> = emptyMap(),
+)

stream-android-core/src/main/java/io/getstream/android/core/internal/client/StreamClientImpl.kt

@@ -20,13 +20,13 @@ import io.getstream.android.core.api.StreamClient
 import io.getstream.android.core.api.authentication.StreamTokenManager
 import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.model.StreamTypedKey.Companion.randomExecutionKey
+import io.getstream.android.core.api.model.StreamUser
 import io.getstream.android.core.api.model.connection.StreamConnectedUser
 import io.getstream.android.core.api.model.connection.StreamConnectionState
 import io.getstream.android.core.api.model.connection.lifecycle.StreamLifecycleState
 import io.getstream.android.core.api.model.connection.network.StreamNetworkState
 import io.getstream.android.core.api.model.connection.recovery.Recovery
 import io.getstream.android.core.api.model.value.StreamToken
-import io.getstream.android.core.api.model.value.StreamUserId
 import io.getstream.android.core.api.processing.StreamSerialProcessingQueue
 import io.getstream.android.core.api.processing.StreamSingleFlightProcessor
 import io.getstream.android.core.api.recovery.StreamConnectionRecoveryEvaluator
@@ -49,7 +49,7 @@ import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.launch
 
 internal class StreamClientImpl<T>(
-    private val userId: StreamUserId,
+    private val user: StreamUser,
     private val tokenManager: StreamTokenManager,
     private val singleFlight: StreamSingleFlightProcessor,
     private val serialQueue: StreamSerialProcessingQueue,
@@ -182,13 +182,13 @@ internal class StreamClientImpl<T>(
     ): Result<StreamConnectionState.Connected> {
         val data =
             ConnectUserData(
-                userId = userId.rawValue,
+                userId = user.id.rawValue,
                 token = token.rawValue,
-                name = null,
-                image = null,
+                name = user.name,
+                image = user.imageURL,
                 invisible = false,
                 language = null,
-                custom = null,
+                custom = user.customData,
             )
         return socketSession.connect(data).onTokenError { error, code ->
             logger.e(error) { "Token error: $code" }

stream-android-core/src/test/java/io/getstream/android/core/api/StreamClientFactoryTest.kt

@@ -25,6 +25,7 @@ import io.getstream.android.core.api.authentication.StreamTokenManager
 import io.getstream.android.core.api.authentication.StreamTokenProvider
 import io.getstream.android.core.api.log.StreamLogger
 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.StreamHttpConfig
 import io.getstream.android.core.api.model.config.StreamSocketConfig
@@ -92,7 +93,7 @@ internal class StreamClientFactoryTest {
 
     private data class Dependencies(
         val apiKey: StreamApiKey,
-        val userId: StreamUserId,
+        val user: StreamUser,
         val wsUrl: StreamWsUrl,
         val clientInfo: StreamHttpClientInfoHeader,
         val clientSubscriptionManager: StreamSubscriptionManager<StreamClientListener>,
@@ -116,7 +117,7 @@ internal class StreamClientFactoryTest {
         val deps =
             Dependencies(
                 apiKey = StreamApiKey.fromString("key123"),
-                userId = StreamUserId.fromString("user-123"),
+                user = StreamUser(id = StreamUserId.fromString("user-123")),
                 wsUrl = StreamWsUrl.fromString("wss://test.stream/video"),
                 clientInfo =
                     StreamHttpClientInfoHeader.create(
@@ -147,7 +148,7 @@ internal class StreamClientFactoryTest {
             StreamClient(
                 context = mockk(relaxed = true),
                 apiKey = deps.apiKey,
-                userId = deps.userId,
+                user = deps.user,
                 wsUrl = deps.wsUrl,
                 products = listOf("feeds"),
                 clientInfoHeader = deps.clientInfo,
@@ -185,7 +186,7 @@ internal class StreamClientFactoryTest {
         assertTrue(client.connectionState.value is StreamConnectionState.Idle)
 
         // Verify client level wiring
-        client.assertFieldEquals("userId", deps.userId.rawValue)
+        client.assertFieldEquals("user", deps.user)
         client.assertFieldEquals("tokenManager", deps.tokenManager)
         client.assertFieldEquals("singleFlight", deps.singleFlight)
         client.assertFieldEquals("serialQueue", deps.serialQueue)
@@ -323,12 +324,19 @@ internal class StreamClientFactoryTest {
                     StreamToken.fromString("token")
             }
 
+        val user =
+            StreamUser(
+                id = StreamUserId.fromString("user-123"),
+                name = "name",
+                imageURL = "image",
+                customData = mapOf("custom" to "data"),
+            )
         val client =
             StreamClient(
                 scope = testScope,
                 context = context,
                 apiKey = StreamApiKey.fromString("key123"),
-                userId = StreamUserId.fromString("user-123"),
+                user = user,
                 wsUrl = StreamWsUrl.fromString("wss://test.stream/video"),
                 products = listOf("feeds"),
                 clientInfoHeader =

stream-android-core/src/test/java/io/getstream/android/core/internal/client/StreamClientIImplTest.kt

@@ -21,6 +21,7 @@ package io.getstream.android.core.internal.client
 import io.getstream.android.core.api.authentication.StreamTokenManager
 import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.model.StreamTypedKey
+import io.getstream.android.core.api.model.StreamUser
 import io.getstream.android.core.api.model.connection.StreamConnectedUser
 import io.getstream.android.core.api.model.connection.StreamConnectionState
 import io.getstream.android.core.api.model.connection.lifecycle.StreamLifecycleState
@@ -48,8 +49,18 @@ import io.getstream.android.core.internal.observers.StreamNetworkAndLifeCycleMon
 import io.getstream.android.core.internal.observers.StreamNetworkAndLifecycleMonitorListener
 import io.getstream.android.core.internal.recovery.StreamConnectionRecoveryEvaluatorImpl
 import io.getstream.android.core.internal.socket.StreamSocketSession
+import io.getstream.android.core.internal.socket.model.ConnectUserData
 import io.getstream.android.core.testing.TestLogger
-import io.mockk.*
+import io.mockk.Awaits
+import io.mockk.MockKAnnotations
+import io.mockk.coEvery
+import io.mockk.coJustRun
+import io.mockk.coVerify
+import io.mockk.every
+import io.mockk.just
+import io.mockk.justRun
+import io.mockk.mockk
+import io.mockk.verify
 import kotlin.time.ExperimentalTime
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
@@ -61,14 +72,23 @@ import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlinx.coroutines.test.advanceTimeBy
 import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.runTest
-import org.junit.Assert.*
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertSame
+import org.junit.Assert.assertTrue
 import org.junit.Before
 import org.junit.Test
 
 @OptIn(ExperimentalTime::class)
 class StreamClientIImplTest {
 
-    private var userId: StreamUserId = StreamUserId.fromString("u1")
+    private var user: StreamUser =
+        StreamUser(
+            id = StreamUserId.fromString("u1"),
+            name = "name",
+            imageURL = "image",
+            customData = mapOf("custom" to "data"),
+        )
     private lateinit var tokenManager: StreamTokenManager
     private lateinit var singleFlight: StreamSingleFlightProcessor
     private lateinit var serialQueue: StreamSerialProcessingQueue
@@ -122,7 +142,7 @@ class StreamClientIImplTest {
         connectionRecoveryEvaluator: StreamConnectionRecoveryEvaluator = mockk(relaxed = true),
     ) =
         StreamClientImpl(
-            userId = userId,
+            user = user,
             tokenManager = tokenManager,
             singleFlight = singleFlight,
             serialQueue = serialQueue,
@@ -372,6 +392,14 @@ class StreamClientIImplTest {
             coEvery { socketSession.connect(any()) } returns Result.success(connectedState)
 
             every { connectionIdHolder.setConnectionId("conn-1") } returns Result.success("conn-1")
+            val expectedConnectUserData =
+                ConnectUserData(
+                    userId = user.id.rawValue,
+                    token = token.rawValue,
+                    name = user.name,
+                    image = user.imageURL,
+                    custom = user.customData,
+                )
 
             val result = client.connect()
 
@@ -385,7 +413,7 @@ class StreamClientIImplTest {
             // interactions
             verify(exactly = 1) { socketSession.subscribe(any<StreamClientListener>(), any()) }
             coVerify(exactly = 1) { tokenManager.loadIfAbsent() }
-            coVerify(exactly = 1) { socketSession.connect(match { it.token == "tok" }) }
+            coVerify(exactly = 1) { socketSession.connect(expectedConnectUserData) }
             verify(exactly = 1) { connectionIdHolder.setConnectionId("conn-1") }
         }
 
@@ -629,7 +657,7 @@ class StreamClientIImplTest {
             {
                 if (firstCall) {
                     firstCall = false
-                    connFlow.update { StreamConnectionState.Connecting.Opening(userId.rawValue) }
+                    connFlow.update { StreamConnectionState.Connecting.Opening(user.id.rawValue) }
                     firstConnectDeferred.await()
                 } else {
                     Result.success(connectedState)