From 8c962cac7ad413e5674bf300384c2387a14353e5 Mon Sep 17 00:00:00 2001 From: Gian <47775302+gpunto@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:50:01 +0200 Subject: [PATCH] Connect the websocket through the v2 endpoint behind an internal flag --- .../chat/android/client/ChatClient.kt | 14 ++ .../chat/android/client/api/ChatApiConfig.kt | 2 + .../chat/android/client/di/ChatModule.kt | 10 +- .../chat/android/client/socket/ChatSocket.kt | 4 +- .../android/client/socket/SocketFactory.kt | 70 ++++++- .../android/client/socket/StreamWebSocket.kt | 30 ++- .../android/client/socket/V2SocketProtocol.kt | 41 ++++ .../android/client/socket/V2SocketTest.kt | 193 ++++++++++++++++++ .../client/user/CurrentUserFetcherTests.kt | 66 +++++- 9 files changed, 413 insertions(+), 17 deletions(-) create mode 100644 stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/V2SocketProtocol.kt create mode 100644 stream-chat-android-client/src/test/java/io/getstream/chat/android/client/socket/V2SocketTest.kt diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/ChatClient.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/ChatClient.kt index 2dab38a938d5..34c480670a7b 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/ChatClient.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/ChatClient.kt @@ -5153,6 +5153,7 @@ internal constructor( private var forceInsecureConnection = false private var forceHttpUrl: String? = null private var forceWsUrl: String? = null + private var useV2WebSocket: Boolean = false private var baseUrl: String = "chat.stream-io-api.com" private var cdnUrl: String? = null private var logLevel = ChatLogLevel.NOTHING @@ -5360,6 +5361,18 @@ internal constructor( forceWsUrl = value } + /** + * Connects the websocket through the v2 endpoint instead of v1. + * + * This for internal use only, until the v2 websocket becomes the default. + * + * @param enabled Whether to use the v2 websocket. + */ + @InternalStreamChatApi + public fun useV2WebSocket(enabled: Boolean): Builder = apply { + useV2WebSocket = enabled + } + /** * Sets a custom [CDN] implementation to be used by the client. * @@ -5486,6 +5499,7 @@ internal constructor( debugRequests = debugRequests, notificationConfig = notificationConfig, fastEventParsing = chatClientConfig.fastEventParsing, + useV2WebSocket = useV2WebSocket, ) setupStreamLog() diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/ChatApiConfig.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/ChatApiConfig.kt index bc8e31ef00ec..514e249116c0 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/ChatApiConfig.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/ChatApiConfig.kt @@ -38,6 +38,7 @@ import io.getstream.chat.android.client.notifications.handler.NotificationConfig * When `true`, supported event types are parsed directly into domain models, bypassing the DTO * intermediate layer; unsupported event types fall back to the default DTO-based parser. * Currently supported event types: `message.new`. Disabled by default. + * @param useV2WebSocket Connects the websocket through the v2 endpoint instead of v1. Disabled by default. */ @Suppress("LongParameterList") internal class ChatApiConfig @JvmOverloads constructor( @@ -51,6 +52,7 @@ internal class ChatApiConfig @JvmOverloads constructor( val debugRequests: Boolean, val notificationConfig: NotificationConfig, val fastEventParsing: Boolean = false, + val useV2WebSocket: Boolean = false, ) { var isAnonymous: Boolean = false } diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/di/ChatModule.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/di/ChatModule.kt index 8d58620f4c0b..5f4ea302ba14 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/di/ChatModule.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/di/ChatModule.kt @@ -78,6 +78,7 @@ import io.getstream.chat.android.client.notifications.handler.NotificationHandle import io.getstream.chat.android.client.parser.ChatParser import io.getstream.chat.android.client.parser2.DirectEventParser import io.getstream.chat.android.client.parser2.MoshiChatParser +import io.getstream.chat.android.client.parser2.V2EventParser import io.getstream.chat.android.client.parser2.withErrorLogging import io.getstream.chat.android.client.plugins.requests.ApiRequestsAnalyser import io.getstream.chat.android.client.scope.ClientScope @@ -200,7 +201,14 @@ constructor( directEventParser = directEventParser, ) } - private val socketFactory: SocketFactory by lazy { SocketFactory(moshiParser, tokenManager, headersUtil) } + private val socketFactory: SocketFactory by lazy { + SocketFactory( + parser = moshiParser, + tokenManager = tokenManager, + headersUtil = headersUtil, + v2EventParser = if (config.useV2WebSocket) V2EventParser(v2DomainMapping) else null, + ) + } private val defaultNotifications by lazy { buildChatNotifications(notificationsHandler, config.notificationConfig) } private val defaultApi by lazy { buildApi(config) } diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/ChatSocket.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/ChatSocket.kt index 5f6494b66e50..87d24648a244 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/ChatSocket.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/ChatSocket.kt @@ -65,7 +65,7 @@ internal open class ChatSocket( userScope = userScope, checkCallback = { (chatSocketStateService.currentState as? State.Connected)?.event?.let { - if (sendEvent(it)) { + if (streamWebSocket?.sendKeepAlive(it) == true) { serverClockOffset.onHealthCheckSent() } } @@ -275,8 +275,6 @@ internal open class ChatSocket( * * @see [okhttp3.WebSocket.send] */ - internal fun sendEvent(event: ChatEvent): Boolean = streamWebSocket?.send(event) ?: false - internal fun isConnected(): Boolean = chatSocketStateService.currentState is State.Connected /** diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/SocketFactory.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/SocketFactory.kt index f64c28d6644f..7d212eacb26a 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/SocketFactory.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/SocketFactory.kt @@ -19,6 +19,7 @@ package io.getstream.chat.android.client.socket import io.getstream.chat.android.PrivacySettings import io.getstream.chat.android.client.api2.mapping.USER_DEVICES_CUSTOM_KEY import io.getstream.chat.android.client.parser.ChatParser +import io.getstream.chat.android.client.parser2.V2EventParser import io.getstream.chat.android.client.token.TokenManager import io.getstream.chat.android.client.utils.HeadersUtil import io.getstream.chat.android.models.User @@ -29,21 +30,86 @@ import java.io.UnsupportedEncodingException import java.net.URLEncoder import java.nio.charset.StandardCharsets +/** + * @param v2EventParser Connects through the v2 endpoint and parses its events with it, or null for v1. + */ internal class SocketFactory( private val parser: ChatParser, private val tokenManager: TokenManager, private val headersUtil: HeadersUtil, private val httpClient: OkHttpClient = OkHttpClient(), + private val v2EventParser: V2EventParser? = null, ) { private val logger by taggedLogger("Chat:SocketFactory") @Throws(UnsupportedEncodingException::class) fun createSocket(connectionConf: ConnectionConf): StreamWebSocket { + if (v2EventParser != null) return createV2Socket(connectionConf, v2EventParser) val request = buildRequest(connectionConf) logger.i { "new web socket: ${request.url}" } return StreamWebSocket(parser) { httpClient.newWebSocket(request, it) } } + /** + * The v2 socket authenticates with a frame sent once open, which must arrive within the server's handshake + * deadline, so the token is resolved before the upgrade. The request carries no Authorization header: the server + * prefers it over the frame's token. + */ + private fun createV2Socket(connectionConf: ConnectionConf, eventParser: V2EventParser): StreamWebSocket { + val authType = when (connectionConf) { + is ConnectionConf.AnonymousConnectionConf -> "anonymous" + is ConnectionConf.UserConnectionConf -> "jwt" + } + val request = Request.Builder() + .url( + "${connectionConf.endpoint}api/v2/connect?api_key=${connectionConf.apiKey}" + + "&stream-auth-type=$authType" + + "&$X_STREAM_CLIENT=${encodedSdkTrackingHeaders()}", + ) + .addHeader(X_STREAM_CLIENT, headersUtil.buildSdkTrackingHeaders()) + .build() + logger.i { "new v2 web socket: ${request.url}" } + val protocol = V2SocketProtocol(buildAuthFrame(connectionConf), eventParser) + return StreamWebSocket(parser, protocol) { httpClient.newWebSocket(request, it) } + } + + private fun encodedSdkTrackingHeaders(): String = + URLEncoder.encode(headersUtil.buildSdkTrackingHeaders(), StandardCharsets.UTF_8.name()) + + private fun buildAuthFrame(connectionConf: ConnectionConf): String { + val token = when (connectionConf) { + // iOS sends this placeholder until anonymous connections with an empty token are deployed everywhere. + is ConnectionConf.AnonymousConnectionConf -> ANONYMOUS_TOKEN + is ConnectionConf.UserConnectionConf -> { + tokenManager.ensureTokenLoaded() + tokenManager.getToken() + } + } + val frame = mapOf( + "token" to token, + "user_details" to connectionConf.reduceV2UserDetails(), + "products" to listOf("chat"), + ) + return V2SocketProtocol.frameAdapter.toJson(frame) + } + + /** + * The user details the v2 connect reads, with custom data nested under `custom`; it ignores any other field. + * Reconnections send the id only, as on v1. + */ + private fun ConnectionConf.reduceV2UserDetails(): Map = mutableMapOf("id" to id) + .apply { + if (!isReconnection) { + if (user.name.isNotBlank()) put("name", user.name) + if (user.image.isNotBlank()) put("image", user.image) + if (user.language.isNotBlank()) put("language", user.language) + user.invisible?.also { put("invisible", it) } + user.privacySettings?.also { put("privacy_settings", it.reducePrivacySettings()) } + val custom = user.extraData - USER_DEVICES_CUSTOM_KEY + if (custom.isNotEmpty()) put("custom", custom) + } + } + @Throws(UnsupportedEncodingException::class) private fun buildRequest(connectionConf: ConnectionConf): Request = Request.Builder() @@ -57,8 +123,7 @@ internal class SocketFactory( var json = buildUserDetailJson(connectionConf) return try { json = URLEncoder.encode(json, StandardCharsets.UTF_8.name()) - val sdkTrackingHeaders = - URLEncoder.encode(headersUtil.buildSdkTrackingHeaders(), StandardCharsets.UTF_8.name()) + val sdkTrackingHeaders = encodedSdkTrackingHeaders() val baseWsUrl = "${connectionConf.endpoint}connect?" + "json=$json" + "&api_key=${connectionConf.apiKey}" + @@ -149,5 +214,6 @@ internal class SocketFactory( companion object { private const val X_STREAM_CLIENT = "X-Stream-Client" + private const val ANONYMOUS_TOKEN = "anon" } } diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/StreamWebSocket.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/StreamWebSocket.kt index 9c843b0d98f1..eb0c40182837 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/StreamWebSocket.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/StreamWebSocket.kt @@ -37,8 +37,12 @@ private const val EVENTS_BUFFER_SIZE = 100 private const val CLOSE_SOCKET_CODE = 1000 private const val CLOSE_SOCKET_REASON = "Connection close by client" +/** + * @param v2 What the v2 socket does differently, or null for the v1 socket. + */ internal class StreamWebSocket( private val parser: ChatParser, + private val v2: V2SocketProtocol? = null, socketCreator: (WebSocketListener) -> WebSocket, ) { private val logger by taggedLogger("Chat:StreamWebSocket") @@ -48,9 +52,17 @@ internal class StreamWebSocket( private var connected = false private val webSocket = socketCreator(object : WebSocketListener() { + override fun onOpen(webSocket: WebSocket, response: Response) { + v2?.let { webSocket.send(it.authFrame) } + } + override fun onMessage(webSocket: WebSocket, text: String) { StreamLog.v("Chat:Events") { "[handleEvent] event: `$text`" } - parseMessage(text)?.let(eventFlow::tryEmit) + val event = when (v2) { + null -> parseMessage(text) + else -> parseV2Message(text, v2) + } + event?.let(eventFlow::tryEmit) } override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) { @@ -79,6 +91,10 @@ internal class StreamWebSocket( }) fun send(chatEvent: ChatEvent): Boolean = webSocket.send(parser.toJson(chatEvent)) + + /** Keeps the connection [connectedEvent] established alive: v1 echoes the event, v2 sends a health check. */ + fun sendKeepAlive(connectedEvent: ConnectedEvent): Boolean = + v2?.let { webSocket.send(it.keepAliveFrame(connectedEvent)) } ?: send(connectedEvent) fun close(): Boolean = webSocket.close(CLOSE_SOCKET_CODE, CLOSE_SOCKET_REASON) fun listen(): Flow = eventFlow.asSharedFlow() @@ -96,6 +112,18 @@ internal class StreamWebSocket( is Result.Failure -> parseFailure(text, result.value) } + /** Null for a frame the v2 parser ignores, or one dropped as [parseFailure] describes. */ + @Suppress("TooGenericExceptionCaught") + private fun parseV2Message(text: String, v2: V2SocketProtocol): StreamWebSocketEvent? = try { + v2.eventParser.parse(text)?.let { event -> + if (event is ConnectedEvent) connected = true + StreamWebSocketEvent.Message(event) + } + } catch (expected: Throwable) { + // Throwable, as v1's parser catches: an error escaping here would reach OkHttp's reader thread. + parseFailure(text, Error.ThrowableError("Failed to parse the v2 event", expected)) + } + private fun parseFailure(text: String, parseChatError: Error): StreamWebSocketEvent? { val errorResponse = when (val chatErrorResult = parser.fromJsonOrError(text, SocketErrorMessage::class.java)) { is Result.Success -> chatErrorResult.value.error diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/V2SocketProtocol.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/V2SocketProtocol.kt new file mode 100644 index 000000000000..11d959fc4f52 --- /dev/null +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/V2SocketProtocol.kt @@ -0,0 +1,41 @@ +/* + * 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-chat-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.chat.android.client.socket + +import io.getstream.chat.android.client.events.ConnectedEvent +import io.getstream.chat.android.client.parser2.V2EventParser +import io.getstream.chat.android.models.EventType +import io.getstream.chat.android.network.infrastructure.Serializer + +/** + * What the v2 socket does differently from v1: it authenticates with [authFrame] once open, parses events with + * [eventParser], and keeps the connection alive with a minimal health check, which the server answers whatever the + * frame carries. + */ +internal class V2SocketProtocol( + val authFrame: String, + val eventParser: V2EventParser, +) { + /** The keepalive frame for the connection [connectedEvent] established. */ + fun keepAliveFrame(connectedEvent: ConnectedEvent): String = frameAdapter.toJson( + listOf(mapOf("type" to EventType.HEALTH_CHECK, "client_id" to connectedEvent.connectionId)), + ) + + internal companion object { + internal val frameAdapter = Serializer.moshi.adapter(Any::class.java) + } +} diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/socket/V2SocketTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/socket/V2SocketTest.kt new file mode 100644 index 000000000000..0ae8c5010a97 --- /dev/null +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/socket/V2SocketTest.kt @@ -0,0 +1,193 @@ +/* + * 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-chat-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.chat.android.client.socket + +import io.getstream.chat.android.PrivacySettings +import io.getstream.chat.android.TypingIndicators +import io.getstream.chat.android.client.api2.mapping.DomainMapping +import io.getstream.chat.android.client.events.ConnectedEvent +import io.getstream.chat.android.client.events.HealthEvent +import io.getstream.chat.android.client.parser2.ParserFactory +import io.getstream.chat.android.client.parser2.V2EventParser +import io.getstream.chat.android.client.token.FakeTokenManager +import io.getstream.chat.android.client.utils.HeadersUtil +import io.getstream.chat.android.models.NoOpChannelTransformer +import io.getstream.chat.android.models.NoOpMessageTransformer +import io.getstream.chat.android.models.NoOpUserTransformer +import io.getstream.chat.android.models.User +import io.kotest.assertions.json.shouldEqualJson +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.UnconfinedTestDispatcher +import kotlinx.coroutines.test.runTest +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import org.amshove.kluent.shouldBeEqualTo +import org.amshove.kluent.shouldBeInstanceOf +import org.amshove.kluent.shouldBeNull +import org.junit.jupiter.api.Test +import org.mockito.kotlin.any +import org.mockito.kotlin.argumentCaptor +import org.mockito.kotlin.doReturn +import org.mockito.kotlin.mock +import org.mockito.kotlin.never +import org.mockito.kotlin.verify +import org.mockito.kotlin.whenever +import java.util.Date + +/** The v2 websocket: its connect request, the auth frame it sends once open, its keepalive and its event parsing. */ +internal class V2SocketTest { + + private val webSocket: WebSocket = mock() + private val httpClient: OkHttpClient = mock().apply { + whenever(newWebSocket(any(), any())) doReturn webSocket + } + private val headersUtil: HeadersUtil = mock().apply { + whenever(buildSdkTrackingHeaders()) doReturn "stream-chat-android-v1.0" + } + private val socketFactory = SocketFactory( + parser = ParserFactory.createMoshiChatParser(), + tokenManager = FakeTokenManager(TOKEN), + headersUtil = headersUtil, + httpClient = httpClient, + v2EventParser = V2EventParser( + DomainMapping( + currentUserIdProvider = { USER.id }, + channelTransformer = NoOpChannelTransformer, + messageTransformer = NoOpMessageTransformer, + userTransformer = NoOpUserTransformer, + hidesStoredDeclaredCustom = true, + ), + ), + ) + + /** Creates the socket, opens it and returns the request and the listener it was created with. */ + private fun open(conf: SocketFactory.ConnectionConf): Pair { + socketFactory.createSocket(conf) + val request = argumentCaptor() + val listener = argumentCaptor() + verify(httpClient).newWebSocket(request.capture(), listener.capture()) + listener.firstValue.onOpen(webSocket, mock()) + return request.firstValue to listener.firstValue + } + + private fun sentFrame(): String = argumentCaptor().also { verify(webSocket).send(it.capture()) }.firstValue + + @Test + fun `A user connects through the v2 endpoint without an Authorization header`() { + val (request, _) = open(SocketFactory.ConnectionConf.UserConnectionConf(ENDPOINT, API_KEY, USER)) + + // OkHttp keeps a wss URL as https. + request.url.toString() shouldBeEqualTo "https://chat.example.com/api/v2/connect?api_key=$API_KEY" + + "&stream-auth-type=jwt&X-Stream-Client=stream-chat-android-v1.0" + request.header("X-Stream-Client") shouldBeEqualTo "stream-chat-android-v1.0" + request.header("Authorization").shouldBeNull() + } + + @Test + fun `The first connection authenticates with the user details v2 reads, custom data nested`() { + open(SocketFactory.ConnectionConf.UserConnectionConf(ENDPOINT, API_KEY, USER)) + + sentFrame() shouldEqualJson """{"token":"$TOKEN","products":["chat"],"user_details":{"id":"filip", + "name":"Chewbacca","image":"https://example.com/chewie.png","language":"en","invisible":true, + "privacy_settings":{"typing_indicators":{"enabled":false}},"custom":{"flair":"gold"}}}""" + } + + @Test + fun `A reconnection authenticates with the user id only`() { + open(SocketFactory.ConnectionConf.UserConnectionConf(ENDPOINT, API_KEY, USER).asReconnectionConf()) + + sentFrame() shouldEqualJson """{"token":"$TOKEN","products":["chat"],"user_details":{"id":"filip"}}""" + } + + @Test + fun `An anonymous user connects with the anonymous auth type and placeholder token`() { + val user = User(id = "!anon") + val (request, _) = open(SocketFactory.ConnectionConf.AnonymousConnectionConf(ENDPOINT, API_KEY, user)) + + request.url.queryParameter("stream-auth-type") shouldBeEqualTo "anonymous" + sentFrame() shouldEqualJson """{"token":"anon","products":["chat"],"user_details":{"id":"anon"}}""" + } + + @Test + fun `The keepalive is a minimal health check for the connection`() { + val socket = socketFactory.createSocket( + SocketFactory.ConnectionConf.UserConnectionConf(ENDPOINT, API_KEY, USER).asReconnectionConf(), + ) + val connected = ConnectedEvent("health.check", Date(), "", USER, "conn-1") + + socket.sendKeepAlive(connected) + + sentFrame() shouldEqualJson """[{"type":"health.check","client_id":"conn-1"}]""" + } + + @Test + fun `Without the v2 parser the socket stays v1, sending nothing once open and echoing the connection`() { + val parser = ParserFactory.createMoshiChatParser() + val v1Factory = SocketFactory(parser, FakeTokenManager(TOKEN), headersUtil, httpClient) + val socket = v1Factory.createSocket(SocketFactory.ConnectionConf.UserConnectionConf(ENDPOINT, API_KEY, USER)) + val listener = argumentCaptor().also { verify(httpClient).newWebSocket(any(), it.capture()) } + listener.firstValue.onOpen(webSocket, mock()) + verify(webSocket, never()).send(any()) + val connected = ConnectedEvent("health.check", Date(), "", USER, "conn-1") + + socket.sendKeepAlive(connected) + + sentFrame() shouldBeEqualTo parser.toJson(connected) + } + + @Test + fun `Frames parse with the v2 parser, ignored ones emit nothing`() = runTest(UnconfinedTestDispatcher()) { + val socket = socketFactory.createSocket( + SocketFactory.ConnectionConf.UserConnectionConf(ENDPOINT, API_KEY, USER).asReconnectionConf(), + ) + val listener = argumentCaptor().also { verify(httpClient).newWebSocket(any(), it.capture()) } + val events = mutableListOf() + val job = launch { socket.listen().toList(events) } + + listener.firstValue.onMessage(webSocket, """{"type":"feeds.feed.created","created_at":1791976405614650759}""") + listener.firstValue.onMessage( + webSocket, + """{"type":"health.check","connection_id":"conn-1","cid":"*","created_at":1791976405614650759,"custom":{}}""", + ) + listener.firstValue.onMessage(webSocket, """{"type":"message.new"}""") + job.cancel() + + events.size shouldBeEqualTo 2 + (events[0] as StreamWebSocketEvent.Message).chatEvent.shouldBeInstanceOf() + events[1].shouldBeInstanceOf() + } + + private companion object { + private const val ENDPOINT = "wss://chat.example.com/" + private const val API_KEY = "key" + private const val TOKEN = "token" + private val USER = User( + id = "filip", + role = "user", + name = "Chewbacca", + image = "https://example.com/chewie.png", + language = "en", + invisible = true, + privacySettings = PrivacySettings(typingIndicators = TypingIndicators(enabled = false)), + extraData = mapOf("flair" to "gold", "devices" to listOf(mapOf("id" to "tok"))), + ) + } +} diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/user/CurrentUserFetcherTests.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/user/CurrentUserFetcherTests.kt index 6cce41d79a3c..039633697c9f 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/user/CurrentUserFetcherTests.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/user/CurrentUserFetcherTests.kt @@ -18,17 +18,24 @@ package io.getstream.chat.android.client.user import io.getstream.chat.android.client.Mother.randomConnectedEvent import io.getstream.chat.android.client.api.ChatApiConfig +import io.getstream.chat.android.client.api2.mapping.DomainMapping import io.getstream.chat.android.client.errors.ChatErrorCode import io.getstream.chat.android.client.events.ChatEvent import io.getstream.chat.android.client.network.NetworkStateProvider import io.getstream.chat.android.client.notifications.handler.NotificationConfig import io.getstream.chat.android.client.parser.ChatParser +import io.getstream.chat.android.client.parser2.ParserFactory +import io.getstream.chat.android.client.parser2.V2EventParser import io.getstream.chat.android.client.scope.UserScope import io.getstream.chat.android.client.scope.UserTestScope import io.getstream.chat.android.client.socket.ErrorResponse import io.getstream.chat.android.client.socket.SocketErrorMessage import io.getstream.chat.android.client.socket.SocketFactory import io.getstream.chat.android.client.socket.StreamWebSocket +import io.getstream.chat.android.client.socket.V2SocketProtocol +import io.getstream.chat.android.models.NoOpChannelTransformer +import io.getstream.chat.android.models.NoOpMessageTransformer +import io.getstream.chat.android.models.NoOpUserTransformer import io.getstream.chat.android.randomBoolean import io.getstream.chat.android.randomString import io.getstream.chat.android.randomUser @@ -68,6 +75,16 @@ internal class CurrentUserFetcherTests { lateinit var webSocketListener: WebSocketListener lateinit var streamWebSocket: StreamWebSocket private val ws: WebSocket = mock() + private val config = ChatApiConfig( + apiKey = randomString(), + httpUrl = randomString(), + cdnHttpUrl = randomString(), + wssUrl = randomString(), + warmUp = randomBoolean(), + loggerConfig = mock(), + debugRequests = randomBoolean(), + notificationConfig = NotificationConfig(), + ).apply { isAnonymous = randomBoolean() } @BeforeEach fun before() { @@ -82,16 +99,7 @@ internal class CurrentUserFetcherTests { fetcher = CurrentUserFetcher( networkStateProvider = networkStateProvider, socketFactory = socketFactory, - ChatApiConfig( - apiKey = randomString(), - httpUrl = randomString(), - cdnHttpUrl = randomString(), - wssUrl = randomString(), - warmUp = randomBoolean(), - loggerConfig = mock(), - debugRequests = randomBoolean(), - notificationConfig = NotificationConfig(), - ).apply { isAnonymous = randomBoolean() }, + config, ) } @@ -196,4 +204,42 @@ internal class CurrentUserFetcherTests { result.isSuccess `should be equal to` true result.getOrThrow() `should be equal to` connectedEvent.me } + + @Test + fun `When the v2 socket connects, the own user is fetched`() = runTest { + /* Given */ + var listener: WebSocketListener? = null + val v2Socket = StreamWebSocket( + ParserFactory.createMoshiChatParser(), + V2SocketProtocol( + authFrame = "{}", + eventParser = V2EventParser( + DomainMapping({ "filip" }, NoOpChannelTransformer, NoOpMessageTransformer, NoOpUserTransformer, true), + ), + ), + ) { + listener = it + ws + } + val socketFactory: SocketFactory = mock() + whenever(socketFactory.createSocket(any())) doReturn v2Socket + val v2Fetcher = CurrentUserFetcher(networkStateProvider, socketFactory, config) + whenever(networkStateProvider.isConnected()) doReturn true + + /* When */ + val localScope = testCoroutines.scope + Job() + val deferredResult = localScope.async { v2Fetcher.fetch(randomUser()) } + testCoroutines.scope.advanceTimeBy(2_000) + listener!!.onMessage( + ws, + """{"type":"connection.ok","connection_id":"conn-1","created_at":1791976405614650759,"me":{"id":"filip", + "role":"user","language":"en","banned":false,"online":true,"invisible":false,"total_unread_count":0, + "unread_channels":0,"unread_count":0,"unread_threads":0,"created_at":1634853490403855000, + "updated_at":1791791284528654000,"custom":{}}}""", + ) + testCoroutines.scope.advanceTimeBy(2_000) + + /* Then */ + deferredResult.await().getOrNull()?.id `should be equal to` "filip" + } }