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

0.2.5-snapshot native support for tcp client/server

Browse Source
This commit is contained in:
sergeych committed 2024-08-03 15:36:04 +02:00
1 parent 26d1f3522f
commit ffcdcf7350
16 files changed
+524 -580
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/commonMain/kotlin/net/sergeych/kiloparsec/KiloServer.kt
+2
View File
@@ -43,6 +43,8 @@ class KiloServer<S>(
}
fun close() {
println("PRREEEC")
job.cancel()
println("POOOSTC")
}
}
src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/NetworkProvider.kt
+49 -49
View File
@@ -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.
* Notice to implementors. It must provide correct and effective [equals] and [hashCode].
* Multiplatform internet address.
*/
interface NetworkAddress {
val host: String
data class NetworkAddress(
val host: String,
val port: Int
}
/**
* 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 {
val incoming: ReceiveChannel<Datagram>
suspend fun send(message: UByteArray, networkAddress: NetworkAddress)
@Suppress("unused")
suspend fun send(message: UByteArray, datagramAddress: String) {
send(message, datagramAddress.toNetworkAddress())
) {
override fun toString(): String {
return "$host:$port"
}
suspend fun send(message: UByteArray,host: String,port: Int) =
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())
//
///**
// * 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 {
//
// val incoming: ReceiveChannel<Datagram>
// suspend fun send(message: UByteArray, networkAddress: NetworkAddress)
// @Suppress("unused")
// suspend fun send(message: UByteArray, datagramAddress: String) {
// send(message, datagramAddress.toNetworkAddress())
// }
//
// suspend fun send(message: UByteArray,host: String,port: Int) =
// 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
-2
View File
@@ -1,2 +0,0 @@
package net.sergeych.kiloparsec.adapter
src/jsMain/kotlin/net/sergeych/kiloparsec/adapter/NetworkProvider.js.kt
-15
View File
@@ -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/UdpServer.kt
+118 -118
View File
@@ -1,118 +1,118 @@
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)
}
}
}
//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
+155 -155
View File
@@ -1,155 +1,155 @@
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()
}
//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
+4 -59
View File
@@ -3,22 +3,20 @@ package net.sergeych.kiloparsec
import assertThrows
import io.ktor.server.engine.*
import io.ktor.server.netty.*
import kotlinx.coroutines.delay
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())
@@ -26,60 +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(27101)) {
Session("unknown")
}
val client = KiloClient<Unit>() {
addErrors(cli)
connect { connectTcpDevice("localhost:27101") }
}
delay(500)
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
-100
View File
@@ -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
+40 -34
View File
@@ -2,27 +2,16 @@ package net.sergeych.kiloparsec.adapter
import io.ktor.network.selector.*
import io.ktor.network.sockets.*
import io.ktor.util.network.*
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ClosedReceiveChannelException
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import net.sergeych.kiloparsec.AsyncVarint
import net.sergeych.kiloparsec.LocalInterface
import net.sergeych.mp_logger.*
import net.sergeych.tools.AtomicCounter
class SocketNetworkAddress(override val host: String, override val port: Int) : NetworkAddress {
override fun toString(): String {
return "$host:$port"
}
}
actual fun NetworkAddress(host: String, port: Int): NetworkAddress = SocketNetworkAddress(host, port)
import net.sergeych.tools.AtomicValue
private val logCounter = AtomicCounter(0)
@@ -30,51 +19,66 @@ class ProtocolException(text: String, cause: Throwable? = null) : RuntimeExcepti
const val MAX_TCP_BLOCK_SIZE = 16776216
actual fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
val selectorManager = SelectorManager(Dispatchers.IO)
val serverSocket = aSocket(selectorManager).tcp().bind("127.0.0.1", port)
val log = LogTag("TCPS${logCounter.incrementAndGet()}")
return flow {
while(true) {
log.info { "Accepting incoming connections on $port" }
while (true) {
serverSocket.accept().let { sock ->
log.info { "Emitting transport device" }
emit(inetTransportDevice(sock, "srv"))
}
}
}
}
actual suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
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)
println("Connected to ${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.toJavaAddress().let { NetworkAddress(it.hostname, it.port) }
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 stopCalled = AtomicValue(false)
fun stop() {
log.info { "stopping" }
runCatching { inputBlocks.close()}
runCatching { outputBlocks.close()}
if( !sock.isClosed ) runCatching { sock.close()}
stopCalled.mutate {
if (!it) {
log.debug { "stopping" }
runCatching { inputBlocks.close() }
runCatching { outputBlocks.close() }
if (!sock.isClosed)
runCatching {
log.debug { "closing socket by stop" }
sock.close()
}
else
log.debug { "socket is already closed when stop is called" }
} else
log.debug { "already stopped" }
true
}
}
sock.launch {
log.debug { "opening read channel" }
val sockInput = runCatching { sock.openReadChannel() }.getOrElse {
log.warning { "failed to open read channel $it" }
sock.close()
stop()
throw IllegalStateException("failed to open read channel")
}
while (isActive && sock.isActive) {
@@ -82,15 +86,16 @@ private fun inetTransportDevice(
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")
log.info { "read size: $size" }
val data = ByteArray(size)
log.info { "data ready" }
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()
@@ -105,16 +110,17 @@ private fun inetTransportDevice(
val block = outputBlocks.receive()
AsyncVarint.encodeUnsigned(block.size.toULong(), sockOutput)
sockOutput.writeFully(block.toByteArray(), 0, block.size)
log.info { "Client sock output: ${block.size}" }
sockOutput.flush()
} catch (_: CancellationException) {
log.info { "Caught cancellation, closing transport" }
log.debug { "cancellation exception on output" }
stop()
break
} catch (_: LocalInterface.BreakConnectionException) {
log.info { "requested connection break" }
log.debug { "requested connection break" }
stop()
break
} catch (_: ClosedReceiveChannelException) {
log.info { "receive block channel closed, closing the socket" }
log.debug { "receive block channel closed, closing the socket" }
stop()
break
} catch (e: Exception) {
@@ -124,10 +130,10 @@ private fun inetTransportDevice(
}
}
}
val device = InetTransportDevice(inputBlocks, outputBlocks, networkAddress, {
log.info { "Close has been called" }
stop()
})
log.info { "Transport ready" }
log.debug { "Transport ready" }
return device
}
src/ktorSocketTest/kotlin/TcpTest.kt
+77
View File
@@ -0,0 +1,77 @@
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 net.sergeych.mp_logger.Log
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) }
println(res.exceptionOrNull())
assertIs<TestException>(res.exceptionOrNull())
assertEquals("foobar", client.call(cmdLoad))
println("----------------------------------- pre drops")
assertThrows<RemoteInterface.ClosedException> { client.call(cmdDrop) }
println("----------------------------------- DROPPED")
// reconnect?
assertEquals("start", client.call(cmdLoad))
println("------------------------------=---- RECONNECTED")
server.close()
println("****************************************************************")
}
}
src/nativeMain/kotlin/net/sergeych/kiloparsec/adapter/DatagramProvider.native.kt
-2
View File
@@ -1,2 +0,0 @@
package net.sergeych.kiloparsec.adapter
src/nativeMain/kotlin/net/sergeych/kiloparsec/adapter/NetworkProvider.native.kt
-15
View File
@@ -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")
}