GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Kotlin RUDP ergonomic helpers

Browse source
This commit is contained in:
Meta Vens 2026-06-15 21:23:34 +01:00
commit 0d2bbe7029
2 changed files with 181 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

34
packages/cultmesh-kotlin/README.md
View file

@ -6,6 +6,40 @@ It provides typed MessagePack document codecs, a tiny WebSocket CultNet lane,
a single-peer CultNet RUDP socket transport, an in-memory CultCache, and the
first Eve dashboard/sensor document contracts.
## RUDP Happy Path
Kotlin exposes factory helpers around the shared RUDP transport so callers can
write client/server code without hand-binding sockets or repeating channel
strings:
```kotlin
cultNetRudpServer(
runtimeId = "kotlin-server",
connectionId = 0x10203040,
tuning = CultNetRudpSocketTuning(maxFragmentBytes = 1024),
).use { server ->
cultNetRudpClient(
runtimeId = "kotlin-client",
connectionId = 0x10203040,
remoteHost = "127.0.0.1",
remotePort = server.localPort,
).use { client ->
client.connect("join")
check(pumpRudpPairUntilConnected(client, server))
client.sendSchema("client-state")
val payload = server.receiveSchema(timeoutMs = 1_000)
}
}
```
The sugar delegates to the same `CultNetRudpSocketTransportConnection` and
`cultnet.transport.rudp.v0` packet codec used by the cross-runtime interop
harness. `sendSchema`, `sendLatest`, and `sendRealtime` select the shared
channel semantics; they do not create a Kotlin-only dialect. For a remote peer
that already has its own receive loop, use `connectAndWait(...)`; for two
same-process transports, use `pumpRudpPairUntilConnected(...)` to drive both
sides through the handshake.
`EveDashboardStateDocument` mirrors the CultUI-shaped dashboard surface contract:
`surface.root` is the retained UI tree, `surface.assets` carries cacheable media
references, and the flat `nodes` projection remains a compatibility and binding

147
packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt
View file

