diff --git a/app/build.gradle.kts b/app/build.gradle.kts index f5adcc9..dfb1006 100644 --- a/app/build.gradle.kts +++ b/app/build.gradle.kts @@ -106,6 +106,7 @@ dependencies { // Testing testImplementation("junit:junit:4.13.2") + testImplementation("io.mockk:mockk:1.13.12") testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.10.1") testImplementation("com.squareup.okhttp3:mockwebserver:4.12.0") testImplementation("app.cash.turbine:turbine:1.2.0") diff --git a/app/src/main/AndroidManifest.xml b/app/src/main/AndroidManifest.xml index de4254e..4af185e 100644 --- a/app/src/main/AndroidManifest.xml +++ b/app/src/main/AndroidManifest.xml @@ -5,6 +5,7 @@ (ConnectionState.Disconnected) val connectionState: StateFlow = _connectionState.asStateFlow() - private val _events = MutableSharedFlow(replay = 1, extraBufferCapacity = 64) + private val _events = MutableSharedFlow(extraBufferCapacity = 64) val events: SharedFlow = _events.asSharedFlow() private fun nextId(): String = "a${reqCounter.incrementAndGet()}" diff --git a/app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt b/app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt index a139a97..b3192c9 100644 --- a/app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt +++ b/app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt @@ -26,13 +26,14 @@ class HermesConnectionManager( val tokenVault: TokenVault, val restClient: HermesRestClient = HermesRestClient(), val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default), - val runtimeFactory: (HermesHost) -> HermesHostRuntime = { host -> + val runtimeFactory: (CoroutineScope, HermesHost) -> HermesHostRuntime = { parentScope, host -> + val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[kotlinx.coroutines.Job]) + Dispatchers.Default) HermesHostRuntime( initialHost = host, restClient = restClient, - gatewayClient = JsonRpcGatewayClient(scope = scope), + gatewayClient = JsonRpcGatewayClient(scope = childScope), tokenVault = tokenVault, - scope = scope + scope = childScope ) } ) { @@ -44,7 +45,7 @@ class HermesConnectionManager( private val _activeHostId = MutableStateFlow(null) val activeHostId: StateFlow = _activeHostId.asStateFlow() - private val _allEvents = MutableSharedFlow(replay = 1, extraBufferCapacity = 128) + private val _allEvents = MutableSharedFlow(extraBufferCapacity = 128) val allEvents: SharedFlow = _allEvents.asSharedFlow() init { @@ -89,7 +90,7 @@ class HermesConnectionManager( fun getOrCreateRuntime(host: HermesHost): HermesHostRuntime { return runtimes.computeIfAbsent(host.id) { - val rt = runtimeFactory(host) + val rt = runtimeFactory(scope, host) // Forward events scope.launch { rt.events.collect { event -> diff --git a/app/src/test/java/app/hermes/mobile/core/network/EventStreamReplayTest.kt b/app/src/test/java/app/hermes/mobile/core/network/EventStreamReplayTest.kt new file mode 100644 index 0000000..88c2e82 --- /dev/null +++ b/app/src/test/java/app/hermes/mobile/core/network/EventStreamReplayTest.kt @@ -0,0 +1,44 @@ +package app.hermes.mobile.core.network + +import app.hermes.mobile.core.model.GatewayEvent +import app.hermes.mobile.core.model.GatewayEvent.MessageDeltaEvent +import kotlinx.coroutines.flow.firstOrNull +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.UnconfinedTestDispatcher +import kotlinx.coroutines.test.runTest +import kotlinx.coroutines.test.runCurrent +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Test + +class EventStreamReplayTest { + + @Test + fun testNoReplay() = runTest { + val client = JsonRpcGatewayClient(scope = this) + + // Emit first event BEFORE subscribing + client.handleIncomingMessage("{\"jsonrpc\":\"2.0\",\"method\":\"event\",\"params\":{\"type\":\"message.delta\",\"session_id\":\"s1\",\"payload\":{\"message_id\":\"m1\",\"delta\":\"Hello\"}}}") + runCurrent() + + val events = mutableListOf() + val job = launch { + client.events.collect { events.add(it) } + } + runCurrent() + + // Should not have received the first event + assertTrue("Events should be empty since we subscribed after the first emission", events.isEmpty()) + + // Emit second event AFTER subscribing + client.handleIncomingMessage("{\"jsonrpc\":\"2.0\",\"method\":\"event\",\"params\":{\"type\":\"message.delta\",\"session_id\":\"s1\",\"payload\":{\"message_id\":\"m1\",\"delta\":\" World\"}}}") + runCurrent() + + assertEquals("Should have exactly 1 event", 1, events.size) + val event = events[0] as GatewayEvent.MessageDeltaEvent + assertEquals(" World", event.delta) + + job.cancel() + } +} diff --git a/app/src/test/java/app/hermes/mobile/core/network/JsonRpcGatewayClientTest.kt b/app/src/test/java/app/hermes/mobile/core/network/JsonRpcGatewayClientTest.kt index 6bd99a5..b81066c 100644 --- a/app/src/test/java/app/hermes/mobile/core/network/JsonRpcGatewayClientTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/network/JsonRpcGatewayClientTest.kt @@ -6,6 +6,9 @@ import app.hermes.mobile.core.model.RuntimeSessionId import kotlinx.coroutines.flow.first import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeout +import kotlinx.coroutines.async +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.async import okhttp3.Response import okhttp3.WebSocket import okhttp3.WebSocketListener @@ -123,12 +126,18 @@ class JsonRpcGatewayClientTest { ) val wsUrl = "ws://${server.hostName}:${server.port}/api/ws" + + // Start subscription BEFORE connecting, so we don't miss the event + val eventDeferred = async(start = CoroutineStart.UNDISPATCHED) { + withTimeout(5000) { + client.events.first { it is GatewayEvent.MessageDeltaEvent } + } + } + client.connect(wsUrl, allowCleartext = true) client.awaitGatewayReady(5000) - val event = withTimeout(5000) { - client.events.first { it is GatewayEvent.MessageDeltaEvent } - } + val event = eventDeferred.await() assertTrue(event is GatewayEvent.MessageDeltaEvent) val deltaEvent = event as GatewayEvent.MessageDeltaEvent diff --git a/app/src/test/java/app/hermes/mobile/core/repository/ApprovalRoutingTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/ApprovalRoutingTest.kt index 28c92d4..8f22b2b 100644 --- a/app/src/test/java/app/hermes/mobile/core/repository/ApprovalRoutingTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/repository/ApprovalRoutingTest.kt @@ -59,7 +59,17 @@ class ApprovalRoutingTest { connectionManager = HermesConnectionManager( hostDao = hostDao, tokenVault = tokenVault, - scope = CoroutineScope(testDispatcher) + scope = CoroutineScope(testDispatcher), + runtimeFactory = { parentScope, host -> + val childScope = CoroutineScope(kotlinx.coroutines.SupervisorJob(parentScope.coroutineContext[kotlinx.coroutines.Job]) + testDispatcher) + HermesHostRuntime( + initialHost = host, + restClient = app.hermes.mobile.core.network.HermesRestClient(), + gatewayClient = JsonRpcGatewayClient(scope = childScope), + tokenVault = tokenVault, + scope = childScope + ) + } ) sessionRepo = UnifiedSessionRepository( diff --git a/app/src/test/java/app/hermes/mobile/core/repository/MultiHostConcurrencyExecutionTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/MultiHostConcurrencyExecutionTest.kt index 5e68732..6fbfda9 100644 --- a/app/src/test/java/app/hermes/mobile/core/repository/MultiHostConcurrencyExecutionTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/repository/MultiHostConcurrencyExecutionTest.kt @@ -3,6 +3,7 @@ package app.hermes.mobile.core.repository import app.hermes.mobile.core.model.* import app.hermes.mobile.core.network.ConnectionState import app.hermes.mobile.core.network.JsonRpcGatewayClient +import app.hermes.mobile.core.network.HermesRestClient import app.hermes.mobile.core.runtime.HermesConnectionManager import app.hermes.mobile.core.runtime.HermesHostRuntime import app.hermes.mobile.core.security.InMemoryTokenVault @@ -53,7 +54,17 @@ class MultiHostConcurrencyExecutionTest { connectionManager = HermesConnectionManager( hostDao = hostDao, tokenVault = tokenVault, - scope = CoroutineScope(testDispatcher) + scope = CoroutineScope(testDispatcher), + runtimeFactory = { parentScope, host -> + val childScope = CoroutineScope(kotlinx.coroutines.SupervisorJob(parentScope.coroutineContext[kotlinx.coroutines.Job]) + testDispatcher) + app.hermes.mobile.core.runtime.HermesHostRuntime( + initialHost = host, + restClient = app.hermes.mobile.core.network.HermesRestClient(), + gatewayClient = app.hermes.mobile.core.network.JsonRpcGatewayClient(scope = childScope), + tokenVault = tokenVault, + scope = childScope + ) + } ) sessionRepo = UnifiedSessionRepository( @@ -664,8 +675,10 @@ class MultiHostConcurrencyExecutionTest { scope = CoroutineScope(Dispatchers.Default) ) - val session = testRepo.createUnifiedSession(title = "Reconnect Test", initialHostId = host1Id) - val runtime = testConnectionManager.getRuntime(host1Id)!! + val host1 = HermesHost(id = HermesHostId("host-windows"), displayName = "Server", baseUrl = wsUrl, allowCleartext = true, lastKnownStatus = HostStatus.ONLINE) + testConnectionManager.addHost(host1) + val session = testRepo.createUnifiedSession(title = "Reconnect Test", initialHostId = HermesHostId("host-windows")) + val runtime = testConnectionManager.getRuntime(HermesHostId("host-windows"))!! // Initial connect runtime.connect() diff --git a/app/src/test/java/app/hermes/mobile/core/repository/UnifiedSessionRepositoryTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/UnifiedSessionRepositoryTest.kt index ea4b7ff..fe3b1df 100644 --- a/app/src/test/java/app/hermes/mobile/core/repository/UnifiedSessionRepositoryTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/repository/UnifiedSessionRepositoryTest.kt @@ -5,6 +5,7 @@ import app.hermes.mobile.core.network.ConnectionState import app.hermes.mobile.core.network.HermesRestClient import app.hermes.mobile.core.network.JsonRpcGatewayClient import app.hermes.mobile.core.runtime.HermesConnectionManager +import app.hermes.mobile.core.runtime.HermesHostRuntime import app.hermes.mobile.core.security.InMemoryTokenVault import app.hermes.mobile.core.storage.FakeHostDao import app.hermes.mobile.core.storage.FakeUnifiedSessionDao @@ -49,7 +50,17 @@ class UnifiedSessionRepositoryTest { connectionManager = HermesConnectionManager( hostDao = hostDao, tokenVault = tokenVault, - scope = CoroutineScope(testDispatcher) + scope = CoroutineScope(testDispatcher), + runtimeFactory = { parentScope, host -> + val childScope = CoroutineScope(kotlinx.coroutines.SupervisorJob(parentScope.coroutineContext[kotlinx.coroutines.Job]) + testDispatcher) + app.hermes.mobile.core.runtime.HermesHostRuntime( + initialHost = host, + restClient = app.hermes.mobile.core.network.HermesRestClient(), + gatewayClient = app.hermes.mobile.core.network.JsonRpcGatewayClient(scope = childScope), + tokenVault = tokenVault, + scope = childScope + ) + } ) repository = UnifiedSessionRepository( diff --git a/app/src/test/java/app/hermes/mobile/core/runtime/AppLifecycleTest.kt b/app/src/test/java/app/hermes/mobile/core/runtime/AppLifecycleTest.kt new file mode 100644 index 0000000..3d47f62 --- /dev/null +++ b/app/src/test/java/app/hermes/mobile/core/runtime/AppLifecycleTest.kt @@ -0,0 +1,55 @@ +package app.hermes.mobile.core.runtime + +import app.hermes.mobile.core.model.HermesHost +import app.hermes.mobile.core.model.HermesHostId +import app.hermes.mobile.core.model.HostStatus +import app.hermes.mobile.core.network.HermesRestClient +import app.hermes.mobile.core.security.InMemoryTokenVault +import app.hermes.mobile.core.storage.FakeHostDao +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertSame +import org.junit.Test + +class AppLifecycleTest { + + @Test + fun testSingleHostIdCorrespondsToAtMostOneLiveRuntime() = runTest { + val hostDao = FakeHostDao() + val tokenVault = InMemoryTokenVault() + val restClient = HermesRestClient() + + val connectionManager = HermesConnectionManager( + hostDao = hostDao, + tokenVault = tokenVault, + restClient = restClient, + scope = backgroundScope, + runtimeFactory = { parentScope, host -> + val childScope = kotlinx.coroutines.CoroutineScope(kotlinx.coroutines.SupervisorJob(parentScope.coroutineContext[kotlinx.coroutines.Job]) + kotlinx.coroutines.test.StandardTestDispatcher(testScheduler)) + app.hermes.mobile.core.runtime.HermesHostRuntime( + initialHost = host, + restClient = restClient, + gatewayClient = app.hermes.mobile.core.network.JsonRpcGatewayClient(scope = childScope), + tokenVault = tokenVault, + scope = childScope + ) + } + ) + + val host = HermesHost( + id = HermesHostId("h1"), + displayName = "Host 1", + baseUrl = "http://host1.com", + allowCleartext = true, + enabled = true, + lastSeenAt = 0L, + lastKnownStatus = HostStatus.OFFLINE + ) + + val runtime1 = connectionManager.getOrCreateRuntime(host) + val runtime2 = connectionManager.getOrCreateRuntime(host) + val runtime3 = connectionManager.getRuntime(host.id) + + assertSame(runtime1, runtime2) + assertSame(runtime1, runtime3) + } +}