yilsem/aiturk-mobil
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

perf(#1165): add onCapabilitiesChanged to NetworkMonitor for faster socket recovery (#1179)

Browse Source
This commit is contained in:
M57 authored and GitHub committed 2026-09-17 19:40:02 +03:00
1 parent 8c5a3a061f
commit 880adbef4f
4 files changed
+349 -11
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app/src/main/java/com/m57/hermescontrol/data/remote/NetworkMonitor.kt
+90
View File
@@ -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<Boolean>(extraBufferCapacity = Int.MAX_VALUE)
internal val networkChanges: SharedFlow<Boolean> = networkChangeEvents.asSharedFlow()
// Emitted when default network transports change without onLost/onAvailable (issue #1165).
private val transportChangeEvents = MutableSharedFlow<Unit>(extraBufferCapacity = Int.MAX_VALUE)
internal val transportChanges: SharedFlow<Unit> = transportChangeEvents.asSharedFlow()
private val callbackLock = Any()
private val callbackGeneration = AtomicInteger(0)
private var callbackRegistration: Pair<ConnectivityManager, ConnectivityManager.NetworkCallback>? = null
@@ -26,6 +33,9 @@ object NetworkMonitor {
@Volatile
private var defaultNetwork: Network? = null
@Volatile
private var currentTransports: Set<Int>? = 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<Int> {
if (capabilities == null) return emptySet()
val transports = mutableSetOf<Int>()
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
}
}
}
app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt
+32 -2
View File
@@ -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 =
app/src/test/java/com/m57/hermescontrol/data/remote/NetworkMonitorTest.kt
+143 -8
View File
@@ -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<Boolean>).value = true
NetworkMonitor.resetForTest()
}
private fun createCapabilities(vararg transports: Int): NetworkCapabilities {
val capabilities = mockk<NetworkCapabilities>()
every { capabilities.hasTransport(any()) } answers {
val transport = firstArg<Int>()
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<ConnectivityManager>()
val network = mockk<Network>()
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<ConnectivityManager>()
val network = mockk<Network>()
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<ConnectivityManager>()
val network = mockk<Network>()
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<ConnectivityManager>()
val defaultNet = mockk<Network>()
val otherNet = mockk<Network>()
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<ConnectivityManager>()
val defaultNet = mockk<Network>()
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<ConnectivityManager>()
val defaultNet = mockk<Network>()
val otherNet = mockk<Network>()
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,
)
}
}
app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt
+84 -1
View File
@@ -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)
}
}