Close owned WebSocket clients after reconnects

src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt

@@ -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() } }

src/commonMain/kotlin/net/sergeych/kiloparsec/KiloInterface.kt

@@ -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) }
     }
 }
-

src/commonMain/kotlin/net/sergeych/kiloparsec/KiloServerConnection.kt

@@ -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)
     }
-}
+}

src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/websocketClient.kt

@@ -53,16 +53,18 @@ fun <S> websocketClient(
 /**
  * Create kilopaarsec transport over websocket (ws or wss).
  * @param path websocket path (must start with ws:// or wss:// and contain a path part)
- * @client use default [HttpClient], it installs [WebSockets] plugin
+ * @param client optional caller-owned client. When omitted, a client with the
+ * [WebSockets] plugin is created for this device and closed with it.
  */
 fun websocketTransportDevice(
     path: String,
     useTextFrames: Boolean = false,
-    client: HttpClient = HttpClient {
-        install(WebSockets)
-    },
+    client: HttpClient? = null,
 ): Transport.Device {
 
+    val ownsClient = client == null
+    val actualClient = client ?: HttpClient { install(WebSockets) }
+
     val log = LogTag("WSTD")
     var u = Url(path)
     log.debug { "Creating websocket transport device at $u" }
@@ -82,7 +84,7 @@ fun websocketTransportDevice(
     globalLaunch {
         val log = LogTag("KC:${counter.incrementAndGet()}")
         try {
-            client.webSocket({
+            actualClient.webSocket({
                 url.protocol = u.protocol
                 url.host = u.host
                 url.port = u.port
@@ -148,6 +150,8 @@ fun websocketTransportDevice(
             else log.warning { "unexpected IO error $x" }
             runCatching { output.close() }
             runCatching { input.close() }
+        } finally {
+            if (ownsClient) actualClient.close()
         }
         log.info { "closing connection" }
     }
@@ -162,4 +166,3 @@ fun websocketTransportDevice(
     })
     return device
 }
-

src/commonTest/kotlin/TransportTest.kt

@@ -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

src/jvmTest/kotlin/net/sergeych/kiloparsec/adapter/WebsocketClientResourceTest.kt

@@ -0,0 +1,29 @@
+package net.sergeych.kiloparsec.adapter
+
+import java.nio.file.Files
+import java.nio.file.Path
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+import kotlin.test.assertTrue
+
+class WebsocketClientResourceTest {
+    @Test
+    fun failedReconnectsDoNotLeakFileDescriptors() = runBlocking {
+        val descriptors = Path.of("/proc/self/fd")
+        if (!Files.isDirectory(descriptors)) return@runBlocking
+
+        val client = websocketClient<Unit>("ws://127.0.0.1:1/kp")
+        try {
+            delay(2_500)
+            val before = descriptorCount(descriptors)
+            delay(7_000)
+            val after = descriptorCount(descriptors)
+            assertTrue(after <= before + 3, "file descriptors grew from $before to $after")
+        } finally {
+            client.close()
+        }
+    }
+
+    private fun descriptorCount(path: Path): Long = Files.list(path).use { it.count() }
+}