started more or less MP internet protocols

This commit is contained in:
sergeych committed 2023-11-12 04:26:17 +03:00
1 parent 770555024f
commit 6c10c2e578
12 files changed
+358 -2

No files matched your search

@@ -0,0 +1,8 @@
package net.sergeych.kiloparsec.adapter
import java.net.InetAddress
actual fun networkAddressOf(address: String): NetworkAddress {
val (host,port) = address.split(":")
return JvmNetworkAddress(InetAddress.getByName(host), port.toInt())
}
@@ -0,0 +1,6 @@
package net.sergeych.kiloparsec.adapter
import java.net.InetAddress
actual fun NetworkAddress(host: String, port: Int): NetworkAddress =
JvmNetworkAddress(InetAddress.getByName(host), port)
@@ -0,0 +1,135 @@
package net.sergeych.kiloparsec.adapter
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
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.DatagramPacket
import java.net.DatagramSocket
import java.net.InetAddress
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.hostName }
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
}
override fun hashCode(): Int {
var result = inetAddress.hashCode()
result = 31 * result + port
return result
}
}
class UdpDatagram(override val message: UByteArray, val inetAddress: InetAddress, val port: Int) : Datagram {
override val address: NetworkAddress by lazy {
JvmNetworkAddress(inetAddress, port)
}
private val access = Mutex()
private var socket: DatagramSocket? = null
override suspend fun respondWith(message: UByteArray) {
withContext(Dispatchers.IO) {
access.withLock {
if (socket == null) socket = DatagramSocket()
val packet = DatagramPacket(
message.toByteArray(),
message.size,
inetAddress,
port
)
socket!!.send(packet)
}
}
}
}
@OptIn(DelicateCoroutinesApi::class)
class UdpServer(val port: Int) :
DatagramReceiver, 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)
}
}
}