Failing to subscribe to STOMP server.

[Dipesh424]

Hello @joffrey-bion,
I am having trouble subscribing to a STOMP server in my KMM project. I am trying to connect and subscribe. But i am not receiving any messages. To make sure it is not a server side issue, I also tested using "com.github.NaikSoftware:StompProtocolAndroid" library in my native android project. There I was able to subscribe and received messages from the server.

Here is my code for establishing connection
class WebSocketService {
val httpClient = HttpClient {
install(WebSockets)
}
private val stompConfig = StompConfig().apply {
heartBeat = HeartBeat(10000L.milliseconds, 10000L.milliseconds)
}

private val wsClient = KtorWebSocketClient(httpClient)

private val stompClient = StompClient(wsClient, stompConfig)

companion object {
    val instance = WebSocketService()
}

suspend fun connect() {
    try {
        val session = stompClient.connect(
            url = {URL},
        )

        session.use { s ->
            val subscription = s.subscribeText({END_POINT})
            val message = subscription.first()
            Napier.d("WebSocketService : $message")
            session.sendText("/nqrws/check-txn-status", Json.encodeToString(WebSocketData()))
        }

    } catch (e: Exception) {
        Napier.e("WebSocket connection error: ${e.message}")
        e.printStackTrace()
    }
}

}

this is my code form viewModel

class MainViewModel : BaseKMMViewModel() {
fun connect() {
viewModelScope.coroutineScope.launch {
WebSocketService.instance.connect()
}
}
}

This is my implementation for establishing connection and subscribe. And could you suggest me idea on how to check connection state. For eg, check connected , Disconneted status.

Thanks in advance.

[joffrey-bion]

Hi Dipesh, thanks for reaching out. What is the problem you're noticing exactly? Does it hang?

If yes, I think the problem lies in how you're expecting the first message here. When you use Flow.first(), the call will suspend until a message is received. So if your server doesn't send a message upon connection of a new client, your code will just hang there. If the server sends a message when you send something to /nqrws/check-txn-status, you cannot wait for the response before calling sendText().

You could try swapping the first() call and the sendText() call. Did you try this?

Another possibility would be to start waiting for the first message concurrently using val deferredMessage = async(start = CoroutineStart.UNDISPATCHED) { subscription.first() }, then call sendText(), and then call deferredMessage.await().

[Dipesh424]

Furthermore, I also implemented another library to test in my native android project. In that I am getting a response like this
Emit lifecycle event: OPENED
2024-03-05 16:25:22.780 31652-31678 AbstractCo...onProvider com.practicalaction.stomptest D Send STOMP message: CONNECT
accept-version:1.1,1.2
heart-beat:0,0

                                                                                                ??

2024-03-05 16:25:22.781 31652-31678 StompClient com.practicalaction.stomptest D Publish open
2024-03-05 16:25:22.790 31652-31652 Compatibil...geReporter com.practicalaction.stomptest D Compat change id reported: 147798919; UID 10176; state: ENABLED
2024-03-05 16:25:22.805 31652-31678 AbstractCo...onProvider com.practicalaction.stomptest D Receive STOMP message: CONNECTED
version:1.2
heart-beat:0,0
user-name:20240305162522183d0c76bce-b57d-4ddd-9e75-ef4af9efe6e5

                                                                                                ??

2024-03-05 16:25:22.821 31652-31678 AbstractCo...onProvider com.practicalaction.stomptest D Send STOMP message: SUBSCRIBE
id:1d8dca78-7bec-4ec1-8542-ba872c9a234b
destination:/user/nqrws/check-txn-status
ack:auto

                                                                                                ??

I am expecting same on my KMM project.

[joffrey-bion]

Ok thanks for sharing. I'm not sure I really have the full picture here, so I have a couple more questions if you don't mind:

1. Am I right that the failure you get is actually a failure on connect(), and that we actually don't even get to the subscribe() part?
2. Is your server using STOMP over web sockets, or is it STOMP over a raw TCP connection? Krossbow only supports STOMP over web socket for now. See Support for STOMP over raw TCP connections #287
3. Do you have any errors in the server logs when connecting from Krossbow? It would be great to share if you have access to those logs. If not, is the server public so I could try to test against it?
4. Which Krossbow web socket implementation are you using? Your errors seem to suggest OkHttp. Is it krossbow-websocket-okhttp or krossbow-websocket-ktor with Ktor's OkHttp engine?

[Dipesh424]

1. I think the connection part works well in this case. The error might be caused in subscribe part. Also i get this error log after catching the error like this:
   code -->>
   session.use { s ->
   s.subscribeText("/user/nqrws/check-txn-status")
   .catch {
   Napier.d("OnCatchSubscribe : ${it.message}")
   it.printStackTrace()
   }.collect { frame ->
   Napier.d("Subscribed : $frame")

                }
        }
   

