kotlin - Why does my Android client only collect a single emission from a Ktor SSE server? - Stack Overflow

admin管理员组

文章数量:1336621

I'm working on a project where an Android app connects to a Ktor server via Server-Sent Events (SSE). The server is supposed to continuously send updates whenever new data is emitted from the Arduino sensor. However, my Android client only collects a single emission and then stops listening. I test the endpoint of my backend on postman, and its work ok Here's my setup: Server-Side Code (Ktor):

fun Application.configureRouting() {
    val dataFlow = MutableSharedFlow<WhetherModule>(replay = 1, extraBufferCapacity = Int.MAX_VALUE)

    routing {
        route("/whether") {
            post {
                val receive = call.receive<WhetherModule>()
                val insert: WhetherModule? = whetherDaoImpl.insertData(receive)

                if (insert != null) {
                    dataFlow.emit(insert)
                    call.respond(HttpStatusCode.Created, insert)
                } else {
                    call.respond(HttpStatusCode.BadRequest, "Something went wrong")
                }
            }
        }

        sse("/events") {
            try {
                dataFlow.collect { data ->
                    send(ServerSentEvent(data = Json.encodeToString(data), event = "data-event"))
                }
            } catch (e: Exception) {
                log.error("Error during SSE collection: ${e.message}")
            }
        }
    }
}

Client-Side Code

class WhetherServer @Inject constructor(private val httpClient: HttpClient) {
 fun getServerEvent() =  flow<Resource<WhetherModule>> {

            Log.d(
                "receiveSSE",
                "outside the fun block->" + Thread.currentThread().name
            ) //DefaultDispatcher-worker-3

            httpClient.sse(
                port = 8080,
                path = "events"

            ) {


                incoming.collect { event: ServerSentEvent ->
                    Log.d(
                        "receiveSSE",
                        "inside the fun block->" + Thread.currentThread().name
                    )  


                    Log.d("ali osman", event.toString())
                    if (event.data != null) {
            val decodeFromString = Json.decodeFromString<WhetherModule>(event.data!!)
                                                          emit(Resource.Success(decodeFromString))


                    }

                }

            }





    }
        .onCompletion { cause ->
            val log : String = cause?.message   ?: "goodbye"
            Log.d("onCompletion", log)

        }.catch { cause: Throwable ->
            emit(Resource.Error(cause))

        }
 

the viewModel code

@HiltViewModel
class PostViewModel @Inject constructor(val postServer: PostServer) : ViewModel() {
private var _wheatherState = MutableStateFlow<Resource<WhetherModule>?>(null)
    val _wheatherState = _wsValue.asStateFlow()

  init {

        viewModelScope.launch {
            Log.d("viewModelScope", coroutineContext.toString())

            // receivePost()

            //receiveData()
            receiveSSE ()


        }


    } 
   private suspend fun receiveSSE (){

        postServer.getServerEvent2().collect { e->

            _wsValue.value =  e
        }
    }


}

On the client side, I tried using channelFlow to handle the SSE collection logic, but it did not resolve the issue. I suspect this is because the sse() function on the server runs in a different coroutine scope, causing the ViewModel's coroutine to terminate the Flow collection after the first emission. I expected the Android client to continuously collect and receive data from the server whenever new data was emitted. Any suggestion :)

I'm working on a project where an Android app connects to a Ktor server via Server-Sent Events (SSE). The server is supposed to continuously send updates whenever new data is emitted from the Arduino sensor. However, my Android client only collects a single emission and then stops listening. I test the endpoint of my backend on postman, and its work ok Here's my setup: Server-Side Code (Ktor):

fun Application.configureRouting() {
    val dataFlow = MutableSharedFlow<WhetherModule>(replay = 1, extraBufferCapacity = Int.MAX_VALUE)

    routing {
        route("/whether") {
            post {
                val receive = call.receive<WhetherModule>()
                val insert: WhetherModule? = whetherDaoImpl.insertData(receive)

                if (insert != null) {
                    dataFlow.emit(insert)
                    call.respond(HttpStatusCode.Created, insert)
                } else {
                    call.respond(HttpStatusCode.BadRequest, "Something went wrong")
                }
            }
        }

        sse("/events") {
            try {
                dataFlow.collect { data ->
                    send(ServerSentEvent(data = Json.encodeToString(data), event = "data-event"))
                }
            } catch (e: Exception) {
                log.error("Error during SSE collection: ${e.message}")
            }
        }
    }
}

Client-Side Code

class WhetherServer @Inject constructor(private val httpClient: HttpClient) {
 fun getServerEvent() =  flow<Resource<WhetherModule>> {

            Log.d(
                "receiveSSE",
                "outside the fun block->" + Thread.currentThread().name
            ) //DefaultDispatcher-worker-3

            httpClient.sse(
                port = 8080,
                path = "events"

            ) {


                incoming.collect { event: ServerSentEvent ->
                    Log.d(
                        "receiveSSE",
                        "inside the fun block->" + Thread.currentThread().name
                    )  


                    Log.d("ali osman", event.toString())
                    if (event.data != null) {
            val decodeFromString = Json.decodeFromString<WhetherModule>(event.data!!)
                                                          emit(Resource.Success(decodeFromString))


                    }

                }

            }





    }
        .onCompletion { cause ->
            val log : String = cause?.message   ?: "goodbye"
            Log.d("onCompletion", log)

        }.catch { cause: Throwable ->
            emit(Resource.Error(cause))

        }
 

the viewModel code

@HiltViewModel
class PostViewModel @Inject constructor(val postServer: PostServer) : ViewModel() {
private var _wheatherState = MutableStateFlow<Resource<WhetherModule>?>(null)
    val _wheatherState = _wsValue.asStateFlow()