@ -459,6 +459,74 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) {
enum class CultNetRudpSocketMode { Client, Server }
data class CultNetRudpSocketTuning(
val initialSequence: Long = 1,
val resendDelayMs: Long = 250,
val maxFragmentBytes: Int? = null,
val maxPendingReliablePackets: Int? = null,
)
fun cultNetRudpServer(
runtimeId: String,
connectionId: Long,
bindHost: String = "127.0.0.1",
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(initialSequence = 100),
): CultNetRudpSocketTransportConnection {
val socket = DatagramSocket(bindPort, InetAddress.getByName(bindHost)).also { it.soTimeout = 20 }
return CultNetRudpSocketTransportConnection(
socket = socket,
mode = CultNetRudpSocketMode.Server,
runtimeId = runtimeId,
connectionId = connectionId,
initialSequence = tuning.initialSequence,
resendDelayMs = tuning.resendDelayMs,
maxFragmentBytes = tuning.maxFragmentBytes,
maxPendingReliablePackets = tuning.maxPendingReliablePackets,
)
}
fun cultNetRudpClient(
runtimeId: String,
connectionId: Long,
remoteHost: String,
remotePort: Int,
bindHost: String = "127.0.0.1",
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
): CultNetRudpSocketTransportConnection {
val socket = DatagramSocket(bindPort, InetAddress.getByName(bindHost)).also { it.soTimeout = 20 }
return CultNetRudpSocketTransportConnection(
socket = socket,
mode = CultNetRudpSocketMode.Client,
runtimeId = runtimeId,
connectionId = connectionId,
remoteAddress = InetSocketAddress(InetAddress.getByName(remoteHost), remotePort),
initialSequence = tuning.initialSequence,
resendDelayMs = tuning.resendDelayMs,
maxFragmentBytes = tuning.maxFragmentBytes,
maxPendingReliablePackets = tuning.maxPendingReliablePackets,
)
}
fun pumpRudpPairUntilConnected(
first: CultNetRudpSocketTransportConnection,
second: CultNetRudpSocketTransportConnection,
timeoutMs: Long = 1_000,
pollIntervalMs: Long = 5,
): Boolean {
val deadline = System.nanoTime() + timeoutMs * 1_000_000L
while (System.nanoTime() < deadline) {
first.receiveOnce()
second.receiveOnce()
first.pollResends()
second.pollResends()
if (first.connected && second.connected) return true
Thread.sleep(pollIntervalMs)
}
return first.connected && second.connected
}
class CultNetRudpSocketTransportConnection(
private val socket: DatagramSocket,
private val mode: CultNetRudpSocketMode,
@ -489,6 +557,7 @@ class CultNetRudpSocketTransportConnection(
maxPendingReliablePackets = maxPendingReliablePackets,
)
val connected: Boolean get() = session.connected
val localPort: Int get() = socket.localPort
val stats: CultNetTransportStats get() = CultNetTransportStats(bytesReceived, bytesSent, framesReceived, framesSent)
fun connect(payload: ByteArray = ByteArray(0)) {
@ -496,23 +565,76 @@ class CultNetRudpSocketTransportConnection(
sendPacket(session.createConnect(nowMs(), payload))
}
fun connect(payload: String) = connect(payload.toByteArray(StandardCharsets.UTF_8))
fun connectAndWait(payload: ByteArray = ByteArray(0), timeoutMs: Long = 1_000, pollIntervalMs: Long = 5): Boolean {
connect(payload)
return awaitConnected(timeoutMs, pollIntervalMs)
}
fun connectAndWait(payload: String, timeoutMs: Long = 1_000, pollIntervalMs: Long = 5): Boolean =
connectAndWait(payload.toByteArray(StandardCharsets.UTF_8), timeoutMs, pollIntervalMs)
fun awaitConnected(timeoutMs: Long = 1_000, pollIntervalMs: Long = 5): Boolean {
val deadline = System.nanoTime() + timeoutMs * 1_000_000L
while (System.nanoTime() < deadline) {
receiveOnce()
pollResends()
if (connected) return true
Thread.sleep(pollIntervalMs)
}
return connected
}
fun send(channelId: String, payload: ByteArray) {
session.sendMany(channelId, payload, channelSendOptions(channelId, nowMs()), maxFragmentBytes).forEach { sendPacket(it) }
framesSent += 1
}
fun send(channelId: String, payload: String) = send(channelId, payload.toByteArray(StandardCharsets.UTF_8))
fun sendSchema(payload: ByteArray) = send("schema", payload)
fun sendSchema(payload: String) = sendSchema(payload.toByteArray(StandardCharsets.UTF_8))
fun sendLatest(payload: ByteArray) = send("latest", payload)
fun sendLatest(payload: String) = sendLatest(payload.toByteArray(StandardCharsets.UTF_8))
fun sendRealtime(payload: ByteArray) = send("realtime", payload)
fun sendRealtime(payload: String) = sendRealtime(payload.toByteArray(StandardCharsets.UTF_8))
fun disconnect(reason: ByteArray = ByteArray(0)) {
sendPacket(session.createDisconnect(reason))
}
fun disconnect(reason: String) = disconnect(reason.toByteArray(StandardCharsets.UTF_8))
fun ping(payload: ByteArray = ByteArray(0)) {
sendPacket(session.createPing(payload))
}
fun ping(payload: String) = ping(payload.toByteArray(StandardCharsets.UTF_8))
fun pollPongPayload(): ByteArray? = if (pongPayloads.isEmpty()) null else pongPayloads.removeFirst()
fun checkTimeout(timeoutMs: Long): Boolean = session.checkTimeout(nowMs(), timeoutMs)
fun receiveUntil(timeoutMs: Long, pollIntervalMs: Long = 5, predicate: (CultNetTransportFrame) -> Boolean = { true }): CultNetTransportFrame? {
val deadline = System.nanoTime() + timeoutMs * 1_000_000L
while (System.nanoTime() < deadline) {
val frame = receiveOnce()
if (frame != null && predicate(frame)) return frame
pollResends()
Thread.sleep(pollIntervalMs)
}
return null
}
fun receiveSchema(timeoutMs: Long, pollIntervalMs: Long = 5): ByteArray? =
receiveUntil(timeoutMs, pollIntervalMs) { it.channelId == "schema" }?.payload
fun receiveOnce(): CultNetTransportFrame? {
if (!delivered.isEmpty()) return delivered.removeFirst()
val buffer = ByteArray(65535)
@ -658,6 +780,7 @@ fun main(args: Array<String>) {
rudpSessionPingsAndDetectsReceiveTimeout()
rudpSessionBoundsPendingReliablePacketsBeforeEnqueue()
rudpSessionFragmentsAndReassemblesReliableOrderedPayloads()
rudpSocketTransportErgonomicFactoriesCarrySchemaFrames()
rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames()
rudpSocketTransportCarriesFragmentedReliableOrderedSchemaFrames()
return
@ -890,6 +1013,30 @@ private fun rudpSessionFragmentsAndReassemblesReliableOrderedPayloads() {
check(delivered.first().sequence == packets.first().sequence)
}
private fun rudpSocketTransportErgonomicFactoriesCarrySchemaFrames() {
val connectionId = 0x10203042L
cultNetRudpServer(
runtimeId = "kotlin-rudp-sugar-server",
connectionId = connectionId,
tuning = CultNetRudpSocketTuning(resendDelayMs = 25, maxFragmentBytes = 8, maxPendingReliablePackets = 16),
).use { server ->
cultNetRudpClient(
runtimeId = "kotlin-rudp-sugar-client",
connectionId = connectionId,
remoteHost = "127.0.0.1",
remotePort = server.localPort,
tuning = CultNetRudpSocketTuning(resendDelayMs = 25, maxFragmentBytes = 8, maxPendingReliablePackets = 16),
).use { client ->
client.connect("join")
check(pumpRudpPairUntilConnected(client, server))
client.sendSchema("client-state")
check(String(server.receiveSchema(1_000) ?: error("Server did not receive schema frame"), StandardCharsets.UTF_8) == "client-state")
server.sendSchema("server-state")
check(String(client.receiveSchema(1_000) ?: error("Client did not receive schema frame"), StandardCharsets.UTF_8) == "server-state")
}
}
}
private fun rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames() {
val loopback = InetAddress.getByName("127.0.0.1")
val serverSocket = DatagramSocket(0, loopback).also { it.soTimeout = 20 }