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

@@ -46,7 +46,6 @@ 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() }
@@ -91,8 +90,7 @@ class KiloClientConnection<S>(
                 kiloRemoteInterface.complete(
                     KiloRemoteInterface(deferredParams, clientInterface)
                 )
-                connectedScope = params.scope
-                clientInterface.onConnectHandlers.invokeAll(connectedScope)
+                clientInterface.onConnectHandlers.invokeAll(params.scope)
                 onConnectedStateChanged?.invoke(true)
                 job.join()
 
@@ -101,7 +99,6 @@ 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() } }
@@ -119,4 +116,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,17 +19,16 @@ typealias KiloHandler<S> = KiloScope<S>.()->Unit
  *
  * - It registers common exceptions from [RemoteInterface] and kotlin/java `IllegalArgumentException` and
  *      `IllegalStateException`
- * - It provides [onConnected] and [onDisconnected] handlers
+ * - It provides [onConnected] handler
  *
  * 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 a handler list.
+     * Registers handler [f] for [onConnected] event, to the head or the end of handler list.
      *
      * @param addFirst if true, [f] will be added to the beginning of the list of handlers
      */
@@ -37,18 +36,6 @@ 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) }
@@ -60,3 +47,4 @@ open class KiloInterface<S> : LocalInterface<KiloScope<S>>() {
         registerError { IllegalArgumentException(it) }
     }
 }
+

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

@@ -53,7 +53,6 @@ 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
@@ -86,8 +85,7 @@ class KiloServerConnection<S>(
                 kiloRemoteInterface.complete(
                     KiloRemoteInterface(deferredParams, clientInterface)
                 )
-                connectedScope = p.scope
-                clientInterface.onConnectHandlers.invokeAll(connectedScope)
+                clientInterface.onConnectHandlers.invokeAll(p.scope)
             }
         }
 
@@ -95,12 +93,8 @@ class KiloServerConnection<S>(
             deferredTransport.complete(transport)
         kiloRemoteInterface.complete(KiloRemoteInterface(deferredParams,clientInterface))
         debug { "starting the transport"}
-        try {
-            transport.run()
-            debug { "server transport finished" }
-        } finally {
-            connectedScope?.let { clientInterface.onDisconnectHandlers.invokeAll(it) }
-        }
+        transport.run()
+        debug { "server transport finished" }
     }
 
     companion object {
@@ -114,4 +108,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,18 +53,16 @@ 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)
- * @param client optional caller-owned client. When omitted, a client with the
- * [WebSockets] plugin is created for this device and closed with it.
+ * @client use default [HttpClient], it installs [WebSockets] plugin
  */
 fun websocketTransportDevice(
     path: String,
     useTextFrames: Boolean = false,
-    client: HttpClient? = null,
+    client: HttpClient = HttpClient {
+        install(WebSockets)
+    },
 ): 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" }
@@ -84,7 +82,7 @@ fun websocketTransportDevice(
     globalLaunch {
         val log = LogTag("KC:${counter.incrementAndGet()}")
         try {
-            actualClient.webSocket({
+            client.webSocket({
                 url.protocol = u.protocol
                 url.host = u.host
                 url.port = u.port
@@ -150,8 +148,6 @@ fun websocketTransportDevice(
             else log.warning { "unexpected IO error $x" }
             runCatching { output.close() }
             runCatching { input.close() }
-        } finally {
-            if (ownsClient) actualClient.close()
         }
         log.info { "closing connection" }
     }
@@ -166,3 +162,4 @@ fun websocketTransportDevice(
     })
     return device
 }
+

src/commonTest/kotlin/TransportTest.kt

@@ -170,41 +170,6 @@ 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

@@ -1,29 +0,0 @@
-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() }
-}