- Remove onMessage/onConnected/onDisconnected/onError callback
properties from HermesWsClient — redundant with events SharedFlow
and connectionStatus StateFlow
- Rewrite all 10 tests to use connectionStatus.first{} and
events.first{} instead of callback + CountDownLatch patterns
- Remove polling loop for RECONNECTING status — flow inherently
waits for the right value
Co-authored-by: M57 <hy4ri@users.noreply.github.com>
This commit is contained in:
1 parent
15a4748634
commit
2acfaabfcb
2 files changed
+29
-96
No files matched your search
@@ -93,13 +93,6 @@ object HermesWsClient {
|
|||||||
/** Observable connection status: DISCONNECTED / CONNECTING / CONNECTED / RECONNECTING */
|
/** Observable connection status: DISCONNECTED / CONNECTING / CONNECTED / RECONNECTING */
|
||||||
val connectionStatus: StateFlow<ConnectionStatus> = _connectionStatus.asStateFlow()
|
val connectionStatus: StateFlow<ConnectionStatus> = _connectionStatus.asStateFlow()
|
||||||
|
|
||||||
// ── Callbacks (legacy / convenience) ─────────────────────────────────
|
|
||||||
|
|
||||||
var onMessage: ((WsEvent) -> Unit)? = null
|
|
||||||
var onConnected: (() -> Unit)? = null
|
|
||||||
var onDisconnected: ((reason: String?) -> Unit)? = null
|
|
||||||
var onError: ((Throwable) -> Unit)? = null
|
|
||||||
|
|
||||||
// ── Connection helpers ────────────────────────────────────────────────
|
// ── Connection helpers ────────────────────────────────────────────────
|
||||||
|
|
||||||
val isConnected: Boolean get() = connected.get()
|
val isConnected: Boolean get() = connected.get()
|
||||||
@@ -216,7 +209,6 @@ object HermesWsClient {
|
|||||||
|
|
||||||
private fun emit(event: WsEvent) {
|
private fun emit(event: WsEvent) {
|
||||||
_events.tryEmit(event)
|
_events.tryEmit(event)
|
||||||
onMessage?.invoke(event)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── Listener ─────────────────────────────────────────────────────────
|
// ── Listener ─────────────────────────────────────────────────────────
|
||||||
@@ -230,7 +222,6 @@ object HermesWsClient {
|
|||||||
connected.set(true)
|
connected.set(true)
|
||||||
_connectionStatus.value = ConnectionStatus.CONNECTED
|
_connectionStatus.value = ConnectionStatus.CONNECTED
|
||||||
currentBackoff = INITIAL_BACKOFF_MS
|
currentBackoff = INITIAL_BACKOFF_MS
|
||||||
onConnected?.invoke()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
override fun onMessage(
|
override fun onMessage(
|
||||||
@@ -267,7 +258,6 @@ object HermesWsClient {
|
|||||||
) {
|
) {
|
||||||
Log.i(TAG, "WebSocket closed: $code $reason")
|
Log.i(TAG, "WebSocket closed: $code $reason")
|
||||||
connected.set(false)
|
connected.set(false)
|
||||||
onDisconnected?.invoke(reason)
|
|
||||||
_connectionStatus.value = ConnectionStatus.RECONNECTING
|
_connectionStatus.value = ConnectionStatus.RECONNECTING
|
||||||
scheduleReconnect()
|
scheduleReconnect()
|
||||||
}
|
}
|
||||||
@@ -279,8 +269,6 @@ object HermesWsClient {
|
|||||||
) {
|
) {
|
||||||
Log.e(TAG, "WebSocket failure: ${t.message}", t)
|
Log.e(TAG, "WebSocket failure: ${t.message}", t)
|
||||||
connected.set(false)
|
connected.set(false)
|
||||||
onError?.invoke(t)
|
|
||||||
onDisconnected?.invoke(t.message)
|
|
||||||
_connectionStatus.value = ConnectionStatus.RECONNECTING
|
_connectionStatus.value = ConnectionStatus.RECONNECTING
|
||||||
scheduleReconnect()
|
scheduleReconnect()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,6 +7,10 @@ import io.mockk.mockkObject
|
|||||||
import io.mockk.mockkStatic
|
import io.mockk.mockkStatic
|
||||||
import io.mockk.unmockkAll
|
import io.mockk.unmockkAll
|
||||||
import kotlinx.coroutines.flow.MutableStateFlow
|
import kotlinx.coroutines.flow.MutableStateFlow
|
||||||
|
import kotlinx.coroutines.flow.first
|
||||||
|
import kotlinx.coroutines.launch
|
||||||
|
import kotlinx.coroutines.runBlocking
|
||||||
|
import kotlinx.coroutines.withTimeout
|
||||||
import okhttp3.WebSocket
|
import okhttp3.WebSocket
|
||||||
import okhttp3.WebSocketListener
|
import okhttp3.WebSocketListener
|
||||||
import okhttp3.mockwebserver.MockResponse
|
import okhttp3.mockwebserver.MockResponse
|
||||||
@@ -94,13 +98,8 @@ class HermesWsClientTest {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
val clientLatch = CountDownLatch(1)
|
|
||||||
HermesWsClient.onConnected = {
|
|
||||||
clientLatch.countDown()
|
|
||||||
}
|
|
||||||
|
|
||||||
HermesWsClient.connect()
|
HermesWsClient.connect()
|
||||||
assertTrue("Client failed to connect", clientLatch.await(5, TimeUnit.SECONDS))
|
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
|
||||||
assertTrue("Server failed to accept connection", serverLatch.await(5, TimeUnit.SECONDS))
|
assertTrue("Server failed to accept connection", serverLatch.await(5, TimeUnit.SECONDS))
|
||||||
assertTrue(HermesWsClient.isConnected)
|
assertTrue(HermesWsClient.isConnected)
|
||||||
assertEquals(ConnectionStatus.CONNECTED, HermesWsClient.connectionStatus.value)
|
assertEquals(ConnectionStatus.CONNECTED, HermesWsClient.connectionStatus.value)
|
||||||
@@ -136,22 +135,10 @@ class HermesWsClientTest {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
val clientLatch = CountDownLatch(1)
|
|
||||||
HermesWsClient.onConnected = {
|
|
||||||
clientLatch.countDown()
|
|
||||||
}
|
|
||||||
|
|
||||||
HermesWsClient.connect()
|
HermesWsClient.connect()
|
||||||
assertTrue(clientLatch.await(5, TimeUnit.SECONDS))
|
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
|
||||||
assertTrue(serverLatch.await(5, TimeUnit.SECONDS))
|
assertTrue(serverLatch.await(5, TimeUnit.SECONDS))
|
||||||
|
|
||||||
val eventLatch = CountDownLatch(1)
|
|
||||||
var receivedEvent: WsEvent? = null
|
|
||||||
HermesWsClient.onMessage = { event ->
|
|
||||||
receivedEvent = event
|
|
||||||
eventLatch.countDown()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Server sends a message to client
|
// Server sends a message to client
|
||||||
val jsonResponse =
|
val jsonResponse =
|
||||||
"""
|
"""
|
||||||
@@ -161,9 +148,15 @@ class HermesWsClientTest {
|
|||||||
"result": "success"
|
"result": "success"
|
||||||
}
|
}
|
||||||
""".trimIndent()
|
""".trimIndent()
|
||||||
serverWebSocket?.send(jsonResponse)
|
|
||||||
|
|
||||||
assertTrue("Event not received", eventLatch.await(5, TimeUnit.SECONDS))
|
val receivedEvent =
|
||||||
|
runBlocking {
|
||||||
|
withTimeout(5000) {
|
||||||
|
launch { serverWebSocket?.send(jsonResponse) }
|
||||||
|
HermesWsClient.events.first { it is WsEvent.RpcResult }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
assertTrue(receivedEvent is WsEvent.RpcResult)
|
assertTrue(receivedEvent is WsEvent.RpcResult)
|
||||||
assertEquals("1", (receivedEvent as WsEvent.RpcResult).id)
|
assertEquals("1", (receivedEvent as WsEvent.RpcResult).id)
|
||||||
}
|
}
|
||||||
@@ -194,17 +187,8 @@ class HermesWsClientTest {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
val clientLatch = CountDownLatch(1)
|
|
||||||
val clientDisconnectedLatch = CountDownLatch(1)
|
|
||||||
HermesWsClient.onConnected = {
|
|
||||||
clientLatch.countDown()
|
|
||||||
}
|
|
||||||
HermesWsClient.onDisconnected = { _ ->
|
|
||||||
clientDisconnectedLatch.countDown()
|
|
||||||
}
|
|
||||||
|
|
||||||
HermesWsClient.connect()
|
HermesWsClient.connect()
|
||||||
assertTrue(clientLatch.await(5, TimeUnit.SECONDS))
|
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
|
||||||
assertTrue(serverLatch.await(5, TimeUnit.SECONDS))
|
assertTrue(serverLatch.await(5, TimeUnit.SECONDS))
|
||||||
|
|
||||||
HermesWsClient.disconnect()
|
HermesWsClient.disconnect()
|
||||||
@@ -244,13 +228,8 @@ class HermesWsClientTest {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
val clientLatch = CountDownLatch(1)
|
|
||||||
HermesWsClient.onConnected = {
|
|
||||||
clientLatch.countDown()
|
|
||||||
}
|
|
||||||
|
|
||||||
HermesWsClient.connect()
|
HermesWsClient.connect()
|
||||||
assertTrue("Client failed to connect", clientLatch.await(5, TimeUnit.SECONDS))
|
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
|
||||||
assertTrue("Server failed to accept connection", serverLatch.await(5, TimeUnit.SECONDS))
|
assertTrue("Server failed to accept connection", serverLatch.await(5, TimeUnit.SECONDS))
|
||||||
|
|
||||||
// Use the convenience method
|
// Use the convenience method
|
||||||
@@ -303,37 +282,21 @@ class HermesWsClientTest {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
val clientConnected1Latch = CountDownLatch(1)
|
|
||||||
val clientDisconnectedLatch = CountDownLatch(1)
|
|
||||||
|
|
||||||
HermesWsClient.onConnected = {
|
|
||||||
clientConnected1Latch.countDown()
|
|
||||||
}
|
|
||||||
HermesWsClient.onDisconnected = { _ ->
|
|
||||||
clientDisconnectedLatch.countDown()
|
|
||||||
}
|
|
||||||
|
|
||||||
HermesWsClient.connect()
|
HermesWsClient.connect()
|
||||||
|
|
||||||
assertTrue("Failed initial connection", connect1Latch.await(5, TimeUnit.SECONDS))
|
assertTrue("Failed initial connection", connect1Latch.await(5, TimeUnit.SECONDS))
|
||||||
assertTrue("Client failed initial connection", clientConnected1Latch.await(5, TimeUnit.SECONDS))
|
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
|
||||||
assertEquals(ConnectionStatus.CONNECTED, HermesWsClient.connectionStatus.value)
|
assertEquals(ConnectionStatus.CONNECTED, HermesWsClient.connectionStatus.value)
|
||||||
|
|
||||||
// Force server to close socket 1 to trigger reconnect
|
// Force server to close socket 1 to trigger reconnect
|
||||||
serverSocket1?.close(1001, "Server shutting down")
|
serverSocket1?.close(1001, "Server shutting down")
|
||||||
|
|
||||||
assertTrue("Client didn't detect disconnect", clientDisconnectedLatch.await(5, TimeUnit.SECONDS))
|
// Wait for status to become RECONNECTING
|
||||||
|
runBlocking {
|
||||||
// Poll for status to become RECONNECTING (async race mitigation —
|
withTimeout(
|
||||||
// onClosed fires onDisconnected before updating _connectionStatus)
|
5000,
|
||||||
var status: ConnectionStatus
|
) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.RECONNECTING } }
|
||||||
val deadline = System.currentTimeMillis() + 2000
|
}
|
||||||
do {
|
|
||||||
status = HermesWsClient.connectionStatus.value
|
|
||||||
if (status == ConnectionStatus.RECONNECTING) break
|
|
||||||
Thread.sleep(50)
|
|
||||||
} while (System.currentTimeMillis() < deadline)
|
|
||||||
assertEquals(ConnectionStatus.RECONNECTING, status)
|
|
||||||
|
|
||||||
// The client should now attempt to reconnect after initial backoff (1000ms)
|
// The client should now attempt to reconnect after initial backoff (1000ms)
|
||||||
// Wait for the second connection to hit the server
|
// Wait for the second connection to hit the server
|
||||||
@@ -360,11 +323,8 @@ class HermesWsClientTest {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
val clientLatch = CountDownLatch(1)
|
|
||||||
HermesWsClient.onConnected = { clientLatch.countDown() }
|
|
||||||
|
|
||||||
HermesWsClient.connect()
|
HermesWsClient.connect()
|
||||||
assertTrue(clientLatch.await(5, TimeUnit.SECONDS))
|
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
|
||||||
|
|
||||||
// After connect, backoff should be back to initial
|
// After connect, backoff should be back to initial
|
||||||
val backoffField = HermesWsClient::class.java.getDeclaredField("currentBackoff")
|
val backoffField = HermesWsClient::class.java.getDeclaredField("currentBackoff")
|
||||||
@@ -394,11 +354,8 @@ class HermesWsClientTest {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
val clientLatch = CountDownLatch(1)
|
|
||||||
HermesWsClient.onConnected = { clientLatch.countDown() }
|
|
||||||
|
|
||||||
HermesWsClient.connect()
|
HermesWsClient.connect()
|
||||||
assertTrue(clientLatch.await(5, TimeUnit.SECONDS))
|
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
|
||||||
|
|
||||||
// Disconnect — this sets intentionalClose = true and cancels reconnect
|
// Disconnect — this sets intentionalClose = true and cancels reconnect
|
||||||
HermesWsClient.disconnect()
|
HermesWsClient.disconnect()
|
||||||
@@ -423,22 +380,14 @@ class HermesWsClientTest {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
val clientLatch = CountDownLatch(1)
|
|
||||||
val connectCount = intArrayOf(0)
|
|
||||||
HermesWsClient.onConnected = {
|
|
||||||
connectCount[0]++
|
|
||||||
clientLatch.countDown()
|
|
||||||
}
|
|
||||||
|
|
||||||
HermesWsClient.connect()
|
HermesWsClient.connect()
|
||||||
assertTrue(clientLatch.await(5, TimeUnit.SECONDS))
|
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
|
||||||
assertEquals(1, connectCount[0])
|
|
||||||
assertTrue(HermesWsClient.isConnected)
|
assertTrue(HermesWsClient.isConnected)
|
||||||
|
|
||||||
// Second connect call should be a no-op
|
// Second connect call should be a no-op
|
||||||
HermesWsClient.connect()
|
HermesWsClient.connect()
|
||||||
assertEquals(1, connectCount[0])
|
|
||||||
assertTrue(HermesWsClient.isConnected)
|
assertTrue(HermesWsClient.isConnected)
|
||||||
|
assertEquals(ConnectionStatus.CONNECTED, HermesWsClient.connectionStatus.value)
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
@@ -472,9 +421,7 @@ class HermesWsClientTest {
|
|||||||
assertEquals(ConnectionStatus.CONNECTING, status)
|
assertEquals(ConnectionStatus.CONNECTING, status)
|
||||||
|
|
||||||
// Wait for actual connection
|
// Wait for actual connection
|
||||||
val clientLatch = CountDownLatch(1)
|
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
|
||||||
HermesWsClient.onConnected = { clientLatch.countDown() }
|
|
||||||
assertTrue(clientLatch.await(5, TimeUnit.SECONDS))
|
|
||||||
assertEquals(ConnectionStatus.CONNECTED, HermesWsClient.connectionStatus.value)
|
assertEquals(ConnectionStatus.CONNECTED, HermesWsClient.connectionStatus.value)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -510,11 +457,9 @@ class HermesWsClientTest {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
val clientConnectedLatch = CountDownLatch(1)
|
|
||||||
HermesWsClient.onConnected = { clientConnectedLatch.countDown() }
|
|
||||||
HermesWsClient.connect()
|
HermesWsClient.connect()
|
||||||
assertTrue(connectLatch.await(5, TimeUnit.SECONDS))
|
assertTrue(connectLatch.await(5, TimeUnit.SECONDS))
|
||||||
assertTrue(clientConnectedLatch.await(5, TimeUnit.SECONDS))
|
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
|
||||||
|
|
||||||
// Disconnect (sets intentionalClose) — after this, reconnect should be prevented
|
// Disconnect (sets intentionalClose) — after this, reconnect should be prevented
|
||||||
HermesWsClient.disconnect()
|
HermesWsClient.disconnect()
|
||||||
|
|||||||
Reference in new issue
Block a user