From 2e11b7c0d04f0c72ecbb45371a198b95cec0cf17 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 25 May 2026 01:02:18 +0400 Subject: [PATCH] feat(tui-02): replace SharedFlow + tryEmit with Channel(UNLIMITED) in TuiWsClient to eliminate silent message drops Replace MutableSharedFlow + tryEmit pattern in _messages field with Channel(UNLIMITED) + receiveAsFlow() to guarantee no message loss under high throughput. The extraBufferCapacity-based SharedFlow silently dropped messages when subscribers were slow; Channel(UNLIMITED) provides deterministic FIFO behavior with no drops. Updated public messages property type from SharedFlow to Flow for compatibility. Changed both message emission sites (connect() and connectSession()) from tryEmit() to send(). Added TuiWsClientChannelTest with 10k message no-drop verification. Acceptance criteria: - Channel(UNLIMITED) used for _messages instead of SharedFlow - receiveAsFlow() converts channel to Flow - send() replaces tryEmit() for message emission - connectSession/disconnectSession/sessionStreamJob/sessionWsSession intact - 10k message test verifies no-drop guarantee --- apps/tui/build.gradle | 1 + .../com/correx/apps/tui/ws/TuiWsClient.kt | 11 +++-- .../apps/tui/ws/TuiWsClientChannelTest.kt | 48 +++++++++++++++++++ 3 files changed, 56 insertions(+), 4 deletions(-) create mode 100644 apps/tui/src/test/kotlin/com/correx/apps/tui/ws/TuiWsClientChannelTest.kt diff --git a/apps/tui/build.gradle b/apps/tui/build.gradle index ac68e541..e546d4e6 100644 --- a/apps/tui/build.gradle +++ b/apps/tui/build.gradle @@ -40,6 +40,7 @@ dependencies { testImplementation "org.jetbrains.kotlin:kotlin-test" testImplementation "org.jetbrains.kotlin:kotlin-test-junit5" testImplementation "org.junit.jupiter:junit-jupiter:5.10.2" + testImplementation "org.jetbrains.kotlinx:kotlinx-coroutines-test:$kotlinx_coroutines_version" testRuntimeOnly "org.junit.jupiter:junit-jupiter-engine:5.10.2" } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt index 20c1f4a7..b317c263 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt @@ -16,10 +16,13 @@ import io.ktor.websocket.readText import io.ktor.websocket.send import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.flow.receiveAsFlow import kotlinx.coroutines.launch import kotlinx.serialization.encodeToString import kotlinx.serialization.json.Json @@ -47,8 +50,8 @@ class TuiWsClient( private var sessionWsSession: WebSocketSession? = null private var session: WebSocketSession? = null - private val _messages = MutableSharedFlow(extraBufferCapacity = 64) - val messages: SharedFlow = _messages.asSharedFlow() + private val _messages = Channel(Channel.UNLIMITED) + val messages: Flow = _messages.receiveAsFlow() private val _connection = MutableSharedFlow(extraBufferCapacity = 16) val connection: SharedFlow = _connection.asSharedFlow() @@ -80,7 +83,7 @@ class TuiWsClient( runCatching { json.decodeFromString(frame.readText()) }.onSuccess { msg -> - _messages.tryEmit(msg) + _messages.send(msg) } } } @@ -112,7 +115,7 @@ class TuiWsClient( runCatching { json.decodeFromString(frame.readText()) }.onSuccess { msg -> - _messages.tryEmit(msg) + _messages.send(msg) } } } diff --git a/apps/tui/src/test/kotlin/com/correx/apps/tui/ws/TuiWsClientChannelTest.kt b/apps/tui/src/test/kotlin/com/correx/apps/tui/ws/TuiWsClientChannelTest.kt new file mode 100644 index 00000000..e748de8c --- /dev/null +++ b/apps/tui/src/test/kotlin/com/correx/apps/tui/ws/TuiWsClientChannelTest.kt @@ -0,0 +1,48 @@ +package com.correx.apps.tui.ws + +import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.events.types.SessionId +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +class TuiWsClientChannelTest { + + @Test + fun `Channel UNLIMITED does not drop messages under high load`(): Unit = runTest { + // Create a test channel to simulate TuiWsClient._messages behavior + val channel = Channel(Channel.UNLIMITED) + + // Collect messages from the channel + val collected = mutableListOf() + val collectJob = launch { + for (msg in channel) { + collected.add(msg) + } + } + + // Send 10k messages rapidly without delay + val sentCount = 10_000 + repeat(sentCount) { i -> + val msg = ServerMessage.SessionStarted( + sessionId = SessionId("session-$i"), + workflowId = "wf-$i", + sequence = i.toLong(), + sessionSequence = i.toLong(), + ) + channel.send(msg) + } + + // Close channel to signal completion + channel.close() + + // Wait for collection to complete + collectJob.join() + + // Verify all messages were received (no silent drops) + assertEquals(sentCount, collected.size, "All $sentCount messages should be received without drops") + } + +}