release 0.2.4

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

@@ -0,0 +1,37 @@
+package net.sergeych.kiloparsec
+
+import io.ktor.utils.io.*
+
+object AsyncVarint {
+    suspend fun encodeUnsigned(value: ULong, output: ByteWriteChannel) {
+        var rest = value
+        do {
+            val x = (rest and 127u).toInt()
+            rest = rest shr 7
+            if (rest > 0u)
+                output.writeByte(x or 0x80)
+            else
+                output.writeByte(x)
+
+        } while (rest > 0u)
+    }
+
+    suspend fun decodeUnsigned(source: ByteReadChannel): ULong {
+        var result: ULong = 0u
+        var count = 0
+        while (true) {
+            val x = source.readUByte().toInt()
+            result = result or ((x and 0x7F).toULong() shl count)
+            if ((x and 0x80) == 0)
+                break
+            count += 7
+        }
+        return result
+    }
+}
+
+suspend fun ByteReadChannel.readUByte(): UByte = this.readByte().toUByte()
+
+suspend fun ByteWriteChannel.writeByte(byte: Int) {
+    this.writeByte(byte.toByte())
+}

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

@@ -1,7 +1,7 @@
 package net.sergeych.kiloparsec
 
 /**
- * The local interface to provice functions, register errors for Kiloparsec users. Use it
+ * The local interface to provide functions, register errors for Kiloparsec users. Use it
  * with [KiloClient], [KiloClientConnection], [KiloServerConnection], etc.
  *
  * BAse implementation registers relevant exceptions.

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

@@ -22,12 +22,14 @@ actual fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
             }
         }
         while (true) {
+            println("0 --- $port")
             val connectedSocket = suspendCancellableCoroutine { continuation ->
                 continuation.invokeOnCancellation {
                     socket.close()
                 }
                 socket.accept(continuation, ContinuationHandler())
             }
+            println("1 ---")
             emit(asyncSocketToDevice(connectedSocket))
         }
     }

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

@@ -100,16 +100,9 @@ class ClientTest {
             on(cmdCheckConnected) { connectedCalled }
             on(cmdClose) {
                 throw LocalInterface.BreakConnectionException()
-//                if( closeCounter < 2 ) {
-//                    println("-------------------------- call close!")
-//                    throw RemoteInterface.ClosedException()
-//                }
-//                else {
-//                    println("close counter $closeCounter, ignoring")
-//                }
             }
         }
-//       val server = setupWebsoketServer()
+
         val ns: NettyApplicationEngine = embeddedServer(Netty, port = 8080, host = "0.0.0.0", module = {
             setupWebsocketServer(serverInterface) { Session() }
         }).start(wait = false)

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

@@ -10,6 +10,7 @@ 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
@@ -43,8 +44,11 @@ class NetworkTest {
             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) {
@@ -63,13 +67,22 @@ class NetworkTest {
             }
             yield()
             run {
-                val s = connectTcpDevice("127.0.1.1:17171".toNetworkAddress())
-                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()
+                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())

src/ktorSocketMain/kotlin/adapter/socketClient.kt

@@ -0,0 +1,43 @@
+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.channels.Channel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
+import net.sergeych.kiloparsec.AsyncVarint
+import net.sergeych.mp_logger.LogTag
+import kotlin.io.use
+
+//class SocketNetworkAddress(override val host: String, override val port: Int) : NetworkAddress
+//
+//actual fun NetworkAddress(host: String, port: Int): NetworkAddress = SocketNetworkAddress(host, port)
+//
+//fun acceptTcpSocketDevice(port: Int): Flow<InetTransportDevice> {
+//    val selectorManager = SelectorManager(Dispatchers.IO)
+//    val serverSocket = aSocket(selectorManager).tcp().bind("127.0.0.1", port)
+//    val log = LogTag("TCPS${}")
+//    return flow {
+//        serverSocket.accept().use { sock ->
+//            val closed = CompletableDeferred<Boolean>()
+//            val scope = coroutineScope {
+//                val networkAddress = sock.remoteAddress.toJavaAddress().let { NetworkAddress(it.hostname, it.port) }
+//                val inputBlocks = Channel<UByteArray>(4096)
+//                sock.launch {
+//                    val sockInput = sock.openReadChannel()
+//                    while (isActive && sock.isActive) {
+//                        try {
+//                            val size = AsyncVarint.decodeUnsigned(sockInput)
+//                        } catch (e: Exception) {
+//
+//                        }
+//                    }
+//                }
+//            }
+//        }
+//    }
+//}