From 880adbef4fab6f419f1377ed70ce1798d55f0672 Mon Sep 17 00:00:00 2001 From: M57 <41645758+Hy4ri@users.noreply.github.com> Date: Thu, 17 Sep 2026 19:40:02 +0300 Subject: [PATCH] perf(#1165): add onCapabilitiesChanged to NetworkMonitor for faster socket recovery (#1179) --- .../data/remote/NetworkMonitor.kt | 90 +++++++++++ .../hermescontrol/data/ws/HermesWsClient.kt | 34 +++- .../data/remote/NetworkMonitorTest.kt | 151 +++++++++++++++++- .../data/ws/HermesWsClientTest.kt | 85 +++++++++- 4 files changed, 349 insertions(+), 11 deletions(-) diff --git a/app/src/main/java/com/m57/hermescontrol/data/remote/NetworkMonitor.kt b/app/src/main/java/com/m57/hermescontrol/data/remote/NetworkMonitor.kt index c0326e7b..66cce620 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/remote/NetworkMonitor.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/remote/NetworkMonitor.kt @@ -3,6 +3,9 @@ package com.m57.hermescontrol.data.remote import android.content.Context import android.net.ConnectivityManager import android.net.Network +import android.net.NetworkCapabilities +import android.os.Build +import androidx.annotation.VisibleForTesting import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.SharedFlow @@ -19,6 +22,10 @@ object NetworkMonitor { private val networkChangeEvents = MutableSharedFlow(extraBufferCapacity = Int.MAX_VALUE) internal val networkChanges: SharedFlow = networkChangeEvents.asSharedFlow() + // Emitted when default network transports change without onLost/onAvailable (issue #1165). + private val transportChangeEvents = MutableSharedFlow(extraBufferCapacity = Int.MAX_VALUE) + internal val transportChanges: SharedFlow = transportChangeEvents.asSharedFlow() + private val callbackLock = Any() private val callbackGeneration = AtomicInteger(0) private var callbackRegistration: Pair? = null @@ -26,6 +33,9 @@ object NetworkMonitor { @Volatile private var defaultNetwork: Network? = null + @Volatile + private var currentTransports: Set? = null + fun init(context: Context) { val cm = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager val callback = createCallback(cm) @@ -38,6 +48,10 @@ object NetworkMonitor { defaultNetwork = activeNetwork networkAvailable = activeNetwork != null _isConnected.value = networkAvailable + currentTransports = + runCatching { + activeNetwork?.let { cm.getNetworkCapabilities(it) }?.let { extractTransports(it) } + }.getOrNull() callbackRegistration.also { callbackRegistration = cm to callback } } if (networkChanged) networkChangeEvents.tryEmit(networkAvailable) @@ -56,6 +70,7 @@ object NetworkMonitor { if (callbackGeneration.get() != generation) return (defaultNetwork != network).also { defaultNetwork = network + currentTransports = null _isConnected.value = true } } @@ -71,11 +86,86 @@ object NetworkMonitor { (defaultNetwork != activeNetwork).also { defaultNetwork = activeNetwork networkAvailable = activeNetwork != null + currentTransports = null _isConnected.value = networkAvailable } } if (changed) networkChangeEvents.tryEmit(networkAvailable) } + + override fun onCapabilitiesChanged( + network: Network, + networkCapabilities: NetworkCapabilities, + ) { + val transportChanged: Boolean + synchronized(callbackLock) { + if (callbackGeneration.get() != generation || defaultNetwork != network) return + val newTransports = extractTransports(networkCapabilities) + val oldTransports = currentTransports + currentTransports = newTransports + transportChanged = oldTransports != null && oldTransports != newTransports + } + if (transportChanged) { + transportChangeEvents.tryEmit(Unit) + } + } + + override fun onLosing( + network: Network, + maxMsToLive: Int, + ) { + val isDefault: Boolean + synchronized(callbackLock) { + if (callbackGeneration.get() != generation || defaultNetwork != network) return + isDefault = true + } + if (isDefault) { + transportChangeEvents.tryEmit(Unit) + } + } + } + } + + internal fun extractTransports(capabilities: NetworkCapabilities?): Set { + if (capabilities == null) return emptySet() + val transports = mutableSetOf() + if (capabilities.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR)) { + transports.add(NetworkCapabilities.TRANSPORT_CELLULAR) + } + if (capabilities.hasTransport(NetworkCapabilities.TRANSPORT_WIFI)) { + transports.add(NetworkCapabilities.TRANSPORT_WIFI) + } + if (capabilities.hasTransport(NetworkCapabilities.TRANSPORT_BLUETOOTH)) { + transports.add(NetworkCapabilities.TRANSPORT_BLUETOOTH) + } + if (capabilities.hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET)) { + transports.add(NetworkCapabilities.TRANSPORT_ETHERNET) + } + if (capabilities.hasTransport(NetworkCapabilities.TRANSPORT_VPN)) { + transports.add(NetworkCapabilities.TRANSPORT_VPN) + } + if (capabilities.hasTransport(NetworkCapabilities.TRANSPORT_WIFI_AWARE)) { + transports.add(NetworkCapabilities.TRANSPORT_WIFI_AWARE) + } + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O_MR1 && + capabilities.hasTransport(NetworkCapabilities.TRANSPORT_LOWPAN) + ) { + transports.add(NetworkCapabilities.TRANSPORT_LOWPAN) + } + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.S && + capabilities.hasTransport(NetworkCapabilities.TRANSPORT_USB) + ) { + transports.add(NetworkCapabilities.TRANSPORT_USB) + } + return transports + } + + @VisibleForTesting + internal fun resetForTest() { + synchronized(callbackLock) { + defaultNetwork = null + currentTransports = null + _isConnected.value = true } } } 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 883ffd3a..ffc4ff17 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 @@ -106,10 +106,11 @@ object HermesWsClient { /** * Liveness thresholds aligned with reference clients (desktop/web): - * Ping every 15s using gateway.ping; 45s inbound silence deadline. + * Ping every 15s using gateway.ping; 30s inbound silence deadline (issue #1165) + * as mobile secondary safety net for faster half-open socket recovery. */ private const val HEARTBEAT_INTERVAL_MS = 15_000L - private const val STALE_THRESHOLD_MS = 45_000L + private const val STALE_THRESHOLD_MS = 30_000L private const val LIVENESS_PROBE_TIMEOUT_MS = 5_000L // ── Internal state (all access through synchronized / atomic) ──────── @@ -369,6 +370,7 @@ object HermesWsClient { val isConnected: Boolean get() = connected.get() private var foregroundProbeJob: Job? = null + private var transportProbeJob: Job? = null fun setAppForeground(foreground: Boolean) { appInForeground.set(foreground) @@ -398,6 +400,29 @@ object HermesWsClient { } } + @VisibleForTesting + internal fun probeLivenessOnTransportChange( + onFailureAction: () -> Unit = { + synchronized(outboundLock) { + if (connected.get()) webSocket?.cancel() + } + }, + ) { + if (!connected.get()) return + transportProbeJob?.cancel() + transportProbeJob = + wsScope.launch { + Log.d(TAG, "Network transport changed — probing WebSocket liveness") + val alive = runCatching { ping(LIVENESS_PROBE_TIMEOUT_MS) }.isSuccess + if (alive) { + Log.d(TAG, "WebSocket liveness probe succeeded on new transport") + return@launch + } + Log.w(TAG, "Transport change liveness probe failed — cancelling socket") + onFailureAction() + } + } + fun acquireExternalActivityConnectionLease() { externalActivityConnectionLease.set(true) } @@ -428,6 +453,11 @@ object HermesWsClient { reconnectForNetworkChange(networkAvailable) } } + wsScope.launch { + NetworkMonitor.transportChanges.collect { + probeLivenessOnTransportChange() + } + } } var socketToCancel: WebSocket? = null val shouldOpen = diff --git a/app/src/test/java/com/m57/hermescontrol/data/remote/NetworkMonitorTest.kt b/app/src/test/java/com/m57/hermescontrol/data/remote/NetworkMonitorTest.kt index 18a4b7a8..d9863fcf 100644 --- a/app/src/test/java/com/m57/hermescontrol/data/remote/NetworkMonitorTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/data/remote/NetworkMonitorTest.kt @@ -2,10 +2,10 @@ package com.m57.hermescontrol.data.remote import android.net.ConnectivityManager import android.net.Network +import android.net.NetworkCapabilities import app.cash.turbine.test import io.mockk.every import io.mockk.mockk -import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.test.runTest import org.junit.Assert.assertEquals import org.junit.Assert.assertTrue @@ -15,13 +15,16 @@ import org.junit.Test class NetworkMonitorTest { @Before fun setUp() { - val defaultNetworkField = NetworkMonitor::class.java.getDeclaredField("defaultNetwork") - defaultNetworkField.isAccessible = true - defaultNetworkField.set(NetworkMonitor, null) - val connectedField = NetworkMonitor::class.java.getDeclaredField("_isConnected") - connectedField.isAccessible = true - @Suppress("UNCHECKED_CAST") - (connectedField.get(NetworkMonitor) as MutableStateFlow).value = true + NetworkMonitor.resetForTest() + } + + private fun createCapabilities(vararg transports: Int): NetworkCapabilities { + val capabilities = mockk() + every { capabilities.hasTransport(any()) } answers { + val transport = firstArg() + transports.contains(transport) + } + return capabilities } @Test @@ -94,4 +97,136 @@ class NetworkMonitorTest { expectNoEvents() } } + + @Test + fun `initial capabilities callback after onAvailable sets baseline and does not emit transport change`() = + runTest { + val connectivityManager = mockk() + val network = mockk() + val wifiCaps = createCapabilities(NetworkCapabilities.TRANSPORT_WIFI) + val callback = NetworkMonitor.createCallback(connectivityManager) + + NetworkMonitor.transportChanges.test { + callback.onAvailable(network) + callback.onCapabilitiesChanged(network, wifiCaps) + + expectNoEvents() + } + } + + @Test + fun `capabilities change with different transports emits transport change`() = + runTest { + val connectivityManager = mockk() + val network = mockk() + val wifiCaps = createCapabilities(NetworkCapabilities.TRANSPORT_WIFI) + val vpnWifiCaps = createCapabilities(NetworkCapabilities.TRANSPORT_WIFI, NetworkCapabilities.TRANSPORT_VPN) + val callback = NetworkMonitor.createCallback(connectivityManager) + + NetworkMonitor.transportChanges.test { + callback.onAvailable(network) + callback.onCapabilitiesChanged(network, wifiCaps) + expectNoEvents() + + callback.onCapabilitiesChanged(network, vpnWifiCaps) + assertEquals(Unit, awaitItem()) + expectNoEvents() + } + } + + @Test + fun `capabilities change with identical transports does not emit transport change`() = + runTest { + val connectivityManager = mockk() + val network = mockk() + val wifiCaps1 = createCapabilities(NetworkCapabilities.TRANSPORT_WIFI) + val wifiCaps2 = createCapabilities(NetworkCapabilities.TRANSPORT_WIFI) + val callback = NetworkMonitor.createCallback(connectivityManager) + + NetworkMonitor.transportChanges.test { + callback.onAvailable(network) + callback.onCapabilitiesChanged(network, wifiCaps1) + expectNoEvents() + + callback.onCapabilitiesChanged(network, wifiCaps2) + expectNoEvents() + } + } + + @Test + fun `capabilities change on non-default network is ignored`() = + runTest { + val connectivityManager = mockk() + val defaultNet = mockk() + val otherNet = mockk() + val wifiCaps = createCapabilities(NetworkCapabilities.TRANSPORT_WIFI) + val cellularCaps = createCapabilities(NetworkCapabilities.TRANSPORT_CELLULAR) + val callback = NetworkMonitor.createCallback(connectivityManager) + + NetworkMonitor.transportChanges.test { + callback.onAvailable(defaultNet) + callback.onCapabilitiesChanged(defaultNet, wifiCaps) + expectNoEvents() + + callback.onCapabilitiesChanged(otherNet, cellularCaps) + expectNoEvents() + } + } + + @Test + fun `onLosing on default network emits transport change`() = + runTest { + val connectivityManager = mockk() + val defaultNet = mockk() + val callback = NetworkMonitor.createCallback(connectivityManager) + + NetworkMonitor.transportChanges.test { + callback.onAvailable(defaultNet) + callback.onLosing(defaultNet, 3000) + + assertEquals(Unit, awaitItem()) + expectNoEvents() + } + } + + @Test + fun `onLosing on non-default network is ignored`() = + runTest { + val connectivityManager = mockk() + val defaultNet = mockk() + val otherNet = mockk() + val callback = NetworkMonitor.createCallback(connectivityManager) + + NetworkMonitor.transportChanges.test { + callback.onAvailable(defaultNet) + callback.onLosing(otherNet, 3000) + + expectNoEvents() + } + } + + @Test + fun `extractTransports identifies all supported transport types`() { + val caps = + createCapabilities( + NetworkCapabilities.TRANSPORT_CELLULAR, + NetworkCapabilities.TRANSPORT_WIFI, + NetworkCapabilities.TRANSPORT_ETHERNET, + NetworkCapabilities.TRANSPORT_VPN, + NetworkCapabilities.TRANSPORT_BLUETOOTH, + NetworkCapabilities.TRANSPORT_WIFI_AWARE, + ) + val transports = NetworkMonitor.extractTransports(caps) + assertEquals( + setOf( + NetworkCapabilities.TRANSPORT_CELLULAR, + NetworkCapabilities.TRANSPORT_WIFI, + NetworkCapabilities.TRANSPORT_ETHERNET, + NetworkCapabilities.TRANSPORT_VPN, + NetworkCapabilities.TRANSPORT_BLUETOOTH, + NetworkCapabilities.TRANSPORT_WIFI_AWARE, + ), + transports, + ) + } } diff --git a/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt b/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt index d933bf9f..4829bd1b 100644 --- a/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt @@ -180,7 +180,9 @@ class HermesWsClientTest { response: okhttp3.Response, ) { webSocket.send( - """{"jsonrpc":"2.0","id":"42","result":{"open_requests":[{"id":"srq-secret","method":"secret","params":{"session_id":"session-1","env_var":"API_KEY","prompt":"Enter API key"}}]}}""", + """{"jsonrpc":"2.0","id":"42","result":{"open_requests":[{"id":"srq-secret",""" + + """"method":"secret","params":{"session_id":"session-1","env_var":"API_KEY",""" + + """"prompt":"Enter API key"}}]}}""", ) } }, @@ -2013,4 +2015,85 @@ class HermesWsClientTest { assertEquals(latency, HermesWsClient.lastLatencyMs.value) assertTrue(HermesWsClient.lastPongTimestamp > 0) } + + @Test + fun testProbeLivenessOnTransportChangeCancelsSocketWhenPingFails() { + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true) + + val failureLatch = CountDownLatch(1) + HermesWsClient.probeLivenessOnTransportChange( + onFailureAction = { + failureLatch.countDown() + }, + ) + + assertTrue("Expected failure action to be invoked on failed ping", failureLatch.await(6, TimeUnit.SECONDS)) + } + + @Test + fun testProbeLivenessOnTransportChangeSkipsWhenDisconnected() { + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + + var failureInvoked = false + HermesWsClient.probeLivenessOnTransportChange( + onFailureAction = { + failureInvoked = true + }, + ) + + Thread.sleep(100) + assertFalse(failureInvoked) + } + + @Test + fun testProbeLivenessOnTransportChangeSucceedsWhenServerAnswers() { + val serverLatch = CountDownLatch(1) + + mockWebServer.enqueue( + MockResponse().withWebSocketUpgrade( + object : WebSocketListener() { + override fun onOpen( + webSocket: WebSocket, + response: okhttp3.Response, + ) { + serverLatch.countDown() + } + + override fun onMessage( + webSocket: WebSocket, + text: String, + ) { + if (text.contains(""""method":"gateway.ping"""") || text.contains(""""method":"ping"""")) { + val id = Regex(""""id":"([^"]+)"""").find(text)?.groupValues?.get(1) ?: "1" + webSocket.send( + """{"jsonrpc":"2.0","id":"$id","result":{"pong":true,"timestamp":1700000000.0}}""", + ) + } + } + }, + ), + ) + + HermesWsClient.connect() + runBlocking { + withTimeout(5000) { + HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } + } + } + assertTrue(serverLatch.await(5, TimeUnit.SECONDS)) + + var failureInvoked = false + HermesWsClient.probeLivenessOnTransportChange( + onFailureAction = { + failureInvoked = true + }, + ) + + Thread.sleep(500) + assertFalse("Expected socket to remain connected when ping succeeds", failureInvoked) + } }