  init {

        viewModelScope.launch {
            Log.d("viewModelScope", coroutineContext.toString())

            // receivePost()

            //receiveData()
            receiveSSE ()


        }


    } 
   private suspend fun receiveSSE (){

        postServer.getServerEvent2().collect { e->

            _wsValue.value =  e
        }
    }


}

On the client side, I tried using channelFlow to handle the SSE collection logic, but it did not resolve the issue. I suspect this is because the sse() function on the server runs in a different coroutine scope, causing the ViewModel's coroutine to terminate the Flow collection after the first emission. I expected the Android client to continuously collect and receive data from the server whenever new data was emitted. Any suggestion :)

• kotlin
• android-viewmodel
• kotlin-multiplatform
• ktor
• ktor-client

Share Improve this question Follow asked Nov 19, 2024 at 18:10 Ali OthmanAli Othman 1

Add a comment  | 

3 Answers 3

Sorted by: Reset to default 0

If your Android client is only receiving a single emission from a Ktor Server-Sent Events (SSE) server, the issue could stem from various causes related to either the server configuration or the client-side implementation.

The issue you’re experiencing—receiving only the first event in an SSE stream—is due to how you’re using client.sse and collecting into a Flow. The problem arises because Flow in this context processes the stream as a single collection operation, effectively completing the flow after the first event. This is not suitable for a continuous stream like SSE, where events are pushed asynchronously over time.

To resolve this, you should use channelFlow instead. channelFlow is specifically designed for use cases where you need to emit multiple values asynchronously. By using channelFlow, you can establish a persistent connection to the SSE stream and continuously send incoming events to the flow.

Here’s an example of how you can implement it:

fun listenToSSE(): Flow<SystemEventDto> = channelFlow {
    httpClient.sse(
        HttpRequestBuilder().apply {
            url("https://your-sse-endpoint")
        }
    ) {
        try {
            // Continuously read events and send them to the channel
            incoming.consumeEach { event ->
                 send(event.toSystemEventDto()) // Convert and send the event
            }
        } catch (e: Throwable) {
            // Handle errors or disconnection
            close(e)
        }
    }
}

In this approach: 1. channelFlow ensures the flow remains active while emitting values. 2. consumeEach is used to continuously process events from the stream. 3. You can handle disconnections or errors gracefully using the try-catch block.

This way, you’ll be able to receive and process all events in the SSE stream instead of only the first one.

Current SSE session implementation (e.g. OkHttpSSESession) might not propagate exceptions to the flow collectors. Use "client.prepareGet.execute" chain to properly catch exceptions.

In my case socket timeout was the reason OkHttp: Socket timeouts without timeout configured

Setting socketTimeoutMillis = Long.MAX_VALUE solved the issue for me.

本文标签： kotlinWhy does my Android client only collect a single emission from a Ktor SSE serverStack Overflow