Error Log-->>
OnCatchSubscribe : error in Ktor's websocket: null
2024-03-05 16:50:03.708 32743-32743 System.err com.delivery.merokinmel.android W org.hildan.krossbow.websocket.WebSocketException: error in Ktor's websocket: null
2024-03-05 16:50:03.708 32743-32743 System.err com.delivery.merokinmel.android W at org.hildan.krossbow.websocket.ktor.KtorWebSocketConnectionAdapter$incomingFrames$3.invokeSuspend(KtorWebSocketClient.kt:63)
2024-03-05 16:50:03.708 32743-32743 System.err com.delivery.merokinmel.android W at org.hildan.krossbow.websocket.ktor.KtorWebSocketConnectionAdapter$incomingFrames$3.invoke(Unknown Source:9)
2024-03-05 16:50:03.708 32743-32743 System.err com.delivery.merokinmel.android W at org.hildan.krossbow.websocket.ktor.KtorWebSocketConnectionAdapter$incomingFrames$3.invoke(Unknown Source:6)
2024-03-05 16:50:03.708 32743-32743 System.err com.delivery.merokinmel.android W at kotlinx.coroutines.flow.FlowKt__ErrorsKt$catch$$inlined$unsafeFlow$1.collect(SafeCollector.common.kt:115)
2024-03-05 16:50:03.709 32743-32743 System.err com.delivery.merokinmel.android W at kotlinx.coroutines.flow.FlowKt__ErrorsKt$catch$$inlined$unsafeFlow$1$1.invokeSuspend(Unknown Source:15)
2024-03-05 16:50:03.709 32743-32743 System.err com.delivery.merokinmel.android W at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)

2. Yes, the server is using STOMP over web sockets.
3. I am sorry but I do not have access to server logs. However, we are trying to contact with server team on this. And I am sorry that server is not public.
4. Currently I am using krossbow-websocket-ktor. I will also try using krossbow-websocket-okhttp and get back to you.

Thanks for your support.

[joffrey-bion]

Thanks a lot for your quick replies. So, as far as I understood, the connection goes well. Then the subscription itself goes well too (the SUBSCRIBE frame is sent successfully), but then nothing is received over the flow and the server seems to timeout somehow.

Did you try without heartbeats by any chance? This might avoid the interference of the timeout from the heartbeat, but I don't think it will help if the problem is that we don't receive the on-subscribe message frame from the server (unless the server somehow doesn't like the heartbeat config and there is a negotiation issue).

I am sorry but I don't see how else I can help in the current situation (without knowing from the server-side if something is wrong with the client's behavior, or if the client uses something that is not supported by the server).

Please let me know if you hear back from the server team. I'd be happy to fix Krossbow if there is a bug somewhere.

[Dipesh424]

And also thanks to you for your quick replies. I will let know once I get something from server team.

[joffrey-bion]

And could you suggest me idea on how to check connection state. For eg, check connected , Disconneted status.

Regarding this, the way it currently works is that the call to connect() will just fail with an exception if the client cannot connect. Later on, any consumer of a subscription flow will also throw an exception if the server disconnects.

This is not ideal to handle reconnections, I know. Sorry! I'll work on some solution to fix this properly in #102

[Kova700]

Regarding this, the way it currently works is that the call to connect() will just fail with an exception if the client cannot connect. Later on, any consumer of a subscription flow will also throw an exception if the server disconnects.

If a socket is already connected, does it prevent duplicate socket connections and throw an exception? If so, what exception does it throw?
(+ As far as I know, when a MissingHeartBeatException occurs, the connection to the server is lost. Does that mean the subscription flow also ends?)

[joffrey-bion]

If a socket is already connected, does it prevent duplicate socket connections and throw an exception? If so, what exception does it throw?

No, it won't throw an exception. You can definitely open multiple connections at the same time and get multiple StompSessions.

As far as I know, when a MissingHeartBeatException occurs, the connection to the server is lost. Does that mean the subscription flow also ends?

Yes, the collectors of subscription flows will fail in this case (they will get a MissingHeartBeatException too). This is the test that ensures this behaviour:

krossbow/krossbow-stomp-core/src/commonTest/kotlin/org/hildan/krossbow/stomp/StompSessionHeartBeatsTests.kt

Lines 127 to 146 in b485ab5

	@Test
	fun receiveSubMessage_failsOnHeartBeatTimeOut() = runTest {
	val (wsSession, stompSession) = connectWithMocks(
	StompConnectedHeaders(heartBeat = HeartBeat(minSendPeriod = 1000.milliseconds, expectedPeriod = ZERO))
	) {
	heartBeat = HeartBeat(ZERO, 1000.milliseconds)
	heartBeatTolerance = HeartBeatTolerance(incomingMargin = 100.milliseconds)
	}
	launch {
	wsSession.awaitSubscribeFrameAndSimulateCompletion()
	wsSession.expectClose()
	}
	val messages = stompSession.subscribeText("/dest")
	val exception = assertFailsWith(MissingHeartBeatException::class) {
	messages.first()
	}
	assertEquals(1000.milliseconds, exception.expectedPeriod)
	}

[Kova700]

Thank you very much for your kind reply.😊