fix(#280): decouple WebSocket message parsing from network reader thread (#321)

Moved JSON parsing from the OkHttp `WebSocketListener.onMessage` callback to a background `Dispatchers.IO` coroutine. Instead of processing incoming messages synchronously on the reader thread, raw messages are now emitted into a buffered `MutableSharedFlow`. This prevents slow or backpressured collectors from blocking the network reader, thereby eliminating perceived lag during high-throughput message streaming.
This commit is contained in:
M57 authored and GitHub committed 2026-06-23 12:22:21 +03:00
1 parent 3798d0330f
commit d83ef75cd9
1 file changed
+21 -15
@@ -13,9 +13,13 @@ import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.shareIn
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import okhttp3.CertificatePinner import okhttp3.CertificatePinner
import okhttp3.OkHttpClient import okhttp3.OkHttpClient
@@ -90,10 +94,23 @@ object HermesWsClient {
// ── Public observable stream ───────────────────────────────────────── // ── Public observable stream ─────────────────────────────────────────
private val _events = MutableSharedFlow<WsEvent>(extraBufferCapacity = 64) private val rawMessages = MutableSharedFlow<String>(extraBufferCapacity = 64)
/** Collect this from ViewModels to receive all parsed [WsEvent]s. */ /** Collect this from ViewModels to receive all parsed [WsEvent]s. */
val events: SharedFlow<WsEvent> = _events.asSharedFlow() val events: SharedFlow<WsEvent> =
rawMessages
.buffer()
.map { text ->
try {
val rpc = gson.fromJson(text, JsonRpcResponse::class.java)
EventParser.parse(rpc, text)
} catch (e: Exception) {
Log.e(TAG, "Failed to parse message", e)
WsEvent.Unknown(text)
}
}
.flowOn(Dispatchers.IO)
.shareIn(wsScope, SharingStarted.Eagerly)
// ── Connection status flow ────────────────────────────────────────── // ── Connection status flow ──────────────────────────────────────────
private val _connectionStatus = MutableStateFlow(ConnectionStatus.DISCONNECTED) private val _connectionStatus = MutableStateFlow(ConnectionStatus.DISCONNECTED)
@@ -217,10 +234,6 @@ object HermesWsClient {
} }
} }
private fun emit(event: WsEvent) {
_events.tryEmit(event)
}
// ── Listener ───────────────────────────────────────────────────────── // ── Listener ─────────────────────────────────────────────────────────
private class WsListenerImpl : WebSocketListener() { private class WsListenerImpl : WebSocketListener() {
@@ -242,14 +255,7 @@ object HermesWsClient {
// AI reply tokens and tool output — never stream to logcat in // AI reply tokens and tool output — never stream to logcat in
// release builds. // release builds.
if (BuildConfig.DEBUG) Log.d(TAG, "← $text") if (BuildConfig.DEBUG) Log.d(TAG, "← $text")
try { rawMessages.tryEmit(text)
val rpc = gson.fromJson(text, JsonRpcResponse::class.java)
val event = EventParser.parse(rpc, text)
emit(event)
} catch (e: Exception) {
Log.e(TAG, "Failed to parse message", e)
emit(WsEvent.Unknown(text))
}
} }
override fun onClosing( override fun onClosing(