release 0.2.6: ktor-based async sockets for all native targets

.idea/misc.xml

@@ -1,4 +1,3 @@
-<?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
   <component name="ExternalStorageConfigurationManager" enabled="true" />
   <component name="FrameworkDetectionExcludesConfiguration">

README.md

@@ -1,31 +1,40 @@
 # Kiloparsec
 
-The new generation of __PARanoid SECurity__ protocol, advanced, faster, more secure. It also allows connecting any "block device" transport to the same local interface. Out if the box it
+The new generation of __PARanoid SECurity__ protocol, advanced, faster, more secure. It also allows connecting any "
+block device" transport to the same local interface. Out if the box it
 provides the following transports:
 
 | name           | JVM | JS | native   |
-|----------------|-----|----|--------|
+|----------------|-----|----|----------|
-| TCP/IP server  | *   |    |        |
+| TCP/IP server  | ✓   |    | β @0.2.6 |
-| TCP/IP client  | *   |    |        |
+| TCP/IP client  | ✓   |    | β @0.2.6 |
-| Websock server | *   |    |        |
+| Websock server | ✓   |    |          |
-| Websock client | *   | *  | *      |
+| Websock client | ✓   | ✓  | ✓        |
 
-At the moment we're working on supporting TCP/IP on most native targets. This feature is planned to rach public beta in August and production in early september 2024.
+### Supported native targets
 
+- iosArm64, iosX64
+- macosArm64, macosArm64
+- linxArm64, linuxX64
+
+### Non-native targets
+
+- JS (browser and nodeJS)
+- JVM (android, macos, windows, linx, everywhere where JRE is installed)
 
 ## TCP/IP transport
 
-It is the fastest. JVM implementation uses nio2 async sockets and optimizes TCP socket to play
+It is the fastest based on async socket implementation of ktor client. It works everywhere but JS target as
-well with blocks (smart NO_DELAY mode). It is multiplatform, nut lacks of async TCP/IP support
+there is currently no widely adopted sockets for browser javascript.
-on natvic targetm this is where I need help having little time. I'd prefer to use something asyn like UV on native targets.
-
-I know no existing way to implement it in KotlinJS for the modern browsers.
 
 ## Websock server
 
-While it is much slower than pure TCP, it is still faster than any http-based transport. It uses binary frames based on the Ktor server framework to easily integrate with web services. We recommend using it instead of a classic HTTP API as it beats it in terms of speed and server load even with HTTP/2.
+While it is much slower than pure TCP, it is still faster than any http-based transport. It uses binary frames based on
+the Ktor server framework to easily integrate with web services. We recommend using it instead of a classic HTTP API as
+it beats it in terms of speed and server load even with HTTP/2.
 
-We recommend to create the `KiloInterface<S>` instance and connect it to the websock and tcp servers in real applications to get easy access from anywhere.
+We recommend to create the `KiloInterface<S>` instance and connect it to the websock and tcp servers in real
+applications to get easy access from anywhere.
 
 # Usage
 
@@ -50,7 +59,7 @@ It could be, depending on your project structure, something like:
 ```kotlin
 val commonMain by getting {
     dependencies {
-        api("net.sergeych:kiloparsec:0.2.4")
+        api("net.sergeych:kiloparsec:0.2.6")
     }
 }
 ```
