From d83ef75cd96d6192efda4f5b5c7e339fa9ca4798 Mon Sep 17 00:00:00 2001
From: M57 <41645758+Hy4ri@users.noreply.github.com>
Date: Tue, 23 Jun 2026 12:22:21 +0300
Subject: [PATCH] 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.
---
 .../hermescontrol/data/ws/HermesWsClient.kt   | 36 +++++++++++--------
 1 file changed, 21 insertions(+), 15 deletions(-)

diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt
index 8cebceaa..d8fe7630 100644
--- a/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt
+++ b/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt
@@ -13,9 +13,13 @@ import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.asSharedFlow
 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 okhttp3.CertificatePinner
 import okhttp3.OkHttpClient
@@ -90,10 +94,23 @@ object HermesWsClient {
 
     // ── 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. */
-    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 ──────────────────────────────────────────
     private val _connectionStatus = MutableStateFlow(ConnectionStatus.DISCONNECTED)
@@ -217,10 +234,6 @@ object HermesWsClient {
             }
     }
 
-    private fun emit(event: WsEvent) {
-        _events.tryEmit(event)
-    }
-
     // ── Listener ─────────────────────────────────────────────────────────
 
     private class WsListenerImpl : WebSocketListener() {
@@ -242,14 +255,7 @@ object HermesWsClient {
             // AI reply tokens and tool output — never stream to logcat in
             // release builds.
             if (BuildConfig.DEBUG) Log.d(TAG, "← $text")
-            try {
-                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))
-            }
+            rawMessages.tryEmit(text)
         }
 
         override fun onClosing(