WebSocket Data Sending Performance Issue

[MAXBAF1]

Hello! I’m using your library in my mobile application, and I’ve encountered an issue with WebSocket data sending performance.

WebSocket connection code

private val wsClient = KtorWebSocketClient(httpClient)
private var connection: WebSocketConnectionWithPingPong? = null
private val json = Json {
    ignoreUnknownKeys = true
    coerceInputValues = true
}
private var stompSession: StompSessionWithKxSerialization? = null

suspend fun connectToWebSocket() {
    try {
        connection = wsClient.connect(
            url = AroundConfig.WS_HOST.toString(),
            headers = mapOf(HttpHeaders.Authorization to "Bearer $accessToken"),
        )
        stompSession = connection
            ?.stomp(
                host = AroundConfig.STOMP_HOST.toString(),
                config = StompConfig().apply {
                    receiptTimeout = 60.seconds
                    connectionTimeout = 60.seconds
                    disconnectTimeout = 60.seconds

                    heartBeat = HeartBeat(5.seconds)
                    heartBeatTolerance = HeartBeatTolerance(5.seconds)
                },
            )
            ?.withJsonConversions(json)
    } catch (e: Exception) {
        val cause = e.cause
        if (cause is StompErrorFrameReceived) {
            Log.e(
                "MyLog",
                "STOMP connection failure: ${cause.message}\n${cause.frame.bodyAsText}\n${e.stackTraceToString()}"
            )
        } else {
            Log.e(
                "MyLog", "WebSocket failure: ${cause?.message}\n${e.stackTraceToString()}"
            )
        }
    }
}

Data sending code

