SergeychWorks/kiloparsec
1
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases 7 Wiki Activity

Add disconnect lifecycle handlers

Browse Source
This commit is contained in:
Sergey Chernov 2026-08-11 14:40:50 +04:00
parent 7bebbf7bd7
commit 19b349143d
4 changed files with 65 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt
View File

@ -46,6 +46,7 @@ class KiloClientConnection<S>(
suspend fun run(onConnectedStateChanged: ((Boolean) -> Unit)? = null) {
coroutineScope {
var job: Job? = null
var connectedScope: KiloScope<S>? = null
try {
// in parallel: keys and connection
val deferredKeyPair = async { SafeKeyExchange() }
@ -90,7 +91,8 @@ class KiloClientConnection<S>(
kiloRemoteInterface.complete(
KiloRemoteInterface(deferredParams, clientInterface)
)
clientInterface.onConnectHandlers.invokeAll(params.scope)
connectedScope = params.scope
clientInterface.onConnectHandlers.invokeAll(connectedScope)
onConnectedStateChanged?.invoke(true)
job.join()
@ -99,6 +101,7 @@ class KiloClientConnection<S>(
} catch (x: RemoteInterface.ClosedException) {
debug { "connection closed/refused by remote" }
} finally {
connectedScope?.let { clientInterface.onDisconnectHandlers.invokeAll(it) }
onConnectedStateChanged?.invoke(false)
job?.cancel()
device.apply { runCatching { close() } }
@ -116,4 +119,4 @@ class KiloClientConnection<S>(
}
internal fun <S>Collection<KiloHandler<S>>.invokeAll(scope: KiloScope<S>) =
forEach { runCatching { scope.it() } }
forEach { runCatching { scope.it() } }

18
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloInterface.kt
View File

@ -19,16 +19,17 @@ typealias KiloHandler<S> = KiloScope<S>.()->Unit
*
* - It registers common exceptions from [RemoteInterface] and kotlin/java `IllegalArgumentException` and
* `IllegalStateException`
* - It provides [onConnected] handler
* - It provides [onConnected] and [onDisconnected] handlers
*
* See [KiloServer] for usage sample.
*/
open class KiloInterface<S> : LocalInterface<KiloScope<S>>() {
internal val onConnectHandlers = mutableListOf<KiloHandler<S>>()
internal val onDisconnectHandlers = mutableListOf<KiloHandler<S>>()
/**
* Registers handler [f] for [onConnected] event, to the head or the end of handler list.
* Registers handler [f] for [onConnected] event, to the head or the end of a handler list.
*
* @param addFirst if true, [f] will be added to the beginning of the list of handlers
*/
@ -36,6 +37,18 @@ open class KiloInterface<S> : LocalInterface<KiloScope<S>>() {
if( addFirst ) onConnectHandlers.add(0, f) else onConnectHandlers += f
}
/**
* Registers handler [f] for [onDisconnected] event, to the head or the end of a handler list.
*
* It is called with the same connection [KiloScope] as [onConnected], after an established
* connection is closed.
*
* @param addFirst if true, [f] will be added to the beginning of the list of handlers
*/
fun onDisconnected(addFirst: Boolean = false, f: KiloScope<S>.()->Unit) {
if( addFirst ) onDisconnectHandlers.add(0, f) else onDisconnectHandlers += f
}
init {
registerError { RemoteInterface.UnknownCommand(it) }
registerError { RemoteInterface.InternalError(it) }
@ -47,4 +60,3 @@ open class KiloInterface<S> : LocalInterface<KiloScope<S>>() {
registerError { IllegalArgumentException(it) }
}
}

14
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloServerConnection.kt
View File

@ -53,6 +53,7 @@ class KiloServerConnection<S>(
suspend fun run() {
val deferredParams = CompletableDeferred<KiloParams<S>>()
val deferredTransport = CompletableDeferred<Transport<*>>()
var connectedScope: KiloScope<S>? = null
val l0Interface = KiloL0Interface(clientInterface, deferredParams).apply {
var params: KiloParams<S>? = null
@ -85,7 +86,8 @@ class KiloServerConnection<S>(
kiloRemoteInterface.complete(
KiloRemoteInterface(deferredParams, clientInterface)
)
clientInterface.onConnectHandlers.invokeAll(p.scope)
connectedScope = p.scope
clientInterface.onConnectHandlers.invokeAll(connectedScope)
}
}
@ -93,8 +95,12 @@ class KiloServerConnection<S>(
deferredTransport.complete(transport)
kiloRemoteInterface.complete(KiloRemoteInterface(deferredParams,clientInterface))
debug { "starting the transport"}
transport.run()
debug { "server transport finished" }
try {
transport.run()
debug { "server transport finished" }
} finally {
connectedScope?.let { clientInterface.onDisconnectHandlers.invokeAll(it) }
}
}
companion object {
@ -108,4 +114,4 @@ class KiloServerConnection<S>(
override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
kiloRemoteInterface.await().push(cmd, args)
}
}
}

35
src/commonTest/kotlin/TransportTest.kt
View File

@ -170,6 +170,41 @@ class TransportTest {
d2.close()
}
@Test
fun testDisconnectedHandlers() = runTest {
initCrypto()
val cmdPing by command<String, String>()
val (d1, d2) = createTestDevice()
val serverDisconnected = CompletableDeferred<String>()
val clientDisconnected = CompletableDeferred<String>()
val serverInterface = KiloInterface<String>().apply {
onDisconnected {
serverDisconnected.complete(session)
}
on(cmdPing) {
"pong! [$it]"
}
}
launch { KiloServerConnection(serverInterface, d1, "server session").run() }
val clientInterface = KiloInterface<String>().apply {
onDisconnected {
clientDisconnected.complete(session)
}
}
val client = KiloClientConnection(clientInterface, d2, "client session")
launch { client.run() }
assertEquals("pong! [hello]", client.call(cmdPing, "hello"))
d1.close()
d2.close()
assertEquals("server session", withTimeout(1000) { serverDisconnected.await() })
assertEquals("client session", withTimeout(1000) { clientDisconnected.await() })
}
class TestException(text: String) : Exception(text)
@Test