@@ -136,6 +145,7 @@ val ns: NettyApplicationEngine = embeddedServer(Netty, port = 8080, host = "0.0.
 
 
 ~~~
+
 # Details
 
 It is not compatible with parsec family and no more based on an Universa crypto library. To better fit
@@ -144,9 +154,11 @@ and every connection (while parsec caches session keys to avoid time-consuming k
 keys cryptography for session is shifted to use ed25519 curves which are supposed to provide agreeable strength with
 enough speed to protect every connection with a unique new keys. Also, we completely get rid of SHA2.
 
-Kiloparsec also uses a denser binary format [bipack](https://gitea.sergeych.net/SergeychWorks/mp_bintools), no more key-values, 
+Kiloparsec also uses a denser binary format [bipack](https://gitea.sergeych.net/SergeychWorks/mp_bintools), no more
+key-values,
 which reveals much less on the inner data structure, providing advanced
-typed RPC interfaces with kotlinx.serialization. There is also Rust implementation [bipack_ru](https://gitea.sergeych.net/DiWAN/bipack_ru). 
+typed RPC interfaces with kotlinx.serialization. There is also Rust
+implementation [bipack_ru](https://gitea.sergeych.net/DiWAN/bipack_ru).
 The architecture allows connecting same functional interfaces to several various type channels at once.
 
 Also, the difference from parsecs is that there are no more unencrypted layer commands available to users.
@@ -163,9 +175,10 @@ Integrated tools to prevent MITM attacks include also non-transferred independen
 independently on the ends and is never transferred with the network. Comparing it somehow (visually, with QR code, etc)
 could add a very robust guarantee of the connection safety and ingenuity.
 
-Kiloparsec has built-in completely asynchronous (coroutine based top-down) transport layer based on TCP (JVM only as for now) and the same async Websocket-based transport based on KTOR. Websocket client is multiplatform, though the server is JVM only insofar.
+Kiloparsec has built-in completely asynchronous (coroutine based top-down) transport layer based on TCP (JVM only as for
+now) and the same async Websocket-based transport based on KTOR. Websocket client is multiplatform, though the server is
+JVM only insofar.
 
 # Licensing
 
-When the project reaches some stable state, it will be available under Apache or MIT license. I promise ;) I do
+Currently, you need to obtain a license from https://8-rays.dev or Sergey Chernov.
-write opensource ;)

build.gradle.kts

@@ -1,4 +1,3 @@
-
 plugins {
     kotlin("multiplatform") version "2.0.0"
     id("org.jetbrains.kotlin.plugin.serialization") version "2.0.0"
@@ -6,7 +5,7 @@ plugins {
 }
 
 group = "net.sergeych"
-version = "0.2.4"
+version = "0.2.6"
 
 repositories {
     mavenCentral()
@@ -20,20 +19,8 @@ kotlin {
     js {
         browser {
         }
+        nodejs()
     }
-//    val hostOs = System.getProperty("os.name")
-//    val isArm64 = System.getProperty("os.arch") == "aarch64"
-//    val isMingwX64 = hostOs.startsWith("Windows")
-//    @Suppress("UNUSED_VARIABLE")
-//    val nativeTarget = when {
-//        hostOs == "Mac OS X" && isArm64 -> macosArm64("native")
-//        hostOs == "Mac OS X" && !isArm64 -> macosX64("native")
-//        hostOs == "Linux" && isArm64 -> linuxArm64("native")
-//        hostOs == "Linux" && !isArm64 -> linuxX64("native")
-//        isMingwX64 -> mingwX64("native")
-//        else -> throw GradleException("Host OS is not supported in Kotlin/Native.")
-//    }
-
     macosArm64()
     iosX64()
     iosArm64()
@@ -55,18 +42,16 @@ kotlin {
             dependencies {
                 implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3")
                 implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.7.1")
-
-//                api("com.ionspin.kotlin:bignum:0.3.9")
                 api("io.ktor:ktor-client-core:$ktor_version")
                 api("net.sergeych:crypto2:0.4.2")
             }
         }
-//        val ktorSocketMain by creating {
+        val ktorSocketMain by creating {
-//            dependsOn(commonMain)
+            dependsOn(commonMain)
-//            dependencies {
+            dependencies {
-//                implementation("io.ktor:ktor-network:$ktor_version")
+                implementation("io.ktor:ktor-network:$ktor_version")
-//            }
+            }
-//        }
+        }
         val commonTest by getting {
             dependencies {
                 implementation(kotlin("test"))
@@ -74,6 +59,9 @@ kotlin {
                 implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.8.1")
             }
         }
+        val ktorSocketTest by creating {
+            dependsOn(commonTest)
+        }
         val jvmMain by getting {
             dependencies {
                 implementation("io.ktor:ktor-server-core:$ktor_version")
@@ -82,19 +70,51 @@ kotlin {
                 implementation("io.ktor:ktor-server-netty:$ktor_version")
                 api("io.ktor:ktor-client-cio:$ktor_version")
             }
-//            dependsOn(ktorSocketMain)
+            dependsOn(ktorSocketMain)
         }
-        val jvmTest by getting
+        val jvmTest by getting {
+            dependsOn(ktorSocketTest)
+        }
+
         val jsMain by getting {
             dependencies {
                 implementation("io.ktor:ktor-client-js:$ktor_version")
             }
         }
         val jsTest by getting
-
+        val macosArm64Main by getting {
-//        for (pm in listOf(linuxMain, macosMain, iosMain, mingwMain))
+            dependsOn(ktorSocketMain)
-//            pm { dependsOn(ktorSocketMain) }
+        }
-
+        val macosArm64Test by getting {
+            dependsOn(ktorSocketTest)
+        }
+        val macosX64Main by getting {
+            dependsOn(ktorSocketMain)
+        }
+        val iosX64Main by getting {
+            dependsOn(ktorSocketMain)
+        }
+        val iosX64Test by getting {
+            dependsOn(ktorSocketTest)
+        }
+        val iosArm64Main by getting {
+            dependsOn(ktorSocketMain)
+        }
+        val iosArm64Test by getting {
+            dependsOn(ktorSocketTest)
+        }
+        val linuxArm64Main by getting {
+            dependsOn(ktorSocketMain)
+        }
+        val linuxArm64Test by getting {
+            dependsOn(ktorSocketTest)
+        }
+        val linuxX64Main by getting {
+            dependsOn(ktorSocketMain)
+        }
+        val linuxX64Test by getting {
+            dependsOn(ktorSocketTest)
+        }
     }
 
     publishing {

gradle.properties

@@ -1 +1,2 @@
 kotlin.code.style=official
+kotlin.mpp.applyDefaultHierarchyTemplate=false

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

@@ -1,56 +1,56 @@
 package net.sergeych.kiloparsec.adapter
 
-import kotlinx.coroutines.channels.ReceiveChannel
-import kotlinx.coroutines.flow.Flow
-
 /**
- * Multiplatform implementation of an internet address.
+ * Multiplatform internet address.
- * Notice to implementors. It must provide correct and effective [equals] and [hashCode].
  */
-interface NetworkAddress {
+data class NetworkAddress(
-    val host: String
+    val host: String,
     val port: Int
+) {
+    override fun toString(): String {
+        return "$host:$port"
     }
-
-/**
- * Multiplatform datagram abstraction
- */
-interface Datagram {
-    /**
-     * Received message
-     */
-    val message: UByteArray
-
-    /**
-     * Address from where the message was sent
-     */
-    val address: NetworkAddress
 }
-
+//
-@OptIn(ExperimentalStdlibApi::class)
+///**
-interface DatagramConnector: AutoCloseable {
+// * Multiplatform datagram abstraction
-
+// */
-    val incoming: ReceiveChannel<Datagram>
+//interface Datagram {
-    suspend fun send(message: UByteArray, networkAddress: NetworkAddress)
+//    /**
-    @Suppress("unused")
+//     * Received message
-    suspend fun send(message: UByteArray, datagramAddress: String) {
+//     */
-        send(message, datagramAddress.toNetworkAddress())
+//    val message: UByteArray
-    }
+//
-
+//    /**
-    suspend fun send(message: UByteArray,host: String,port: Int) =
+//     * Address from where the message was sent
-        send(message, NetworkAddress(host,port))
+//     */
-    override fun close()
+//    val address: NetworkAddress
-}
+//}
-
+//
-expect fun NetworkAddress(host: String,port: Int): NetworkAddress
+//@OptIn(ExperimentalStdlibApi::class)
-
+//interface DatagramConnector: AutoCloseable {
-fun String.toNetworkAddress() : NetworkAddress {
+//
-    val (host, port) = this.split(":").map { it.trim()}
+//    val incoming: ReceiveChannel<Datagram>
-    return NetworkAddress(host, port.toInt())
+//    suspend fun send(message: UByteArray, networkAddress: NetworkAddress)
-}
+//    @Suppress("unused")
-
+//    suspend fun send(message: UByteArray, datagramAddress: String) {
-expect fun acceptTcpDevice(port: Int): Flow<InetTransportDevice>
+//        send(message, datagramAddress.toNetworkAddress())
-
+//    }
-expect suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice
+//
-
+//    suspend fun send(message: UByteArray,host: String,port: Int) =
-suspend fun connectTcpDevice(address: String) = connectTcpDevice(address.toNetworkAddress())
+//        send(message, NetworkAddress(host,port))
+//    override fun close()
+//}
+//
+//expect fun NetworkAddress(host: String,port: Int): NetworkAddress
+//
+//fun String.toNetworkAddress() : NetworkAddress {
+//    val (host, port) = this.split(":").map { it.trim()}
+//    return NetworkAddress(host, port.toInt())
+//}
+//
+//expect fun acceptTcpDevice(port: Int): Flow<InetTransportDevice>
+//
+//expect suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice
+//
+//suspend fun connectTcpDevice(address: String) = connectTcpDevice(address.toNetworkAddress())

src/jsMain/kotlin/net/sergeych/kiloparsec/adapter/DatagramProvider.js.kt

@@ -1,2 +0,0 @@
-package net.sergeych.kiloparsec.adapter
-

src/jsMain/kotlin/net/sergeych/kiloparsec/adapter/NetworkProvider.js.kt

@@ -1,15 +0,0 @@
-package net.sergeych.kiloparsec.adapter
-
-import kotlinx.coroutines.flow.Flow
-
-actual fun NetworkAddress(host: String, port: Int): NetworkAddress {
-    TODO("Not yet implemented")
-}
-
-actual fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
-    TODO("Not yet implemented")
-}
-
-actual suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
-    TODO("Not yet implemented")
-}

src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/ContinuationHandler.kt

@@ -1,20 +0,0 @@
-package net.sergeych.kiloparsec.adapter
-
-import java.nio.channels.CompletionHandler
-import kotlin.coroutines.Continuation
-import kotlin.coroutines.resume
-import kotlin.coroutines.resumeWithException
-
-open class ContinuationHandler<T> : CompletionHandler<T, Continuation<T>> {
-    override fun completed(result: T, attachment: Continuation<T>) {
-        attachment.resume(result)
-    }
-
-    override fun failed(exc: Throwable, attachment: Continuation<T>) {
-        attachment.resumeWithException(exc)
-    }
-}
-
-object VoidCompletionHandler: ContinuationHandler<Void>()
-
-object IntCompletionHandler: ContinuationHandler<Int>()

src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/NetworkProvider.jvm.kt

@@ -1,49 +0,0 @@
-package net.sergeych.kiloparsec.adapter
-
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.suspendCancellableCoroutine
-import kotlinx.coroutines.withContext
-import java.net.InetAddress
-import java.net.InetSocketAddress
-import java.nio.channels.AsynchronousServerSocketChannel
-import java.nio.channels.AsynchronousSocketChannel
-import kotlin.coroutines.suspendCoroutine
-
-actual fun NetworkAddress(host: String, port: Int): NetworkAddress =
-    JvmNetworkAddress(InetAddress.getByName(host), port)
-
-actual fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
-    return flow {
-        val socket = withContext(Dispatchers.IO) {
-            AsynchronousServerSocketChannel.open().also {
-                it.bind(InetSocketAddress(port))
-            }
-        }
-        while (true) {
-            println("0 --- $port")
-            val connectedSocket = suspendCancellableCoroutine { continuation ->
-                continuation.invokeOnCancellation {
-                    socket.close()
-                }
-                socket.accept(continuation, ContinuationHandler())
-            }
-            println("1 ---")
-            emit(asyncSocketToDevice(connectedSocket))
-        }
-    }
-}
-
-@Suppress("unused")
-suspend fun connectTcpDevice(host: String, port: Int) = connectTcpDevice(NetworkAddress(host,port))
-actual suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
-    address as JvmNetworkAddress
-    val socket = withContext(Dispatchers.IO) {
-        AsynchronousSocketChannel.open()
-    }
-    suspendCoroutine { cont ->
-        socket.connect(address.socketAddress, cont, VoidCompletionHandler)
-    }
-    return asyncSocketToDevice(socket)
-}

src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/UdpServer.kt

@@ -1,118 +0,0 @@
-package net.sergeych.kiloparsec.adapter
-
-import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.BufferOverflow
-import kotlinx.coroutines.channels.Channel
-import net.sergeych.mp_logger.LogTag
-import net.sergeych.mp_logger.exception
-import net.sergeych.mp_logger.info
-import net.sergeych.mp_logger.warning
-import java.net.*
-import java.util.concurrent.atomic.AtomicInteger
-
-private val counter = AtomicInteger(0)
-
-class JvmNetworkAddress(val inetAddress: InetAddress, override val port: Int) : NetworkAddress {
-    override val host: String by lazy { inetAddress.canonicalHostName }
-    override fun equals(other: Any?): Boolean {
-        if (this === other) return true
-        if (other !is JvmNetworkAddress) return false
-
-        if (inetAddress != other.inetAddress) return false
-        if (port != other.port) return false
-
-        return true
-    }
-
-    val socketAddress: SocketAddress by lazy { InetSocketAddress(inetAddress,port) }
-
-    override fun hashCode(): Int {
-        var result = inetAddress.hashCode()
-        result = 31 * result + port
-        return result
-    }
-
-    override fun toString(): String = "$host:$port"
-}
-
-class UdpDatagram(override val message: UByteArray, val inetAddress: InetAddress, val port: Int) : Datagram {
-
-    override val address: NetworkAddress by lazy {
-        JvmNetworkAddress(inetAddress, port)
-    }
-
-}
-
-
-@OptIn(DelicateCoroutinesApi::class)
-class UdpServer(val port: Int) :
-    DatagramConnector, LogTag("UDPS:${counter.incrementAndGet()}") {
-    private var isClosed = false
-
-
-    private val deferredSocket = CompletableDeferred<DatagramSocket>()
-    private var job: Job? = null
-
-    private suspend fun start() = try {
-        coroutineScope {
-            val socket = DatagramSocket(port)
-            val buffer = ByteArray(16384)
-            val packet = DatagramPacket(buffer, buffer.size)
-            deferredSocket.complete(socket)
-            while (isActive && !isClosed) {
-                try {
-                    socket.receive(packet)
-                    val data = packet.data.sliceArray(0..<packet.length)
-                    val datagram = UdpDatagram(data.toUByteArray(), packet.address, packet.port)
-                    if (!channel.trySend(datagram).isSuccess) {
-                        warning { "packet lost!" }
-                        // and we cause overflow that overwrites the oldest
-                        channel.send(datagram)
-                    }
-                } catch (e: Exception) {
-                    if (!isClosed)
-                        e.printStackTrace()
-                    throw e
-                }
-            }
-            info { "closing socket and reception loop" }
-
-        }
-    } catch (_: CancellationException) {
-        info { "server is closed" }
-    } catch (t: Throwable) {
-        exception { "unexpected end of server" to t }
-    }
-
-    init {
-        job = GlobalScope.launch(Dispatchers.IO) {
-            start()
-        }
-    }
-
-    override fun close() {
-        if (!isClosed) {
-            if (deferredSocket.isCompleted) {
-                runCatching {
-                    deferredSocket.getCompleted().close()
-                }
-            }
-            isClosed = true
-            job?.cancel(); job = null
-        }
-    }
-
-    private val channel = Channel<Datagram>(2048, BufferOverflow.DROP_OLDEST)
-    override val incoming = channel
-
-    override suspend fun send(message: UByteArray, networkAddress: NetworkAddress) {
-        networkAddress as JvmNetworkAddress
-        withContext(Dispatchers.IO) {
-            val packet = DatagramPacket(
-                message.toByteArray(), message.size,
-                networkAddress.inetAddress, networkAddress.port
-            )
-            deferredSocket.await().send(packet)
-        }
-    }
-}

src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/asyncSocketToDevice.kt

@@ -1,155 +0,0 @@
-package net.sergeych.kiloparsec.adapter
-
-import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.channels.ClosedReceiveChannelException
-import kotlinx.coroutines.flow.MutableStateFlow
-import net.sergeych.crypto2.Contrail
-import net.sergeych.crypto2.encodeVarUnsigned
-import net.sergeych.crypto2.readVarUnsigned
-import net.sergeych.kiloparsec.RemoteInterface
-import net.sergeych.kiloparsec.Transport
-import net.sergeych.mp_logger.LogTag
-import net.sergeych.mp_logger.warning
-import net.sergeych.mp_tools.globalLaunch
-import net.sergeych.tools.waitFor
-import java.net.InetSocketAddress
-import java.net.StandardSocketOptions.TCP_NODELAY
-import java.nio.ByteBuffer
-import java.nio.channels.AsynchronousSocketChannel
-import kotlin.coroutines.cancellation.CancellationException
-import kotlin.coroutines.suspendCoroutine
-
-private val log = LogTag("ASTD")
-
-/**
- * Prepend block with its size, varint-encoded
- */
-private fun encode(block: UByteArray): ByteArray {
-    val c = Contrail.create(block)
-    return (encodeVarUnsigned(c.size.toUInt()) + c).toByteArray()
-}
-
-/**
- * Convert asynchronous socket to a [Transport.Device] using non-blocking nio,
- * in a coroutine-effective manner. Note that it runs coroutines to read/write
- * to the socket in a global scope.These are closed when transport is closed
- *  or the socket is closed, for example, by network failure.
- */
-suspend fun asyncSocketToDevice(socket: AsynchronousSocketChannel): InetTransportDevice {
-    val deferredDevice = CompletableDeferred<InetTransportDevice>()
-    globalLaunch {
-        coroutineScope {
-            val sendQueueEmpty = MutableStateFlow(true)
-            val receiving = MutableStateFlow(false)
-            // We're in block mode, every block we send worth immediate sending, we do not
-            // send partial blocks, so:
-            socket.setOption(TCP_NODELAY, true)
-
-            // socket input is to be parsed for blocks, so we receive bytes
-            // and decode them to blocks
-            val input = Channel<UByte>(1024)
-            val inputBlocks = Channel<UByteArray>()
-            // output is blocks, so we sent transformed, framed blocks:
-            val outputBlocks = Channel<UByteArray>()
-
-            fun stop() {
-                kotlin.runCatching { inputBlocks.close(RemoteInterface.ClosedException()) }
-                kotlin.runCatching { outputBlocks.close() }
-                socket.close()
-                cancel()
-            }
-
-
-            // copy incoming data from the socket to input channel:
-            launch {
-                val data = ByteArray(1024)
-                val inb = ByteBuffer.wrap(data)
-                kotlin.runCatching {
-                    while (isActive) {
-                        inb.position(0)
-                        val size: Int = suspendCoroutine { continuation ->
-                            socket.read(inb, continuation, IntCompletionHandler)
-                        }
-                        if (size < 0) stop()
-                        else {
-//                        println("recvd:\n${data.sliceArray(0..<size).toDump()}\n------------------")
-                            for (i in 0..<size) input.send(data[i].toUByte())
-                        }
-                    }
-                }
-            }
-
-            // copy from output to socket:
-            launch {
-                try {
-                    while (isActive) {
-                        // wait for the first block to send
-                        sendQueueEmpty.value = outputBlocks.isEmpty
-                        var data = encode(outputBlocks.receive())
-
-                        // now we're sending, so queue state is sending:
-                        sendQueueEmpty.value = false
-
-                        // if there are more, take them all (NO_DELAY optimization)
-                        while (!outputBlocks.isEmpty)
-                            data += encode(outputBlocks.receive())
-
-                        // now send it all together:
-                        val outBuff = ByteBuffer.wrap(data)
-                        val cnt = suspendCoroutine { continuation ->
-                            socket.write(outBuff, continuation, IntCompletionHandler)
-                        }
-                        // be sure it was all sent
-                        if (outBuff.position() != data.size || cnt != data.size) {
-                            throw RuntimeException("unexpected partial write")
-                        }
-                    }
-                    // in the case of just breaking out of the loop:
-                    sendQueueEmpty.value = true
-                } catch (_: ClosedReceiveChannelException) {
-                    stop()
-                }
-            }
-            // transport device copes with blocks:
-            // decode blocks from a byte channel read from the socket:
-            launch {
-                try {
-                    while (isActive) {
-                        receiving.value = !input.isEmpty
-                        val size = readVarUnsigned(input)
-                        receiving.value = true
-                        if (size == 0u) log.warning { "zero size block is ignored!" }
-                        else {
-                            val block = UByteArray(size.toInt())
-                            for (i in 0..<size.toInt()) {
-                                block[i] = input.receive()
-                            }
-                            Contrail.unpack(block)?.let { inputBlocks.send(it) }
-                                ?: log.warning { "skipping bad block ${block.size} bytes" }
-                        }
-                    }
-                } catch (_: CancellationException) {
-                } catch (_: ClosedReceiveChannelException) {
-                    stop()
-                }
-                receiving.value = false
-            }
-
-            val addr = socket.remoteAddress as InetSocketAddress
-            deferredDevice.complete(
-                InetTransportDevice(inputBlocks, outputBlocks, JvmNetworkAddress(addr.address, addr.port), {
-                    yield()
-                    // wait until all received data are parsed, but not too long
-                    withTimeoutOrNull(500) {
-                        receiving.waitFor { !it }
-                    }
-                    // then stop it
-                    stop()
-                })
-            )
-        }
-        globalLaunch { socket.close() }
-    }
-    return deferredDevice.await()
-}

src/jvmTest/kotlin/net/sergeych/kiloparsec/ClientTest.kt

@@ -6,18 +6,17 @@ import io.ktor.server.netty.*
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runTest
 import net.sergeych.crypto2.initCrypto
-import net.sergeych.kiloparsec.adapter.acceptTcpDevice
-import net.sergeych.kiloparsec.adapter.connectTcpDevice
 import net.sergeych.kiloparsec.adapter.setupWebsocketServer
 import net.sergeych.kiloparsec.adapter.websocketClient
 import net.sergeych.mp_logger.Log
 import java.net.InetAddress
-import kotlin.test.*
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
 
 class ClientTest {
 
-    class TestException : Exception("test1")
-
     @Test
     fun testAddresses() {
         println(InetAddress.getLocalHost())
@@ -25,59 +24,7 @@ class ClientTest {
         println(InetAddress.getByName("127.0.0.1"))
         println(InetAddress.getByName("mail.ru"))
     }
-    @Test
-    fun testClient() = runTest {
-        initCrypto()
-        Log.connectConsole(Log.Level.DEBUG)
-        data class Session(
-            var data: String
-        )
 
-        val cmdSave by command<String,Unit>()
-        val cmdLoad by command<Unit,String>()
-        val cmdDrop by command<Unit,Unit>()
-        val cmdException by command<Unit,Unit>()
-
-        val cli = KiloInterface<Session>().apply {
-            registerError { TestException() }
-            onConnected { session.data = "start" }
-            on(cmdSave) { session.data = it }
-            on(cmdLoad) {
-                session.data
-            }
-            on(cmdException) {
-                throw TestException()
-            }
-            on(cmdDrop) {
-                throw LocalInterface.BreakConnectionException()
-            }
-        }
-        val server = KiloServer(cli, acceptTcpDevice(17101)) {
-            Session("unknown")
-        }
-
-        val client = KiloClient<Unit>() {
-            addErrors(cli)
-            connect { connectTcpDevice("localhost:17101") }
-        }
-        println(client.call(cmdLoad))
-
-        assertEquals("start", client.call(cmdLoad))
-        client.call(cmdSave, "foobar")
-        assertEquals("foobar", client.call(cmdLoad))
-
-        val res = kotlin.runCatching {  client.call(cmdException) }
-        println(res.exceptionOrNull())
-        assertIs<TestException>(res.exceptionOrNull())
-        assertEquals("foobar", client.call(cmdLoad))
-
-        assertThrows<RemoteInterface.ClosedException> { client.call(cmdDrop) }
-
-        // reconnect?
-        assertEquals("start", client.call(cmdLoad))
-
-        server.close()
-    }
 
     @Test
     fun webSocketTest() = runTest {

src/jvmTest/kotlin/net/sergeych/kiloparsec/adapters/NetworkTest.kt

@@ -1,100 +0,0 @@
-package net.sergeych.kiloparsec.adapters
-
-import kotlinx.coroutines.*
-import kotlinx.coroutines.test.runTest
-import net.sergeych.crypto2.initCrypto
-import net.sergeych.kiloparsec.adapter.UdpServer
-import net.sergeych.kiloparsec.adapter.acceptTcpDevice
-import net.sergeych.kiloparsec.adapter.connectTcpDevice
-import net.sergeych.kiloparsec.adapter.toNetworkAddress
-import net.sergeych.kiloparsec.decodeFromUByteArray
-import net.sergeych.kiloparsec.encodeToUByteArray
-import net.sergeych.mp_logger.Log
-import net.sergeych.mp_logger.LogTag
-import net.sergeych.synctools.ProtectedOp
-import net.sergeych.synctools.invoke
-import kotlin.test.Test
-import kotlin.test.assertContains
-import kotlin.test.assertEquals
-
-class NetworkTest {
-
-    @Test
-    fun udpProviderTest() = runTest {
-        Log.connectConsole(Log.Level.DEBUG)
-        val s1 = UdpServer(17120)
-        val s2 = UdpServer(17121)
-        s1.send("Hello".encodeToUByteArray(), "localhost", 17121)
-        val d1 = s2.incoming.receive()
-        assertEquals(d1.address.port, 17120)
-        assertEquals("Hello", d1.message.toByteArray().decodeToString())
-        s1.send("world".encodeToUByteArray(), d1.address)
-        assertEquals("world", s1.incoming.receive().message.toByteArray().decodeToString())
-        //        println("s1: ${s1.bindAddress()}")
-
-    }
-
-    @Test
-    fun tcpAsyncConnectionTest() = runTest {
-        initCrypto()
-        Log.connectConsole(Log.Level.DEBUG)
-
-        coroutineScope {
-            val serverFlow = acceptTcpDevice(17171)
-            val op = ProtectedOp()
-            var pills = setOf<String>()
-            val j = launch {
-                println("serf")
-                serverFlow.collect { device ->
-                    println("serf 0")
-                    launch {
-                        println("serf 1")
-                        device.output.send("Hello, world!".encodeToUByteArray())
-                        device.output.send("Great".encodeToUByteArray())
-                        while (true) {
-                            val x = device.input.receive().decodeFromUByteArray()
-                            if (x.startsWith("die")) {
-                                op.invoke {
-                                    pills += x
-                                }
-                                cancel()
-                            }
-                            else
-                                println("ignoring unexpected input: $x")
-                        }
-                    }
-                }
-            }
-            yield()
-            run {
-                try {
-                    println("pre-con")
-                    val s = connectTcpDevice("127.0.1.1:17171".toNetworkAddress())
-                    println("2")
-                    assertEquals("Hello, world!", s.input.receive().decodeFromUByteArray())
-                    assertEquals("Great", s.input.receive().decodeFromUByteArray())
-                    s.output.send("Goodbye".encodeToUByteArray())
-                    s.output.send("die1".encodeToUByteArray())
-                    s.close()
-                }
-                catch(t: Throwable) {
-                    t.printStackTrace()
-                    throw t
-                }
-            }
-            println("pre-con2")
-            val s1 = connectTcpDevice("127.0.1.1:17171".toNetworkAddress())
-            assertEquals("Hello, world!", s1.input.receive().decodeFromUByteArray())
-            assertEquals("Great", s1.input.receive().decodeFromUByteArray())
-            s1.output.send("die2".encodeToUByteArray())
-            s1.close()
-
-            // check that channels were flushed prior to closed:
-            assertContains(pills, "die1")
-            assertContains(pills, "die2")
-
-            // Check that server jobs are closed
-            j.cancelAndJoin()
-        }
-    }
-}

src/ktorSocketMain/kotlin/adapter/socketClient.kt

@@ -2,42 +2,196 @@ package net.sergeych.kiloparsec.adapter
 
 import io.ktor.network.selector.*
 import io.ktor.network.sockets.*
-import io.ktor.util.network.*
 import io.ktor.utils.io.*
-import io.ktor.utils.io.core.*
 import kotlinx.coroutines.*
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ClosedReceiveChannelException
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.datetime.Clock
 import net.sergeych.kiloparsec.AsyncVarint
-import net.sergeych.mp_logger.LogTag
+import net.sergeych.kiloparsec.LocalInterface
-import kotlin.io.use
+import net.sergeych.mp_logger.*
+import net.sergeych.mp_tools.globalLaunch
+import net.sergeych.tools.AtomicCounter
+import net.sergeych.tools.AtomicValue
+import kotlin.time.Duration.Companion.seconds
 
-//class SocketNetworkAddress(override val host: String, override val port: Int) : NetworkAddress
+private val logCounter = AtomicCounter(0)
-//
+
-//actual fun NetworkAddress(host: String, port: Int): NetworkAddress = SocketNetworkAddress(host, port)
+class ProtocolException(text: String, cause: Throwable? = null) : RuntimeException(text, cause)
-//
+
-//fun acceptTcpSocketDevice(port: Int): Flow<InetTransportDevice> {
+const val MAX_TCP_BLOCK_SIZE = 16776216
-//    val selectorManager = SelectorManager(Dispatchers.IO)
+val PING_INACTIVITY_TIME = 30.seconds
-//    val serverSocket = aSocket(selectorManager).tcp().bind("127.0.0.1", port)
+
-//    val log = LogTag("TCPS${}")
+fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
-//    return flow {
+    val selectorManager = SelectorManager(Dispatchers.IO)
-//        serverSocket.accept().use { sock ->
+    val serverSocket = aSocket(selectorManager).tcp().bind("127.0.0.1", port)
-//            val closed = CompletableDeferred<Boolean>()
+    return flow {
-//            val scope = coroutineScope {
+        while (true) {
-//                val networkAddress = sock.remoteAddress.toJavaAddress().let { NetworkAddress(it.hostname, it.port) }
+            serverSocket.accept().let { sock ->
-//                val inputBlocks = Channel<UByteArray>(4096)
+                emit(inetTransportDevice(sock, "srv"))
-//                sock.launch {
+            }
-//                    val sockInput = sock.openReadChannel()
+        }
-//                    while (isActive && sock.isActive) {
+    }
-//                        try {
+}
-//                            val size = AsyncVarint.decodeUnsigned(sockInput)
+
-//                        } catch (e: Exception) {
+suspend fun connectTcpDevice(address: String) = connectTcpDevice(address.toNetworkAddress())
-//
+
-//                        }
+suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
-//                    }
+    val selectorManager = SelectorManager(Dispatchers.IO)
-//                }
+    val socket = aSocket(selectorManager).tcp().connect(address.host, address.port)
-//            }
+    return inetTransportDevice(socket)
-//        }
+}
-//    }
+
-//}
+fun String.toNetworkAddress(): NetworkAddress {
+    val (host, port) = this.split(":").map { it.trim() }
+    return NetworkAddress(host, port.toInt())
+}
+
+private fun inetTransportDevice(
+    sock: Socket,
+    suffix: String = "cli",
+): InetTransportDevice {
+    val networkAddress = (sock.remoteAddress as InetSocketAddress).let { NetworkAddress(it.hostname, it.port) }
+    val inputBlocks = Channel<UByteArray>(4096)
+    val outputBlocks = Channel<UByteArray>(4096)
+
+    val log = LogTag("TCPT${logCounter.incrementAndGet()}:$suffix:$networkAddress")
+    val job = AtomicValue<Job?>(null)
+
+    val sockOutput = sock.openWriteChannel()
+    val sockInput = runCatching { sock.openReadChannel() }.getOrElse {
+        log.warning { "failed to open read channel $it" }
+        throw IllegalStateException("failed to open read channel")
+    }
+
+    fun stop() {
+        job.mutate {
+            if ( it != null ) {
+                log.debug { "stopping" }
+                runCatching { inputBlocks.close() }
+                runCatching { outputBlocks.close() }
+                // The problem: on mac platofrms closing the socket does not close its input
+                // and output channels!
+                runCatching { sockInput.cancel() }
+                runCatching { sockOutput.close() }
+                if (!sock.isClosed)
+                    runCatching {
+                        log.debug { "closing socket by stop" }
+                        sock.close()
+                    }
+                else
+                    log.debug { "socket is already closed when stop is called" }
+                it.cancel()
+                log.debug { "implementation job cancel called" }
+            } else
+                log.debug { "already stopped" }
+            null
+        }
+    }
+
+    var lastActiveAt = Clock.System.now()
+    job.value = globalLaunch {
+        launch {
+
+            log.debug { "opening read channel" }
+
+            while (isActive && sock.isActive) {
+                try {
+                    val size = AsyncVarint.decodeUnsigned(sockInput).toInt()
+                    if (size > MAX_TCP_BLOCK_SIZE) // 16M is a max command block
+                        throw ProtocolException("Illegal block size: $size should be < $MAX_TCP_BLOCK_SIZE")
+                    val data = ByteArray(size)
+                    if (size == 0) {
+                        log.debug { "ping received" }
+                        lastActiveAt = Clock.System.now()
+                    } else {
+                        sockInput.readFully(data, 0, size)
+                        inputBlocks.send(data.toUByteArray())
+                    }
+                } catch (e: ClosedReceiveChannelException) {
+                    log.error { "closed receive channel " }
+                    stop()
+                    break
+                } catch (_: CancellationException) {
+                    log.error { "cancellation exception " }
+                    break
+                } catch (e: Exception) {
+                    log.exception { "unexpected exception in TCP socket read" to e }
+                    stop()
+                    break
+                }
+            }
+        }
+
+        launch {
+            val outAccess = Mutex()
+            var lastSentAt = Clock.System.now()
+            launch {
+                while (isActive && sock.isActive) {
+                    delay(500)
+                    val activityTime = if(lastSentAt > lastActiveAt) lastSentAt else lastActiveAt
+                    if (Clock.System.now() - activityTime > PING_INACTIVITY_TIME) {
+                        log.debug { "pinging for inactivity" }
+                        val repeat = outAccess.withLock {
+                            try {
+                                sockOutput.writeByte(0)
+                                sockOutput.flush()
+                                lastSentAt = Clock.System.now()
+                                true
+                            } catch (e: ClosedReceiveChannelException) {
+                                e.printStackTrace()
+                                stop()
+                                false
+                            } catch (_: CancellationException) {
+                                false
+                            } catch (e: Throwable) {
+                                e.printStackTrace()
+                                stop()
+                                false
+                            }
+                        }
+                        if (!repeat) break
+                    }
+                }
+            }
+
+            while (isActive && sock.isActive) {
+                try {
+                    val block = outputBlocks.receive()
+                    outAccess.withLock {
+                        AsyncVarint.encodeUnsigned(block.size.toULong(), sockOutput)
+                        sockOutput.writeFully(block.toByteArray(), 0, block.size)
+                        sockOutput.flush()
+                        lastSentAt = Clock.System.now()
+                    }
+                } catch (_: CancellationException) {
+                    log.debug { "cancellation exception on output" }
+                    stop()
+                    break
+                } catch (_: LocalInterface.BreakConnectionException) {
+                    log.debug { "requested connection break" }
+                    stop()
+                    break
+                } catch (_: ClosedReceiveChannelException) {
+                    log.debug { "receive block channel closed, closing the socket" }
+                    stop()
+                    break
+                } catch (e: Exception) {
+                    log.exception { "unexpected exception. closing." to e }
+                    stop()
+                    break
+                }
+            }
+        }
+    }
+    val device = InetTransportDevice(inputBlocks, outputBlocks, networkAddress, {
+        stop()
+    })
+    log.debug { "Transport ready" }
+    return device
+}

src/ktorSocketTest/kotlin/TcpTest.kt

@@ -0,0 +1,70 @@
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.test.runTest
+import net.sergeych.crypto2.initCrypto
+import net.sergeych.kiloparsec.*
+import net.sergeych.kiloparsec.adapter.acceptTcpDevice
+import net.sergeych.kiloparsec.adapter.connectTcpDevice
+import kotlin.random.Random
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertIs
+
+class TcpTest {
+    class TestException : Exception("test1")
+
+    @Test
+    fun tcpTest() = runTest {
+        initCrypto()
+//        Log.connectConsole(Log.Level.DEBUG)
+        data class Session(
+            var data: String
+        )
+
+        val port = 27170 + Random.nextInt(1, 200)
+
+        val cmdSave by command<String, Unit>()
+        val cmdLoad by command<Unit, String>()
+        val cmdDrop by command<Unit, Unit>()
+        val cmdException by command<Unit, Unit>()
+
+        val cli = KiloInterface<Session>().apply {
+            registerError { TestException() }
+            onConnected { session.data = "start" }
+            on(cmdSave) { session.data = it }
+            on(cmdLoad) {
+                session.data
+            }
+            on(cmdException) {
+                throw TestException()
+            }
+            on(cmdDrop) {
+                throw LocalInterface.BreakConnectionException()
+            }
+        }
+        val server = KiloServer(cli, acceptTcpDevice(port)) {
+            Session("unknown")
+        }
+
+        val client = KiloClient<Unit>() {
+            addErrors(cli)
+            connect { connectTcpDevice("localhost:$port") }
+        }
+        delay(500)
+
+        assertEquals("start", client.call(cmdLoad))
+
+        client.call(cmdSave, "foobar")
+        assertEquals("foobar", client.call(cmdLoad))
+
+        val res = kotlin.runCatching { client.call(cmdException) }
+        assertIs<TestException>(res.exceptionOrNull())
+        assertEquals("foobar", client.call(cmdLoad))
+
+        assertThrows<RemoteInterface.ClosedException> { client.call(cmdDrop) }
+
+        // reconnect?
+        assertEquals("start", client.call(cmdLoad))
+
+        server.close()
+    }
+}

src/nativeMain/kotlin/net/sergeych/kiloparsec/adapter/DatagramProvider.native.kt

@@ -1,2 +0,0 @@
-package net.sergeych.kiloparsec.adapter
-

src/nativeMain/kotlin/net/sergeych/kiloparsec/adapter/NetworkProvider.native.kt

@@ -1,15 +0,0 @@
-package net.sergeych.kiloparsec.adapter
-
-import kotlinx.coroutines.flow.Flow
-
-actual fun NetworkAddress(host: String, port: Int): NetworkAddress {
-    TODO("Not yet implemented")
-}
-
-actual fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
-    TODO("Not yet implemented")
-}
-
-actual suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
-    TODO("Not yet implemented")
-}