suspend fun sendCellToWebSocket(cell: CellDTO) {
    withContext(Dispatchers.IO) {
        try {
            Log.d("MyLog", "SEND CELL: $cell")
            stompSession?.convertAndSend(
                headers = StompSendHeaders(destination = AroundConfig.CELLS_CHANGES_FROM_USER.toString()),
                body = cell,
                serializer = CellDTO.serializer(),
            )
        } catch (e: Exception) {
            Log.e("MyLog", "sendCellToWebSocket ERROR $e")
        }
    }

Data receiving code

fun receiveCellsFromWebSocket(): Flow<List<CellDTO>> = flow {
    try {
        val changes = stompSession?.subscribe(
            StompSubscribeHeaders(destination = AroundConfig.CELLS_CHANGES_EVENT.toString())
        )

        changes?.collect { stompFrame ->
            val receivedText = stompFrame.bodyAsText
            val receivedCells = Json.decodeFromString<List<CellDTO>>(receivedText)
            Log.i("MyLog", "SUCCESS get cells from ws: $receivedCells")
            emit(receivedCells)
        }
    } catch (e: Exception) {
        Log.e("MyLog", "receiveCellsFromWebSocket ERROR $e")
    }
}

Issue

Initially, data sending works quickly, and data from the server arrives almost instantly. For example:

However, after receiving some data and waiting for a while, the delay in sending starts to grow significantly. Here are some examples:
In these two cases, the delay is approximately 20 seconds:


In this case, it exceeds a minute:

We conducted testing by creating a tunnel and monitoring message delivery. The delay occurs exclusively on my side. For instance, in the last screenshot (focusing on the cell with the identifier ending in 89bfff), out of the total time of 1 minute and 2 seconds required for the message to travel to the backend and back to me, exactly 1 minute was spent on sending it from my side. The backend received the message at approximately 19:45:14, processed it, and sent the response, which I received in less than a second.

To confirm this, we ran an additional test directly on the backend. This test sending thousands of cells via WebSocket as a user, using the same tunnel (not localhost). The backend processed them and sent them back, while I was subscribed to this transmission. The time taken for a single cell to travel from the backend test to the backend itself was no more than 100 ms. From the backend to me, the cells arrived in milliseconds.

Based on this, it can be concluded that the delay in sending data through WebSocket occurs exclusively on the mobile application's side.

Questions

• Is it possible that I’m doing something wrong when sending messages?
• I added a "SEND CELL" log right before sending. Is it possible to enable built-in logging in the library for more accurate diagnostics, instead of relying on my custom logs?

I tested on krossbow versions 7.1.0 and 8.2.0, there is no difference.

[joffrey-bion]

Hi! Thanks for the report. This is definitely something worth looking at.

If I understood correctly your client code runs on Android, and from the code above it looks like you're using the Ktor client (with krossbow-websocket-ktor). Is this correct?
Which Ktor engine are you using behind the scenes? Are you using ktor-client-android, ktor-client-okhttp, ktor-client-cio, or maybe another one? Did you reproduce this issue with several engines?

Also, did you try and reproduce this with another Krossbow web socket implementation (like krossbow-websocket-okhttp)?

To confirm this, we ran an additional test directly on the backend. This test sending thousands of cells via WebSocket as a user

Which web socket client did you use for this test?

Thanks in advance for your answers

[MAXBAF1]

Yes, we are testing krossbow-websocket-okhttp right now. I’ll let you know if the same issue occurs; right now, we are working on heartbeats.

[MAXBAF1]

Using krossbow-websocket-tor everything works fast.

private val accessToken = tokenManager.getTokens()?.accessToken
    private val wsClient = OkHttpWebSocketClient()
    private val json = Json {
        ignoreUnknownKeys = true
        coerceInputValues = true
    }
    private var stompSession: StompSessionWithKxSerialization? = null

    suspend fun connectToWebSocket() {
        try {
            val connection = wsClient.connect(
                url = AroundConfig.WS_HOST.toString(),
                headers = mapOf(HttpHeaders.Authorization to "Bearer $accessToken"),
            )
            stompSession = connection
                .stomp(
                    host = AroundConfig.STOMP_HOST.toString(),
                    config = StompConfig().apply {
                        heartBeat = HeartBeat(0.seconds, 10.seconds)
                    },
                )
                .withJsonConversions(json)
        } catch (e: Exception) {
            val cause = e.cause
            if (cause is StompErrorFrameReceived) {
                Log.e(
                    "MyLog",
                    "STOMP connection failure: ${cause.message}\n${cause.frame.bodyAsText}\n${e.stackTraceToString()}"
                )
            } else {
                Log.e(
                    "MyLog", "WebSocket failure: ${cause?.message}\n${e.stackTraceToString()}"
                )
            }
        }
    }

We experimented with different heartbeat values, it only works fine with HeartBeat(0.seconds, 10.seconds), as we realized, this means that the client does not send anything, but the backend sends it every 10 seconds. With any other values, error 1002 appears due to the fact that RabbitMQ breaks the connection with an error.
Error on the backend:

Error on the RabbitMQ:

Even though we have achieved continuous operation using HeartBeat(0.seconds, 10.seconds), it seems to us that this is not a very good solution.

Is there anyway to enable logs in krossbow itself to track the sending and receiving of heartbeats, as well as the sending and receiving of messages?

[joffrey-bion]

Hi again! Sorry for the delay, Christmas vacation and all :)

Using krossbow-websocket-tor everything works fast

You meant krossbow-websocket-okhttp, right?

Is there anyway to enable logs in krossbow itself to track the sending and receiving of heartbeats, as well as the sending and receiving of messages?

There is no internal logging in Krossbow at the moment. However, you can provide an implementation of KrossbowInstrumentation in your StompConfig (see StompConfig.instrumentation). Some of the callbacks in the instrumentation will probably help you add the logs you need. For example onWebSocketFrameReceived can be used to log every web socket frame (including heart beats and messages).

[joffrey-bion]

Regarding the heartbeat issue, have you tried adding some outgoing heartbeat tolerance? There is a config option called heartBeatTolerance. It allows to define some tolerance for the heartbeats in both directions. For the outgoing heartbeats (the ones sent by the client), this means sending the heartbeats slightly earlier than the deadline in case the server is too strict (to account for network latency and other potential slowdown). This might be what you need.

[MAXBAF1]

Yes, I was referring to krossbow-websocket-okhttp. I'll try CrossbowInstrumentation and heartBeatTolerance, thanks.