diff --git a/SDK/Android/necto-core/src/main/kotlin/im/toss/necto/sdk/NectoSdkRuntime.kt b/SDK/Android/necto-core/src/main/kotlin/im/toss/necto/sdk/NectoSdkRuntime.kt index 6665272..96f9ae7 100644 --- a/SDK/Android/necto-core/src/main/kotlin/im/toss/necto/sdk/NectoSdkRuntime.kt +++ b/SDK/Android/necto-core/src/main/kotlin/im/toss/necto/sdk/NectoSdkRuntime.kt @@ -24,6 +24,7 @@ import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay import kotlinx.coroutines.ensureActive @@ -49,6 +50,8 @@ internal class NectoSdkRuntime( /** Held for the life of the connection. */ private var session: NectoMessageSession? = null + /** Sessions still saying hello. One takes over [session] once the host accepts it. */ + private val handshaking = HashSet() private val statusFlow = MutableStateFlow(NectoSDK.Status.Stopped) private val registered = ArrayList() private val registeringIDs = HashSet() @@ -148,7 +151,10 @@ internal class NectoSdkRuntime( } } - /** Accepts connections until the listener goes away. One session at a time. */ + /** + * Accepts connections until the listener goes away. Each is handled on its own, so a + * newer host can take over from a connection that went quiet without being closed. + */ private suspend fun serve(listen: () -> NectoListener, generation: Any) { val listener = try { listen() @@ -170,13 +176,15 @@ internal class NectoSdkRuntime( } try { - while (currentCoroutineContext().isActive) { - val session = listener.accept() - if (!currentCoroutineContext().isActive) { - session.close() - break + coroutineScope { + while (isActive) { + val session = listener.accept() + if (!isActive) { + session.close() + break + } + launch { handle(session) } } - handle(session) } } catch (error: CancellationException) { throw error @@ -193,6 +201,8 @@ internal class NectoSdkRuntime( fun stop() { val closing = synchronized(lock) { val closing = Triple(serveJob, listener, session) + val greeting = handshaking.toList() + handshaking.clear() serveJob = null generation = null listener = null @@ -201,13 +211,14 @@ internal class NectoSdkRuntime( val pending = requests.values.map { it.second } requests.clear() setStatusLocked(NectoSDK.Status.Stopped) - closing to pending + Triple(closing, pending, greeting) } - val (jobs, pending) = closing + val (jobs, pending, greeting) = closing jobs.first?.cancel() pending.forEach { it.cancel() } // Closing is what wakes a blocked accept or read. jobs.third?.close() + greeting.forEach { it.close() } jobs.second?.stop() } @@ -224,23 +235,19 @@ internal class NectoSdkRuntime( handle(session) } - /** Says hello, tells the host what this app offers, then reads until it goes away. */ + /** + * Says hello, tells the host what this app offers, then reads until it goes away. + * + * A session replaces the current one only once the host accepts its handshake. A phone + * can hold a connection open after its cable is pulled, so the host's next one has to be + * able to take over, and one that never answers must not end a session that works. + */ private suspend fun handle(session: NectoMessageSession) { val active = currentCoroutineContextIsActive() - val previous = synchronized(lock) { - if (!active) return@synchronized null - val previous = this.session to requests.values.map { it.second } - requests.clear() - stopRegistrationUpdatesLocked() - this.session = session - previous - } - if (previous == null) { + if (!synchronized(lock) { active && handshaking.add(session) }) { session.close() return } - previous.first?.close() - previous.second.forEach { it.cancel() } try { val identity = identity @@ -259,14 +266,27 @@ internal class NectoSdkRuntime( NectoHandshakeAck.fromJson(session.receiveJson(MAXIMUM_ACK_BYTES)) } if (!ack.accepted) { - setStatus(NectoSDK.Status.Failed("Necto refused the connection: ${ack.rejection?.rawValue ?: "unknown"}"), session) + setHandshakeStatus(NectoSDK.Status.Failed("Necto refused the connection: ${ack.rejection?.rawValue ?: "unknown"}"), session) session.close() return } - if (synchronized(lock) { this.session !== session } || !currentCoroutineContextIsActive()) { + + val stillActive = currentCoroutineContextIsActive() + val previous = synchronized(lock) { + // Gone from the set when the runtime stopped while the host was answering. + if (!handshaking.remove(session) || !stillActive) return@synchronized null + val previous = this.session to requests.values.map { it.second } + requests.clear() + stopRegistrationUpdatesLocked() + this.session = session + previous + } + if (previous == null) { session.close() return } + previous.first?.close() + previous.second.forEach { it.cancel() } setStatus(NectoSDK.Status.Connected(hello.appBundleID), session) startRegistrationUpdates(session) @@ -274,13 +294,21 @@ internal class NectoSdkRuntime( } catch (error: CancellationException) { throw error } catch (error: Exception) { - setStatus(NectoSDK.Status.Failed(error.message ?: error.toString()), session) + setHandshakeStatus(NectoSDK.Status.Failed(error.message ?: error.toString()), session) session.close() } finally { + synchronized(lock) { handshaking.remove(session) } disconnect(session) } } + /** Reports on [session] unless another one is connected, which keeps its own status. */ + private fun setHandshakeStatus(value: NectoSDK.Status, session: NectoMessageSession) { + synchronized(lock) { + if (this.session === session || (this.session == null && session in handshaking)) setStatusLocked(value) + } + } + /** Past the deadline the session is closed, which fails the pending read or write. */ private suspend fun handshake(session: NectoMessageSession, block: suspend () -> T): T { val watchdog = scope.launch { diff --git a/SDK/Android/necto-core/src/test/kotlin/im/toss/necto/RuntimeTests.kt b/SDK/Android/necto-core/src/test/kotlin/im/toss/necto/RuntimeTests.kt index d41a6bc..b148107 100644 --- a/SDK/Android/necto-core/src/test/kotlin/im/toss/necto/RuntimeTests.kt +++ b/SDK/Android/necto-core/src/test/kotlin/im/toss/necto/RuntimeTests.kt @@ -310,6 +310,39 @@ class RuntimeTests { } } + @Test + fun aNewerHostReplacesOneThatWentQuiet() = runBlocking { + withTimeout(10_000) { + runtime.register(testPlugin) + // A host whose cable was pulled: the connection stays open and says nothing. + val quiet = connect() + val next = FakeHost(listen()) + next.handshake() + quiet.awaitClosed() + next.invoke("replaced", "test.echo") + assertEquals("replaced", next.result().requestID) + assertTrue(runtime.status.value is NectoSDK.Status.Connected) + } + } + + @Test + fun aHostStillSayingHelloLeavesTheConnectedOneAlone() = runBlocking { + withTimeout(10_000) { + runtime.register(testPlugin) + val host = connect() + val pending = FakeHost(listen()) + pending.receive() + host.invoke("kept", "test.echo") + assertEquals("kept", host.result().requestID) + // A refused handshake does not replace it either, nor change what it reports. + pending.session.send(NectoHandshakeAck(accepted = false, rejection = NectoHandshakeAck.Rejection.MISSING_APP_IDENTITY).toJson()) + pending.awaitClosed() + host.invoke("still", "test.echo") + assertEquals("still", host.result().requestID) + assertTrue(runtime.status.value is NectoSDK.Status.Connected) + } + } + // MARK: A host that misbehaves /**