lazzexus/Kafkatorio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

finish migrating to ktor!

Browse source 
- remove http4k
- renamed 'schema' subproject to 'library', because it contains more than just data models
- tiles are now hosted per factorio-server
- changed api base to `kafkatorio/data`
- tidied up some Gradle conf
- remove 'events-processor' subproject - I don't want to set up loads of microservices right now
This commit is contained in:
Adam 2022-07-02 08:45:21 +02:00
parent 8a90c4fb5e
commit c0c82e7c9c
80 changed files with 1878 additions and 1280 deletions
Download patch file Download diff file Expand all files Collapse all files

2
modules/events-server/.gitignore vendored
View file

@ -1,2 +1,2 @@
kafka-state/
kafkatorio-tiles/
server-data/

23
modules/events-server/build.gradle.kts
View file

@ -1,5 +1,6 @@
import dev.adamko.kafkatorio.task.KafkaConsumerGroupsTask
import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
import org.jetbrains.kotlin.utils.addToStdlib.applyIf
plugins {
@ -22,9 +23,7 @@ dependencies {
implementation(libs.kafka.streams)
testImplementation(libs.kafka.streamsTestUtils)
implementation(projects.modules.eventsSchema)
// implementation("com.ionspin.kotlin:bignum:0.3.3")
implementation(projects.modules.eventsLibrary)
implementation(libs.kotlinx.serialization.core)
implementation(libs.kotlinx.serialization.json)
@ -32,17 +31,12 @@ dependencies {
implementation(libs.kotlinx.coroutines.core)
implementation(libs.bundles.http4k)
implementation(libs.http4k.format.kotlinx)
implementation(libs.bundles.logging)
implementation(libs.scrimage.core)
implementation(libs.okio.core)
//<editor-fold desc="Ktor">
implementation(libs.ktorSerialization.kotlinxJson)
@ -81,15 +75,7 @@ dependencies {
application {
// mainClass.set("dev.adamko.kafkatorio.processor.EventProcessorKt")
mainClass.set("dev.adamko.kafkatorio.server.KafkatorioEventsServerKt")
// applicationDefaultJvmArgs += listOf(
// "-Dcom.sun.management.jmxremote.port=9186",
// "-Dcom.sun.management.jmxremote.rmi.port=9186",
// "-Dcom.sun.management.jmxremote.ssl=false",
// "-Dcom.sun.management.jmxremote.local.only=false",
// "-Dcom.sun.management.jmxremote.authenticate=false",
// )
mainClass.set("dev.adamko.kafkatorio.server.EventsServerKt")
}
tasks.withType<KotlinCompile>().configureEach {
@ -111,9 +97,6 @@ val kafkaConsumersDelete by tasks.registering(KafkaConsumerGroupsTask.DeleteAll:
group = project.name
}
//tasks.run.configure {
// dependsOn(kafkatorioEventsServerKafkaForceReset)
//}
idea {
module {

58
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt
View file

@ -1,58 +0,0 @@
package dev.adamko.kafkatorio.processor
import dev.adamko.kafkatorio.processor.admin.KafkatorioKafkaAdmin
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import dev.adamko.kafkatorio.processor.tileserver.SyslogSocketServer
import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer
import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer
import dev.adamko.kafkatorio.processor.tileserver.webServer
import kotlinx.coroutines.awaitCancellation
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.job
import kotlinx.coroutines.launch
import org.http4k.server.Http4kServer
suspend fun main() {
val appProps = ApplicationProperties.load()
val admin = KafkatorioKafkaAdmin(appProps)
admin.createKafkatorioTopics()
val syslogServer = SyslogSocketServer(appProps)
val wsServer = WebsocketServer()
val tileServer = WebMapTileServer(appProps)
val topology = KafkatorioTopology(appProps, wsServer, syslogServer)
coroutineScope {
launch {
syslogServer.start()
}
launch {
topology.start()
println("launched KafkatorioTopology")
awaitCancellation()
}
launch {
val webServer: Http4kServer = webServer(tileServer, wsServer, appProps)
currentCoroutineContext().job.invokeOnCompletion {
println("webserver stopping. Cause: $it")
webServer.stop()
}
webServer.start()
println("launched webserver port:${webServer.port()}")
awaitCancellation()
}
}
}

193
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/FileEventsFlow.kt
View file

@ -1,193 +0,0 @@
package dev.adamko.kafkatorio.processor.misc
import java.io.File
import java.nio.file.FileSystems
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.StandardWatchEventKinds.ENTRY_CREATE
import java.nio.file.StandardWatchEventKinds.ENTRY_DELETE
import java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY
import java.nio.file.WatchEvent
import java.nio.file.WatchKey
import java.nio.file.WatchService
import kotlin.coroutines.CoroutineContext
import kotlin.io.path.isReadable
import kotlin.time.Duration.Companion.seconds
import kotlinx.coroutines.CoroutineExceptionHandler
import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.cancellable
import kotlinx.coroutines.flow.debounce
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.flatMapConcat
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.mapNotNull
import kotlinx.coroutines.flow.runningFold
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
import kotlinx.coroutines.yield
fun File.fileEventsFlow() = FileEventsFlow(
root = this.toPath(),
)
fun Path.fileEventsFlow() = FileEventsFlow(
root = this,
)
/** [SharedFlow] wrapper for Java's [WatchService] */
class FileEventsFlow private constructor(
private val root: Path,
private val fileEvents: MutableSharedFlow<KWatchEvent>,
) : SharedFlow<KWatchEvent> by fileEvents, CoroutineScope {
constructor(root: Path) : this(root, MutableSharedFlow(0, Int.MAX_VALUE, BufferOverflow.SUSPEND))
override val coroutineContext: CoroutineContext =
CoroutineName("FileEventsFlow") +
Dispatchers.Unconfined +
CoroutineExceptionHandler { _, e ->
println("handling exception in FileEventsFlow")
e.printStackTrace()
watchService.close()
watchKeys.update {
it.cancelAll()
emptyList()
}
}
private val watchService: WatchService = FileSystems.getDefault().newWatchService()
private val watchKeys: MutableStateFlow<List<WatchKey>> = MutableStateFlow(emptyList())
init {
println("launching FileEventsFlow... $root")
registerPaths()
pollFileEvents()
launch {
fileEvents.emit(KWatchEvent.Initialized)
}
println("launched FileEventsFlow $root")
root.toFile().walk()
// .filter { it.isFile }
.onEach {
launch {
fileEvents.emit(KWatchEvent.Created(it.toPath()))
}
}
.count()
.let {
println("initializing KWatchEvent.Created $it")
}
}
/** Watch for any file tree changes and re-register the [watchKeys] */
private fun registerPaths() = launch {
println("launching registerPaths")
fileEvents
// .onEach { println("file event! $it") }
.filter { event ->
when (event) {
is KWatchEvent.Initialized -> true
is KWatchEvent.Created -> event.path.isDirectory
is KWatchEvent.Deleted -> event.path.isDirectory
is KWatchEvent.Modified -> false
}
}.debounce(1.seconds)
.runningFold(emptyList<WatchKey>()) { previousKeys, directoryTreeAlteredEvent ->
println("re-registering paths $directoryTreeAlteredEvent")
previousKeys.cancelAll()
root.toFile().walk()
.filter { it.isDirectory }
.map {
// println("re-registering watch on $it")
it.toPath().register(watchService, ENTRY_CREATE, ENTRY_MODIFY, ENTRY_DELETE)
}
.toList()
.also {
println("${it.count()} registered watches")
}
}
.cancellable()
.collect(watchKeys)
}
/** Poll [watchService] for updates while [coroutineContext] is active. */
private fun pollFileEvents() = launch {
flow<WatchKey> {
while (coroutineContext.isActive) {
emit(
withContext(Dispatchers.IO) { watchService.take() }
)
println("emitted watchService.take()")
yield()
}
}.mapNotNull { monitorKey ->
when (val keyPath = monitorKey.watchable()) {
is Path -> monitorKey to keyPath
else -> null
}
}.filter { it.second.isReadable() }
.flatMapConcat { (monitorKey: WatchKey, keyPath: Path) ->
val events = monitorKey.pollEvents()
monitorKey.reset()
events
.onEach { println("monitorKey event $it") }
.filter { it.context() as? Path != null }
.mapNotNull { event ->
val eventPath = keyPath.resolve(event.context() as Path)
when (event.kind()) {
ENTRY_CREATE -> KWatchEvent.Created(eventPath)
ENTRY_DELETE -> KWatchEvent.Deleted(eventPath)
ENTRY_MODIFY -> KWatchEvent.Modified(eventPath)
else -> null
}
}.asFlow()
}.cancellable()
.collect(fileEvents)
}
companion object {
private val Path.isDirectory: Boolean get() = Files.isDirectory(this)
private fun List<WatchKey>.cancelAll() = forEach { it.cancel(); it.pollEvents() }
}
}
/** Wrapper around [WatchEvent] that comes with properly resolved absolute path */
sealed interface KWatchEvent {
object Initialized : KWatchEvent
sealed interface PathEvent : KWatchEvent {
/** path of modified file or directory */
val path: Path
}
data class Created(override val path: Path) : PathEvent
data class Modified(override val path: Path) : PathEvent
data class Deleted(override val path: Path) : PathEvent
}

160
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt
View file

@ -1,160 +0,0 @@
package dev.adamko.kafkatorio.processor.serdes
import kotlinx.serialization.BinaryFormat
import kotlinx.serialization.DeserializationStrategy
import kotlinx.serialization.SerializationStrategy
import kotlinx.serialization.builtins.ByteArraySerializer
import kotlinx.serialization.descriptors.SerialDescriptor
import kotlinx.serialization.encoding.AbstractDecoder
import kotlinx.serialization.encoding.AbstractEncoder
import kotlinx.serialization.encoding.CompositeDecoder
import kotlinx.serialization.encoding.CompositeEncoder
import kotlinx.serialization.modules.EmptySerializersModule
import kotlinx.serialization.modules.SerializersModule
import okio.Buffer
import okio.BufferedSink
import okio.BufferedSource
val kxsBinary: KxsBinary = KxsBinary()
/** [`https://github.com/Kotlin/kotlinx.serialization/blob/master/docs/formats.md#efficient-binary-format`](https://github.com/Kotlin/kotlinx.serialization/blob/master/docs/formats.md#efficient-binary-format) */
class KxsBinary(
override val serializersModule: SerializersModule = EmptySerializersModule
) : BinaryFormat {
override fun <T> encodeToByteArray(serializer: SerializationStrategy<T>, value: T): ByteArray {
val output = Buffer()
val encoder = KxsDataOutputEncoder(output, serializersModule)
encoder.encodeSerializableValue(serializer, value)
return output.readByteArray()
}
override fun <T> decodeFromByteArray(
deserializer: DeserializationStrategy<T>,
bytes: ByteArray
): T {
val input = Buffer().write(bytes)
val decoder = KxsDataInputDecoder(input, serializersModule)
return decoder.decodeSerializableValue(deserializer)
}
}
private val byteArraySerializer = ByteArraySerializer()
class KxsDataOutputEncoder(
private val output: BufferedSink,
override val serializersModule: SerializersModule = EmptySerializersModule,
) : AbstractEncoder() {
private operator fun BufferedSink.invoke(action: BufferedSink.() -> Unit): Unit = action()
override fun encodeBoolean(value: Boolean) = output { writeByte(if (value) 1 else 0) }
override fun encodeByte(value: Byte) = output { writeByte(value.toInt()) }
override fun encodeChar(value: Char) = output { writeUtf8CodePoint(value.code) }
override fun encodeDouble(value: Double) = output { writeLong(value.toRawBits()) }
override fun encodeEnum(enumDescriptor: SerialDescriptor, index: Int) = output { writeInt(index) }
override fun encodeFloat(value: Float) = output { writeInt(value.toRawBits()) }
override fun encodeInt(value: Int) = output { writeInt(value) }
override fun encodeLong(value: Long) = output { writeLong(value) }
override fun encodeShort(value: Short) = output { writeShort(value.toInt()) }
override fun encodeString(value: String) = output {
encodeCompactSize(value.length)
writeUtf8(value)
}
override fun beginCollection(
descriptor: SerialDescriptor,
collectionSize: Int
): CompositeEncoder {
encodeCompactSize(collectionSize)
return this
}
override fun encodeNull() = encodeBoolean(false)
override fun encodeNotNullMark() = encodeBoolean(true)
override fun <T> encodeSerializableValue(serializer: SerializationStrategy<T>, value: T) {
when (serializer.descriptor) {
byteArraySerializer.descriptor -> encodeByteArray(value as ByteArray)
else -> super.encodeSerializableValue(serializer, value)
}
}
private fun encodeByteArray(bytes: ByteArray) {
encodeCompactSize(bytes.size)
output.write(bytes)
}
private fun encodeCompactSize(value: Int) {
if (value < 0xff) {
output.writeByte(value)
} else {
output.writeByte(0xff)
output.writeInt(value)
}
}
}
class KxsDataInputDecoder(
private val input: BufferedSource,
override val serializersModule: SerializersModule = EmptySerializersModule,
private var elementsCount: Int = 0,
) : AbstractDecoder() {
private var elementIndex = 0
override fun decodeBoolean(): Boolean = input.readByte().toInt() != 0
override fun decodeByte(): Byte = input.readByte()
override fun decodeChar(): Char = input.readUtf8CodePoint().toChar()
override fun decodeDouble(): Double = Double.fromBits(input.readLong())
override fun decodeEnum(enumDescriptor: SerialDescriptor): Int = input.readInt()
override fun decodeFloat(): Float = Float.fromBits(input.readInt())
override fun decodeInt(): Int = input.readInt()
override fun decodeLong(): Long = input.readLong()
override fun decodeShort(): Short = input.readShort()
override fun decodeString(): String {
val size = decodeCompactSize()
return input.readUtf8(size.toLong())
}
override fun decodeElementIndex(descriptor: SerialDescriptor): Int {
if (elementIndex == elementsCount) return CompositeDecoder.DECODE_DONE
return elementIndex++
}
override fun beginStructure(descriptor: SerialDescriptor): CompositeDecoder =
KxsDataInputDecoder(input, serializersModule, descriptor.elementsCount)
override fun decodeSequentially(): Boolean = true
override fun decodeCollectionSize(descriptor: SerialDescriptor): Int =
decodeCompactSize().also { elementsCount = it }
override fun decodeNotNullMark(): Boolean = decodeBoolean()
override fun <T> decodeSerializableValue(
deserializer: DeserializationStrategy<T>,
previousValue: T?
): T {
@Suppress("UNCHECKED_CAST")
return when (deserializer.descriptor) {
byteArraySerializer.descriptor -> decodeByteArray() as T
else -> super.decodeSerializableValue(deserializer, previousValue)
}
}
private fun decodeByteArray(): ByteArray {
val bytes = ByteArray(decodeCompactSize())
input.readFully(bytes)
return bytes
}
private fun decodeCompactSize(): Int {
val byte = input.readByte().toInt() and 0xff
if (byte < 0xff) return byte
return input.readInt()
}
}

328
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/SyslogSocketServer.kt
View file

@ -1,168 +1,168 @@
package dev.adamko.kafkatorio.processor.tileserver
import com.github.palindromicity.syslog.AllowableDeviations
import com.github.palindromicity.syslog.SyslogParser
import com.github.palindromicity.syslog.SyslogParserBuilder
import com.github.palindromicity.syslog.SyslogSpecification
import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import dev.adamko.kafkatorio.processor.tileserver.SyslogMsg.Companion.toSyslogMsg
import io.ktor.network.selector.SelectorManager
import io.ktor.network.sockets.aSocket
import io.ktor.network.sockets.awaitClosed
import io.ktor.network.sockets.openReadChannel
import io.ktor.utils.io.readUTF8Line
import java.util.EnumSet
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.yield
class SyslogSocketServer(
appProps: ApplicationProperties,
dispatcher: CoroutineDispatcher = Dispatchers.IO,
) {
private val selectorManager: SelectorManager = SelectorManager(dispatcher)
private val serverSocketTcp = aSocket(selectorManager)
.tcp()
.bind(
appProps.socketServerHost,
appProps.socketServerPort,
)
// private val serverSocketUdp: BoundDatagramSocket = aSocket(selectorManager)
// .udp()
// .bind(
// localAddress = InetSocketAddress(
// appProps.socketServerHost,
// appProps.socketServerPortUdp,
// ),
// )
private val syslogParser: SyslogParser = SyslogParserBuilder()
.withDeviations(EnumSet.of(AllowableDeviations.PRIORITY, AllowableDeviations.VERSION))
.forSpecification(SyslogSpecification.RFC_5424)
.build()
private val _messages: MutableSharedFlow<SyslogMsg> = MutableSharedFlow()
val messages: SharedFlow<SyslogMsg>
get() = _messages.asSharedFlow()
suspend fun start() {
// udp()
tcp()
}
// private suspend fun udp() {
// println("[SyslogSocketServer] started listening: ${serverSocketUdp.localAddress}")
//import com.github.palindromicity.syslog.AllowableDeviations
//import com.github.palindromicity.syslog.SyslogParser
//import com.github.palindromicity.syslog.SyslogParserBuilder
//import com.github.palindromicity.syslog.SyslogSpecification
//import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
//import dev.adamko.kafkatorio.server.config.ApplicationProperties
//import dev.adamko.kafkatorio.processor.tileserver.SyslogMsg.Companion.toSyslogMsg
//import io.ktor.network.selector.SelectorManager
//import io.ktor.network.sockets.aSocket
//import io.ktor.network.sockets.awaitClosed
//import io.ktor.network.sockets.openReadChannel
//import io.ktor.utils.io.readUTF8Line
//import java.util.EnumSet
//import kotlinx.coroutines.CoroutineDispatcher
//import kotlinx.coroutines.Dispatchers
//import kotlinx.coroutines.coroutineScope
//import kotlinx.coroutines.flow.MutableSharedFlow
//import kotlinx.coroutines.flow.SharedFlow
//import kotlinx.coroutines.flow.asSharedFlow
//import kotlinx.coroutines.isActive
//import kotlinx.coroutines.launch
//import kotlinx.coroutines.yield
//
// serverSocketUdp.incoming
// .receiveAsFlow()
// .map {udpPacket: Datagram ->
// val msg = udpPacket.packet.readText()
// println("[SyslogSocketServer] received msg: $msg")
// syslogParser.parseLine(msg).toSyslogMsg()
// }.collect(_messages)
//
//class SyslogSocketServer(
// appProps: ApplicationProperties,
// dispatcher: CoroutineDispatcher = Dispatchers.IO,
//) {
//
// private val selectorManager: SelectorManager = SelectorManager(dispatcher)
//
//
// private val serverSocketTcp = aSocket(selectorManager)
// .tcp()
// .bind(
// appProps.socketServerHost,
// appProps.socketServerPort,
// )
//
//
//// private val serverSocketUdp: BoundDatagramSocket = aSocket(selectorManager)
//// .udp()
//// .bind(
//// localAddress = InetSocketAddress(
//// appProps.socketServerHost,
//// appProps.socketServerPortUdp,
//// ),
//// )
//
//
// private val syslogParser: SyslogParser = SyslogParserBuilder()
// .withDeviations(EnumSet.of(AllowableDeviations.PRIORITY, AllowableDeviations.VERSION))
// .forSpecification(SyslogSpecification.RFC_5424)
// .build()
//
//
// private val _messages: MutableSharedFlow<SyslogMsg> = MutableSharedFlow()
// val messages: SharedFlow<SyslogMsg>
// get() = _messages.asSharedFlow()
//
//
// suspend fun start() {
//// udp()
// tcp()
// }
private suspend fun tcp() = coroutineScope {
println("[SyslogSocketServer] started listening: ${serverSocketTcp.localAddress}")
while (isActive) {
val socket = serverSocketTcp.accept()
launch {
val channel = socket.openReadChannel()
try {
while (socket.socketContext.isActive) {
val line = channel.readUTF8Line()
println("[SyslogSocketServer] received: $line")
val syslog = syslogParser.parseLine(line).toSyslogMsg()
_messages.emit(syslog)
yield()
}
} catch (e: Exception) {
socket.awaitClosed()
}
}
}
yield()
}
}
data class SyslogMsg(
val src: Map<String, String?>
) {
val message: String?
get() = src[SyslogFieldKeys.MESSAGE.field]
val headerAppName: String?
get() = src[SyslogFieldKeys.HEADER_APPNAME.field]
val headerHostName: String?
get() = src[SyslogFieldKeys.HEADER_HOSTNAME.field]
val headerPri: String?
get() = src[SyslogFieldKeys.HEADER_PRI.field]
val headerPriSeverity: String?
get() = src[SyslogFieldKeys.HEADER_PRI_SEVERITY.field]
val headerPriFacility: String?
get() = src[SyslogFieldKeys.HEADER_PRI_FACILITY.field]
val headerProcId: String?
get() = src[SyslogFieldKeys.HEADER_PROCID.field]
val headerTimestamp: String?
get() = src[SyslogFieldKeys.HEADER_TIMESTAMP.field]
val headerMsgId: String?
get() = src[SyslogFieldKeys.HEADER_MSGID.field]
val headerVersion: String?
get() = src[SyslogFieldKeys.HEADER_VERSION.field]
val structuredBase: String?
get() = src[SyslogFieldKeys.STRUCTURED_BASE.field]
val structuredElementIdFmt: String?
get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_FMT.field]
val structuredElementIdPnameFmt: String?
get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_FMT.field]
val structuredElementIdPnamePattern: String?
get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_PATTERN.field]
companion object {
fun Map<String, Any?>.toSyslogMsg(): SyslogMsg {
val map = entries.mapNotNull { (k, v) ->
when (v) {
is String -> k to v
else -> null
}
}.toMap()
return SyslogMsg(map)
}
}
}
//
//// private suspend fun udp() {
//// println("[SyslogSocketServer] started listening: ${serverSocketUdp.localAddress}")
////
//// serverSocketUdp.incoming
//// .receiveAsFlow()
//// .map {udpPacket: Datagram ->
//// val msg = udpPacket.packet.readText()
//// println("[SyslogSocketServer] received msg: $msg")
//// syslogParser.parseLine(msg).toSyslogMsg()
//// }.collect(_messages)
//// }
//
// private suspend fun tcp() = coroutineScope {
// println("[SyslogSocketServer] started listening: ${serverSocketTcp.localAddress}")
//
// while (isActive) {
// val socket = serverSocketTcp.accept()
//
// launch {
// val channel = socket.openReadChannel()
// try {
// while (socket.socketContext.isActive) {
// val line = channel.readUTF8Line()
// println("[SyslogSocketServer] received: $line")
//
// val syslog = syslogParser.parseLine(line).toSyslogMsg()
//
// _messages.emit(syslog)
//
// yield()
// }
// } catch (e: Exception) {
// socket.awaitClosed()
// }
// }
// }
//
// yield()
// }
//}
//
//
//data class SyslogMsg(
// val src: Map<String, String?>
//) {
//
// val message: String?
// get() = src[SyslogFieldKeys.MESSAGE.field]
//
// val headerAppName: String?
// get() = src[SyslogFieldKeys.HEADER_APPNAME.field]
//
// val headerHostName: String?
// get() = src[SyslogFieldKeys.HEADER_HOSTNAME.field]
//
// val headerPri: String?
// get() = src[SyslogFieldKeys.HEADER_PRI.field]
//
// val headerPriSeverity: String?
// get() = src[SyslogFieldKeys.HEADER_PRI_SEVERITY.field]
//
// val headerPriFacility: String?
// get() = src[SyslogFieldKeys.HEADER_PRI_FACILITY.field]
//
// val headerProcId: String?
// get() = src[SyslogFieldKeys.HEADER_PROCID.field]
//
// val headerTimestamp: String?
// get() = src[SyslogFieldKeys.HEADER_TIMESTAMP.field]
//
// val headerMsgId: String?
// get() = src[SyslogFieldKeys.HEADER_MSGID.field]
//
// val headerVersion: String?
// get() = src[SyslogFieldKeys.HEADER_VERSION.field]
//
// val structuredBase: String?
// get() = src[SyslogFieldKeys.STRUCTURED_BASE.field]
//
// val structuredElementIdFmt: String?
// get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_FMT.field]
//
// val structuredElementIdPnameFmt: String?
// get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_FMT.field]
//
// val structuredElementIdPnamePattern: String?
// get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_PATTERN.field]
//
// companion object {
// fun Map<String, Any?>.toSyslogMsg(): SyslogMsg {
// val map = entries.mapNotNull { (k, v) ->
// when (v) {
// is String -> k to v
// else -> null
// }
// }.toMap()
// return SyslogMsg(map)
// }
// }
//}

168
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebMapTileServer.kt
View file

@ -1,86 +1,86 @@
package dev.adamko.kafkatorio.processor.tileserver
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import java.security.MessageDigest
import org.http4k.core.Filter
import org.http4k.core.HttpHandler
import org.http4k.core.MemoryBody
import org.http4k.core.Method.GET
import org.http4k.core.Request
import org.http4k.core.Response
import org.http4k.core.Status
import org.http4k.core.then
import org.http4k.routing.ResourceLoader
import org.http4k.routing.RoutingHttpHandler
import org.http4k.routing.bind
import org.http4k.routing.routes
import org.http4k.routing.static
internal class WebMapTileServer(
appProps: ApplicationProperties
) {
private val tilesLoader = ResourceLoader.Directory(appProps.serverDataDir.toFile().canonicalPath)
private val filters = listOf(
ETagChecker(),
).reduce(Filter::then)
private val routes: RoutingHttpHandler = routes(
"/tiles" bind GET to static(tilesLoader).withFilter(filters)
)
fun build(): HttpHandler {
return routes
}
private class ETagChecker : Filter {
override fun invoke(next: HttpHandler): HttpHandler = { request: Request ->
val requestedETag = request.header("If-None-Match")
var response = next(request)
val responseBytes = response.body.stream.use { it.readAllBytes() }
val currentETag = responseBytes.md5()
response = response.body(MemoryBody(responseBytes))
when {
currentETag == null -> {
// println("[${request.etagKey}] currentETag is null (requested: $requestedETag), executing request")
response
.header("Cache-Control", "no-cache, max-age=0, must-revalidate")
}
requestedETag == currentETag -> {
// println("[${request.etagKey}] currentETag == requested, returning 304 $requestedETag")
Response(Status.NOT_MODIFIED)
.header("Cache-Control", "no-cache, max-age=0, must-revalidate")
}
else -> {
// println("[${request.etagKey}] currentETag != requested, executing request and adding currentETag:$currentETag (requested:$requestedETag)")
response
.header("ETag", currentETag)
// .header("Cache-Control", "no-store")
.header("Cache-Control", "no-cache, max-age=0, must-revalidate")
}
}
}
}
companion object {
private fun ByteArray.md5(): String? {
return if (isEmpty()) {
null
} else {
MessageDigest.getInstance("MD5")
.digest(this)
.joinToString("") { "%02x".format(it) }
}
}
}
}
//import dev.adamko.kafkatorio.processor.config.ApplicationProperties
//import java.security.MessageDigest
//import org.http4k.core.Filter
//import org.http4k.core.HttpHandler
//import org.http4k.core.MemoryBody
//import org.http4k.core.Method.GET
//import org.http4k.core.Request
//import org.http4k.core.Response
//import org.http4k.core.Status
//import org.http4k.core.then
//import org.http4k.routing.ResourceLoader
//import org.http4k.routing.RoutingHttpHandler
//import org.http4k.routing.bind
//import org.http4k.routing.routes
//import org.http4k.routing.static
//
//internal class WebMapTileServer(
// appProps: ApplicationProperties
//) {
//
// private val tilesLoader = ResourceLoader.Directory(appProps.serverDataDir.toFile().canonicalPath)
//
//
// private val filters = listOf(
// ETagChecker(),
// ).reduce(Filter::then)
//
//
// private val routes: RoutingHttpHandler = routes(
// "/tiles" bind GET to static(tilesLoader).withFilter(filters)
// )
//
//
// fun build(): HttpHandler {
// return routes
// }
//
//
// private class ETagChecker : Filter {
//
// override fun invoke(next: HttpHandler): HttpHandler = { request: Request ->
// val requestedETag = request.header("If-None-Match")
//
// var response = next(request)
// val responseBytes = response.body.stream.use { it.readAllBytes() }
// val currentETag = responseBytes.md5()
//
// response = response.body(MemoryBody(responseBytes))
//
// when {
// currentETag == null -> {
//// println("[${request.etagKey}] currentETag is null (requested: $requestedETag), executing request")
// response
// .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
// }
// requestedETag == currentETag -> {
//// println("[${request.etagKey}] currentETag == requested, returning 304 $requestedETag")
// Response(Status.NOT_MODIFIED)
// .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
// }
// else -> {
//// println("[${request.etagKey}] currentETag != requested, executing request and adding currentETag:$currentETag (requested:$requestedETag)")
// response
// .header("ETag", currentETag)
//// .header("Cache-Control", "no-store")
// .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
// }
// }
// }
// }
//
//
// companion object {
// private fun ByteArray.md5(): String? {
// return if (isEmpty()) {
// null
// } else {
// MessageDigest.getInstance("MD5")
// .digest(this)
// .joinToString("") { "%02x".format(it) }
// }
// }
// }
//}

30
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebServer.kt
View file

@ -1,16 +1,16 @@
package dev.adamko.kafkatorio.processor.tileserver
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import org.http4k.server.Http4kServer
import org.http4k.server.PolyHandler
import org.http4k.server.Undertow
import org.http4k.server.asServer
internal fun webServer(
tileServer: WebMapTileServer,
wsServer: WebsocketServer,
appProps: ApplicationProperties,
): Http4kServer = PolyHandler(
tileServer.build(),
wsServer.build(),
).asServer(Undertow(appProps.webPort()))
//
//import dev.adamko.kafkatorio.processor.config.ApplicationProperties
//import org.http4k.server.Http4kServer
//import org.http4k.server.PolyHandler
//import org.http4k.server.Undertow
//import org.http4k.server.asServer
//
//internal fun webServer(
// tileServer: WebMapTileServer,
// wsServer: WebsocketServer,
// appProps: ApplicationProperties,
//): Http4kServer = PolyHandler(
// tileServer.build(),
// wsServer.build(),
//).asServer(Undertow(appProps.webPort()))

280
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebsocketServer.kt
View file

@ -1,141 +1,141 @@
package dev.adamko.kafkatorio.processor.tileserver
import kotlin.coroutines.resume
import kotlin.coroutines.resumeWithException
import kotlin.time.Duration.Companion.seconds
import kotlin.time.TimeSource
import kotlinx.coroutines.CoroutineExceptionHandler
import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.withIndex
import kotlinx.coroutines.isActive
import kotlinx.coroutines.job
import kotlinx.coroutines.launch
import kotlinx.coroutines.supervisorScope
import kotlinx.coroutines.suspendCancellableCoroutine
import org.http4k.routing.bind
import org.http4k.routing.websockets
import org.http4k.websocket.Websocket
import org.http4k.websocket.WsHandler
import org.http4k.websocket.WsMessage
import org.http4k.websocket.WsStatus
class WebsocketServer {
private val serverScope = CoroutineScope(
CoroutineName("WsServer") +
Dispatchers.IO +
SupervisorJob() +
CoroutineExceptionHandler { _, e ->
println("[ws.WsServer] handling exception ${e.message}")
}
)
private val clients: MutableSharedFlow<Websocket> = MutableSharedFlow(
replay = 100,
extraBufferCapacity = 100,
BufferOverflow.DROP_OLDEST,
)
private val messages: MutableSharedFlow<String> = MutableSharedFlow(
replay = 100,
extraBufferCapacity = 100,
BufferOverflow.DROP_OLDEST,
)
private fun bindClient(ws: Websocket) = serverScope.launch {
println("[ws.bindClient] ${ws.upgradeRequest.toMessage()}")
clients.emit(ws)
}
fun sendMessage(msg: String) = serverScope.launch {
println("[ws.sendMessage] $msg")
messages.emit(msg)
}
fun build(): WsHandler {
val startTime = TimeSource.Monotonic.markNow()
serverScope.launch {
supervisorScope {
clients
.withIndex()
.onEach { (id, ws) ->
createClient(id, ws)
}.launchIn(serverScope)
}
}
serverScope.launch {
while (isActive) {
delay(10.seconds)
sendMessage("elapsed ${startTime.elapsedNow()}")
}
}
return websockets(
"/ws/foo" bind { ws: Websocket ->
serverScope.launch { bindClient(ws) }
}
)
}
private fun createClient(id: Int, ws: Websocket) {
val clientJob = SupervisorJob(serverScope.coroutineContext.job)
serverScope.launch(CoroutineName("ws-client-$id") + clientJob) {
launch {
suspendCancellableCoroutine<WsStatus> { cont ->
println("[ws-client] entering suspendCancellableCoroutine")
cont.invokeOnCancellation { e: Throwable? ->
launch { sendMessage("[ws-client.invokeOnCompletion] closing client ws ${e?.message}") }
when (e) {
null -> ws.close(WsStatus.NORMAL)
else -> ws.close(WsStatus.ABNORMAL_CLOSE)
}
}
ws.onError { e: Throwable ->
launch { sendMessage("[ws-client.onError] client $id error [${e.message}]") }
if (!cont.isCompleted) {
cont.resumeWithException(e)
}
}
ws.onClose { status: WsStatus ->
launch { sendMessage("[ws-client.onClose] client $id disconnected [${status}]") }
if (!cont.isCompleted) {
cont.resume(status)
}
}
ws.onMessage { wsMsg ->
launch { sendMessage("$id says: ${wsMsg.bodyString()}") }
}
}
}
println("[ws-client] $id listening to messages")
messages
.onEach { msg ->
println("[ws-client] sending message to $id")
ws.send(WsMessage(msg))
}.launchIn(this)
}
}
}
//
//import kotlin.coroutines.resume
//import kotlin.coroutines.resumeWithException
//import kotlin.time.Duration.Companion.seconds
//import kotlin.time.TimeSource
//import kotlinx.coroutines.CoroutineExceptionHandler
//import kotlinx.coroutines.CoroutineName
//import kotlinx.coroutines.CoroutineScope
//import kotlinx.coroutines.Dispatchers
//import kotlinx.coroutines.SupervisorJob
//import kotlinx.coroutines.channels.BufferOverflow
//import kotlinx.coroutines.delay
//import kotlinx.coroutines.flow.MutableSharedFlow
//import kotlinx.coroutines.flow.launchIn
//import kotlinx.coroutines.flow.onEach
//import kotlinx.coroutines.flow.withIndex
//import kotlinx.coroutines.isActive
//import kotlinx.coroutines.job
//import kotlinx.coroutines.launch
//import kotlinx.coroutines.supervisorScope
//import kotlinx.coroutines.suspendCancellableCoroutine
//import org.http4k.routing.bind
//import org.http4k.routing.websockets
//import org.http4k.websocket.Websocket
//import org.http4k.websocket.WsHandler
//import org.http4k.websocket.WsMessage
//import org.http4k.websocket.WsStatus
//
//
//class WebsocketServer {
//
// private val serverScope = CoroutineScope(
// CoroutineName("WsServer") +
// Dispatchers.IO +
// SupervisorJob() +
// CoroutineExceptionHandler { _, e ->
// println("[ws.WsServer] handling exception ${e.message}")
// }
// )
//
//
// private val clients: MutableSharedFlow<Websocket> = MutableSharedFlow(
// replay = 100,
// extraBufferCapacity = 100,
// BufferOverflow.DROP_OLDEST,
// )
//
//
// private val messages: MutableSharedFlow<String> = MutableSharedFlow(
// replay = 100,
// extraBufferCapacity = 100,
// BufferOverflow.DROP_OLDEST,
// )
//
//
// private fun bindClient(ws: Websocket) = serverScope.launch {
// println("[ws.bindClient] ${ws.upgradeRequest.toMessage()}")
// clients.emit(ws)
// }
//
//
// fun sendMessage(msg: String) = serverScope.launch {
// println("[ws.sendMessage] $msg")
// messages.emit(msg)
// }
//
//
// fun build(): WsHandler {
// val startTime = TimeSource.Monotonic.markNow()
//
// serverScope.launch {
// supervisorScope {
// clients
// .withIndex()
// .onEach { (id, ws) ->
// createClient(id, ws)
// }.launchIn(serverScope)
// }
// }
//
// serverScope.launch {
// while (isActive) {
// delay(10.seconds)
// sendMessage("elapsed ${startTime.elapsedNow()}")
// }
// }
//
// return websockets(
// "/ws/foo" bind { ws: Websocket ->
// serverScope.launch { bindClient(ws) }
// }
// )
// }
//
//
// private fun createClient(id: Int, ws: Websocket) {
// val clientJob = SupervisorJob(serverScope.coroutineContext.job)
//
// serverScope.launch(CoroutineName("ws-client-$id") + clientJob) {
// launch {
// suspendCancellableCoroutine<WsStatus> { cont ->
// println("[ws-client] entering suspendCancellableCoroutine")
//
// cont.invokeOnCancellation { e: Throwable? ->
// launch { sendMessage("[ws-client.invokeOnCompletion] closing client ws ${e?.message}") }
// when (e) {
// null -> ws.close(WsStatus.NORMAL)
// else -> ws.close(WsStatus.ABNORMAL_CLOSE)
// }
// }
//
// ws.onError { e: Throwable ->
// launch { sendMessage("[ws-client.onError] client $id error [${e.message}]") }
// if (!cont.isCompleted) {
// cont.resumeWithException(e)
// }
// }
//
// ws.onClose { status: WsStatus ->
// launch { sendMessage("[ws-client.onClose] client $id disconnected [${status}]") }
// if (!cont.isCompleted) {
// cont.resume(status)
// }
// }
//
// ws.onMessage { wsMsg ->
// launch { sendMessage("$id says: ${wsMsg.bodyString()}") }
// }
// }
// }
//
// println("[ws-client] $id listening to messages")
// messages
// .onEach { msg ->
// println("[ws-client] sending message to $id")
// ws.send(WsMessage(msg))
// }.launchIn(this)
// }
// }
//}

2
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/ApplicationProperties.kt
View file

@ -1,4 +1,4 @@
package dev.adamko.kafkatorio.processor.config
package dev.adamko.kafkatorio.server.config
import com.sksamuel.hoplite.ConfigLoaderBuilder
import com.sksamuel.hoplite.addEnvironmentSource

6
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/PortVal.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/PortVal.kt
View file

@ -1,9 +1,7 @@
package dev.adamko.kafkatorio.processor.config
import org.http4k.cloudnative.env.Port
package dev.adamko.kafkatorio.server.config
/** Improved (slightly) version of http4k's [Port]. */
/** Improved (slightly) version of http4k's `Port`. */
@JvmInline
value class PortVal(val value: Int) {

5
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsJson.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/kxsJson.kt
View file

@ -1,8 +1,9 @@
package dev.adamko.kafkatorio.processor.serdes
package dev.adamko.kafkatorio.server.config
import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio
import dev.adamko.kafkatorio.library.jsonMapperKafkatorio
import kotlinx.serialization.json.Json
val jsonMapper: Json = Json {
prettyPrint = true
prettyPrintIndent = " "

32
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/eventsServer.kt Normal file
View file

@ -0,0 +1,32 @@
package dev.adamko.kafkatorio.server
import dev.adamko.kafkatorio.server.config.ApplicationProperties
import dev.adamko.kafkatorio.server.socket.SyslogSocketServer
import dev.adamko.kafkatorio.server.web.startWebServer
import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.launch
suspend fun main(): Unit = coroutineScope {
val appProps = ApplicationProperties.load()
val websocketServer = WebmapWebsocketServer()
val syslogSocketServer = SyslogSocketServer(appProps)
launch {
startWebServer(appProps, websocketServer)
}
launch {
syslogSocketServer.start()
}
// launch(Dispatchers.Default.limitedParallelism(2)) {
// eventsProcessor(appProps, syslogSocketServer, websocketServer)
// }
}

38
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/kafkatorioEventsServer.kt
View file

@ -1,38 +0,0 @@
package dev.adamko.kafkatorio.server
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import io.ktor.http.HttpStatusCode
import io.ktor.server.application.Application
import io.ktor.server.application.install
import io.ktor.server.cio.CIO
import io.ktor.server.engine.embeddedServer
import io.ktor.server.plugins.callloging.CallLogging
import io.ktor.server.plugins.statuspages.StatusPages
import io.ktor.server.response.respondText
fun main() {
val appProps = ApplicationProperties.load()
embeddedServer(CIO, port = appProps.webPort.value) {
common()
webmapWebsocketServer()
webmapTileServer(appProps)
syslogSocketServer(appProps)
}.start(wait = true)
}
fun Application.common() {
install(StatusPages) {
exception<Throwable> { call, cause ->
call.respondText(text = "500: $cause", status = HttpStatusCode.InternalServerError)
}
}
install(CallLogging)
}

2
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/DebounceProcessor.kt
View file

@ -1,4 +1,4 @@
package dev.adamko.kafkatorio.processor.misc
package dev.adamko.kafkatorio.server.processor
import dev.adamko.kotka.extensions.component1
import dev.adamko.kotka.extensions.component2

21
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/KafkatorioTopology.kt
View file

@ -1,15 +1,12 @@
package dev.adamko.kafkatorio.processor
package dev.adamko.kafkatorio.server.processor
import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
import dev.adamko.kafkatorio.processor.admin.TOPIC_SRC_SERVER_LOG
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import dev.adamko.kafkatorio.processor.tileserver.SyslogSocketServer
import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer
import dev.adamko.kafkatorio.processor.topology.broadcastToWebsocket
import dev.adamko.kafkatorio.processor.topology.colourMapChunks
import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream
import dev.adamko.kafkatorio.processor.topology.saveMapTiles
import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
import dev.adamko.kafkatorio.server.config.ApplicationProperties
import dev.adamko.kafkatorio.server.processor.topology.colourMapChunks
import dev.adamko.kafkatorio.server.processor.topology.*
import dev.adamko.kafkatorio.server.socket.SyslogSocketServer
import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer
import dev.adamko.kafkatorio.server.web.websocket.broadcastToWebsocket
import java.util.Properties
import kotlin.coroutines.resume
import kotlin.coroutines.resumeWithException
@ -40,7 +37,7 @@ import org.apache.kafka.streams.TopologyDescription
internal class KafkatorioTopology(
private val appProps: ApplicationProperties,
private val websocketServer: WebsocketServer,
private val wsServer: WebmapWebsocketServer,
private val syslogServer: SyslogSocketServer,
) {
@ -86,7 +83,7 @@ internal class KafkatorioTopology(
private suspend fun websocketBroadcaster() {
val builder = StreamsBuilder()
broadcastToWebsocket(websocketServer, builder)
broadcastToWebsocket(wsServer, builder)
val topology = builder.build()

2
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/StreamsExceptionHandler.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/StreamsExceptionHandler.kt
View file

@ -1,4 +1,4 @@
package dev.adamko.kafkatorio.processor
package dev.adamko.kafkatorio.server.processor
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse

5
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/KafkatorioKafkaAdmin.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/admin/KafkatorioKafkaAdmin.kt
View file

@ -1,6 +1,7 @@
package dev.adamko.kafkatorio.processor.admin
package dev.adamko.kafkatorio.server.processor.admin
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import dev.adamko.kafkatorio.server.processor.allTopics
import dev.adamko.kafkatorio.server.config.ApplicationProperties
import java.util.concurrent.TimeUnit
import mu.KotlinLogging
import org.apache.kafka.clients.admin.Admin

37
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/eventProcessor.kt Normal file
View file

@ -0,0 +1,37 @@
package dev.adamko.kafkatorio.server.processor
import dev.adamko.kafkatorio.server.socket.SyslogSocketServer
import dev.adamko.kafkatorio.server.processor.admin.KafkatorioKafkaAdmin
import dev.adamko.kafkatorio.server.config.ApplicationProperties
import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer
import kotlinx.coroutines.awaitCancellation
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.launch
suspend fun eventsProcessor(
appProps: ApplicationProperties,
syslogServer: SyslogSocketServer,
wsServer: WebmapWebsocketServer,
): Unit = coroutineScope {
val admin = KafkatorioKafkaAdmin(appProps)
admin.createKafkatorioTopics()
// val syslogServer = SyslogSocketServer(appProps)
// val wsServer = WebsocketServer()
// val tileServer = WebMapTileServer(appProps)
val topology = KafkatorioTopology(appProps, wsServer, syslogServer)
launch {
topology.start()
println("launched KafkatorioTopology")
awaitCancellation()
}
}

2
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/kafkatorioTopics.kt
View file

@ -1,4 +1,4 @@
package dev.adamko.kafkatorio.processor.admin
package dev.adamko.kafkatorio.server.processor
import dev.adamko.kafkatorio.schema.packets.ConfigurationUpdate
import dev.adamko.kafkatorio.schema.packets.ConsoleChatUpdate

10
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourMapChunks.kt
View file

@ -1,9 +1,9 @@
package dev.adamko.kafkatorio.processor.topology
package dev.adamko.kafkatorio.server.processor.topology
import dev.adamko.kafkatorio.processor.admin.TOPIC_MAP_CHUNK_COLOURED_032_STATE
import dev.adamko.kafkatorio.processor.admin.TOPIC_MAP_CHUNK_COLOURED_STATE
import dev.adamko.kafkatorio.processor.misc.DebounceProcessor.Companion.addDebounceProcessor
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_COLOURED_032_STATE
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_COLOURED_STATE
import dev.adamko.kafkatorio.server.processor.DebounceProcessor.Companion.addDebounceProcessor
import dev.adamko.kafkatorio.library.kxsBinary
import dev.adamko.kafkatorio.schema.common.ChunkSize
import dev.adamko.kafkatorio.schema.common.ColourHex
import dev.adamko.kafkatorio.schema.common.FactorioServerId

9
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt
View file

@ -1,9 +1,8 @@
package dev.adamko.kafkatorio.processor.topology
package dev.adamko.kafkatorio.server.processor.topology
import dev.adamko.kafkatorio.processor.admin.TOPIC_SRC_SERVER_LOG
import dev.adamko.kafkatorio.processor.admin.topicName
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
import dev.adamko.kafkatorio.server.processor.TOPIC_SRC_SERVER_LOG
import dev.adamko.kafkatorio.server.processor.topicName
import dev.adamko.kafkatorio.server.config.jsonMapper
import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kafkatorio.schema.common.Tick
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket

2
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/model.kt
View file

@ -1,4 +1,4 @@
package dev.adamko.kafkatorio.processor.topology
package dev.adamko.kafkatorio.server.processor.topology
import com.sksamuel.scrimage.color.RGBColor
import dev.adamko.kafkatorio.schema.common.ColourHex

6
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/packetDataStreams.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/packetDataStreams.kt
View file

@ -1,7 +1,7 @@
package dev.adamko.kafkatorio.processor.topology
package dev.adamko.kafkatorio.server.processor.topology
import dev.adamko.kafkatorio.processor.admin.topicName
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
import dev.adamko.kafkatorio.server.processor.topicName
import dev.adamko.kafkatorio.server.config.jsonMapper
import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData

4
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/prototypes.kt
View file

@ -1,6 +1,6 @@
package dev.adamko.kafkatorio.processor.topology
package dev.adamko.kafkatorio.server.processor.topology
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
import dev.adamko.kafkatorio.library.kxsBinary
import dev.adamko.kafkatorio.schema.common.ColourHex
import dev.adamko.kafkatorio.schema.common.FactorioPrototype
import dev.adamko.kafkatorio.schema.common.FactorioServerId

10
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/saveMapTiles.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/saveMapTiles.kt
View file

@ -1,12 +1,12 @@
package dev.adamko.kafkatorio.processor.topology
package dev.adamko.kafkatorio.server.processor.topology
import com.sksamuel.scrimage.ImmutableImage
import com.sksamuel.scrimage.ScaleMethod
import com.sksamuel.scrimage.nio.PngWriter
import dev.adamko.kafkatorio.processor.admin.TOPIC_BROADCAST_TO_WEBSOCKET
import dev.adamko.kafkatorio.processor.admin.TOPIC_MAP_CHUNK_COLOURED_STATE
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
import dev.adamko.kafkatorio.server.processor.TOPIC_BROADCAST_TO_WEBSOCKET
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_COLOURED_STATE
import dev.adamko.kafkatorio.server.config.jsonMapper
import dev.adamko.kafkatorio.library.kxsBinary
import dev.adamko.kafkatorio.schema.common.ColourHex
import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kafkatorio.schema.common.ServerMapChunkId

15
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/websocketBroadcast.kt → modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/websocketBroadcaster.kt
View file

@ -1,10 +1,9 @@
package dev.adamko.kafkatorio.processor.topology
package dev.adamko.kafkatorio.server.web.websocket
import dev.adamko.kafkatorio.processor.admin.TOPIC_BROADCAST_TO_WEBSOCKET
import dev.adamko.kafkatorio.processor.admin.topicName
import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer
import dev.adamko.kafkatorio.library.jsonMapperKafkatorio
import dev.adamko.kafkatorio.server.processor.TOPIC_BROADCAST_TO_WEBSOCKET
import dev.adamko.kafkatorio.server.processor.topicName
import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio
import dev.adamko.kafkatorio.schema.packets.EventServerPacket
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
import dev.adamko.kafkatorio.schema.packets.PlayerUpdate
@ -32,8 +31,8 @@ private const val pid = "websocketBroadcast"
fun broadcastToWebsocket(
websocketServer: WebsocketServer,
builder: StreamsBuilder = StreamsBuilder(),
websocketServer: WebmapWebsocketServer,
builder: StreamsBuilder,
) {
builder.broadcastKafkatorioPackets()
@ -45,7 +44,7 @@ fun broadcastToWebsocket(
"$pid.websocket-server.send-message"
) { _, esPacket ->
runBlocking {
websocketServer.sendMessage(jsonMapper.encodeToString(esPacket))
websocketServer.sendMessageToClients(jsonMapper.encodeToString(esPacket))
}
}
}

63
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/socket/SyslogMsg.kt Normal file
View file

@ -0,0 +1,63 @@
package dev.adamko.kafkatorio.server.socket
import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
data class SyslogMsg(
val src: Map<String, String?>
) {
val message: String?
get() = src[SyslogFieldKeys.MESSAGE.field]
val headerAppName: String?
get() = src[SyslogFieldKeys.HEADER_APPNAME.field]
val headerHostName: String?
get() = src[SyslogFieldKeys.HEADER_HOSTNAME.field]
val headerPri: String?
get() = src[SyslogFieldKeys.HEADER_PRI.field]
val headerPriSeverity: String?
get() = src[SyslogFieldKeys.HEADER_PRI_SEVERITY.field]
val headerPriFacility: String?
get() = src[SyslogFieldKeys.HEADER_PRI_FACILITY.field]
val headerProcId: String?
get() = src[SyslogFieldKeys.HEADER_PROCID.field]
val headerTimestamp: String?
get() = src[SyslogFieldKeys.HEADER_TIMESTAMP.field]
val headerMsgId: String?
get() = src[SyslogFieldKeys.HEADER_MSGID.field]
val headerVersion: String?
get() = src[SyslogFieldKeys.HEADER_VERSION.field]
val structuredBase: String?
get() = src[SyslogFieldKeys.STRUCTURED_BASE.field]
val structuredElementIdFmt: String?
get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_FMT.field]
val structuredElementIdPnameFmt: String?
get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_FMT.field]
val structuredElementIdPnamePattern: String?
get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_PATTERN.field]
companion object {
fun Map<String, Any?>.toSyslogMsg(): SyslogMsg {
val map = entries.mapNotNull { (k, v) ->
when (v) {
is String -> k to v
else -> null
}
}.toMap()
return SyslogMsg(map)
}
}
}

99
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/socket/SyslogSocketServer.kt Normal file
View file

@ -0,0 +1,99 @@
package dev.adamko.kafkatorio.server.socket
import com.github.palindromicity.syslog.AllowableDeviations
import com.github.palindromicity.syslog.SyslogParser
import com.github.palindromicity.syslog.SyslogParserBuilder
import com.github.palindromicity.syslog.SyslogSpecification
import dev.adamko.kafkatorio.server.config.ApplicationProperties
import dev.adamko.kafkatorio.server.socket.SyslogMsg.Companion.toSyslogMsg
import io.ktor.network.selector.SelectorManager
import io.ktor.network.sockets.Socket
import io.ktor.network.sockets.aSocket
import io.ktor.network.sockets.openReadChannel
import io.ktor.utils.io.readUTF8Line
import java.util.EnumSet
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
import kotlinx.coroutines.yield
class SyslogSocketServer(
appProps: ApplicationProperties,
) {
private val selectorManager: SelectorManager = SelectorManager(Dispatchers.IO)
private val serverSocketTcp = aSocket(selectorManager)
.tcp()
.bind(
appProps.socketServerHost,
appProps.socketServerPort,
)
private val syslogParser: SyslogParser = SyslogParserBuilder()
.withDeviations(EnumSet.of(AllowableDeviations.PRIORITY, AllowableDeviations.VERSION))
.forSpecification(SyslogSpecification.RFC_5424)
.build()
private val _messages: MutableSharedFlow<SyslogMsg> = MutableSharedFlow()
val messages: SharedFlow<SyslogMsg>
get() = _messages.asSharedFlow()
suspend fun start(): Unit = coroutineScope {
log("started listening: ${serverSocketTcp.localAddress}")
while (isActive) {
val socket: Socket = serverSocketTcp.accept()
log("new socket ${socket.description()}")
launch {
val channel = socket.openReadChannel()
try {
while (socket.socketContext.isActive) {
val line = runCatching {
channel.readUTF8Line()
}.fold(
onSuccess = { line ->
log("received '$line' from $socket")
line
},
onFailure = { e ->
log("reading line from socket failed ${socket.description()}, cause: $e")
null
},
)
if (line != null) {
val syslog = syslogParser.parseLine(line).toSyslogMsg()
_messages.emit(syslog)
}
yield()
}
} catch (e: Exception) {
log("closing socket ${socket.description()}, cause: $e")
withContext(Dispatchers.IO) { socket.close() }
}
}
yield()
}
}
companion object {
private fun log(msg: String) = println("[SyslogSocketServer] $msg")
private fun Socket.description(): String = "Socket[local:$localAddress, remote:$remoteAddress]"
}
}

152
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/syslogSocketServer.kt
View file

@ -1,152 +0,0 @@
package dev.adamko.kafkatorio.server
import com.github.palindromicity.syslog.AllowableDeviations
import com.github.palindromicity.syslog.SyslogParser
import com.github.palindromicity.syslog.SyslogParserBuilder
import com.github.palindromicity.syslog.SyslogSpecification
import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import dev.adamko.kafkatorio.server.SyslogMsg.Companion.toSyslogMsg
import io.ktor.network.selector.SelectorManager
import io.ktor.network.sockets.aSocket
import io.ktor.network.sockets.awaitClosed
import io.ktor.network.sockets.openReadChannel
import io.ktor.server.application.Application
import io.ktor.utils.io.readUTF8Line
import java.util.EnumSet
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.yield
fun Application.syslogSocketServer(appProps: ApplicationProperties) {
val server = SyslogSocketServer(appProps)
launch {
server.start()
}
}
class SyslogSocketServer(
appProps: ApplicationProperties,
) {
private val selectorManager: SelectorManager = SelectorManager(Dispatchers.IO)
private val serverSocketTcp = aSocket(selectorManager)
.tcp()
.bind(
appProps.socketServerHost,
appProps.socketServerPort,
)
private val syslogParser: SyslogParser = SyslogParserBuilder()
.withDeviations(EnumSet.of(AllowableDeviations.PRIORITY, AllowableDeviations.VERSION))
.forSpecification(SyslogSpecification.RFC_5424)
.build()
private val _messages: MutableSharedFlow<SyslogMsg> = MutableSharedFlow()
val messages: SharedFlow<SyslogMsg>
get() = _messages.asSharedFlow()
suspend fun start(): Unit = coroutineScope {
println("[SyslogSocketServer] started listening: ${serverSocketTcp.localAddress}")
while (isActive) {
val socket = serverSocketTcp.accept()
launch {
val channel = socket.openReadChannel()
try {
while (socket.socketContext.isActive) {
val line = runCatching { channel.readUTF8Line() }.getOrNull()
if (line != null) {
println("[SyslogSocketServer] received: $line")
val syslog = syslogParser.parseLine(line).toSyslogMsg()
_messages.emit(syslog)
}
yield()
}
} catch (e: Exception) {
socket.awaitClosed()
}
}
yield()
}
}
}
data class SyslogMsg(
val src: Map<String, String?>
) {
val message: String?
get() = src[SyslogFieldKeys.MESSAGE.field]
val headerAppName: String?
get() = src[SyslogFieldKeys.HEADER_APPNAME.field]
val headerHostName: String?
get() = src[SyslogFieldKeys.HEADER_HOSTNAME.field]
val headerPri: String?
get() = src[SyslogFieldKeys.HEADER_PRI.field]
val headerPriSeverity: String?
get() = src[SyslogFieldKeys.HEADER_PRI_SEVERITY.field]
val headerPriFacility: String?
get() = src[SyslogFieldKeys.HEADER_PRI_FACILITY.field]
val headerProcId: String?
get() = src[SyslogFieldKeys.HEADER_PROCID.field]
val headerTimestamp: String?
get() = src[SyslogFieldKeys.HEADER_TIMESTAMP.field]
val headerMsgId: String?
get() = src[SyslogFieldKeys.HEADER_MSGID.field]
val headerVersion: String?
get() = src[SyslogFieldKeys.HEADER_VERSION.field]
val structuredBase: String?
get() = src[SyslogFieldKeys.STRUCTURED_BASE.field]
val structuredElementIdFmt: String?
get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_FMT.field]
val structuredElementIdPnameFmt: String?
get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_FMT.field]
val structuredElementIdPnamePattern: String?
get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_PATTERN.field]
companion object {
fun Map<String, Any?>.toSyslogMsg(): SyslogMsg {
val map = entries.mapNotNull { (k, v) ->
when (v) {
is String -> k to v
else -> null
}
}.toMap()
return SyslogMsg(map)
}
}
}

121
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/factorioServersRoutes.kt Normal file
View file

@ -0,0 +1,121 @@
package dev.adamko.kafkatorio.server.web.rest
import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kafkatorio.schema.common.PlayerIndex
import dev.adamko.kafkatorio.schema.common.SurfaceIndex
import dev.adamko.kafkatorio.schema.common.TilePngFilename
import dev.adamko.kafkatorio.server.config.ApplicationProperties
import io.ktor.http.HttpStatusCode
import io.ktor.http.content.EntityTagVersion
import io.ktor.http.content.versions
import io.ktor.resources.Resource
import io.ktor.server.application.ApplicationCall
import io.ktor.server.application.application
import io.ktor.server.application.call
import io.ktor.server.application.log
import io.ktor.server.http.content.LastModifiedVersion
import io.ktor.server.resources.get
import io.ktor.server.response.respond
import io.ktor.server.response.respondFile
import io.ktor.server.routing.Routing
import io.ktor.util.combineSafe
import io.ktor.util.pipeline.PipelineContext
import kotlin.io.path.Path
import kotlinx.serialization.Serializable
fun Routing.factorioServersRoutes(
appProps: ApplicationProperties,
) {
get<FactorioServers> {
call.respond(HttpStatusCode.NotImplemented)
}
get<FactorioServer.MapData.Tile> { tile ->
handleTileRequest(tile, appProps)
}
}
private suspend fun PipelineContext<Unit, ApplicationCall>.handleTileRequest(
tile: FactorioServer.MapData.Tile,
appProps: ApplicationProperties,
) {
val tileFilename = TilePngFilename(
tile.serverId,
tile.surfaceIndex,
tile.zoomLevel,
tile.chunkX,
tile.chunkY,
)
val tileFile = appProps.serverDataDir.combineSafe(Path(tileFilename.value))
if (tileFile.exists()) {
call.respondFile(tileFile) {
versions = versions + listOf(
EntityTagVersion(tileFile.lastModified().hashCode().toString()),
LastModifiedVersion(tileFile.lastModified()),
)
}
} else {
call.respond(HttpStatusCode.NotFound)
}
}
@Serializable
@Resource("kafkatorio/data/servers")
private class FactorioServers
@Serializable
@Resource("kafkatorio/data/servers/{factorioServerId}")
data class FactorioServer(
val factorioServerId: FactorioServerId,
) {
@Serializable
@Resource("map")
data class MapData(
val server: FactorioServer,
) {
@Serializable
@Resource("tiles/s{surfaceIndex}/z{zoomLevel}/x{chunkX}/y{chunkY}.png")
data class Tile(
val mapData: MapData,
val surfaceIndex: SurfaceIndex,
val zoomLevel: Int,
val chunkX: Int,
val chunkY: Int,
)
@Serializable
@Resource("buildings/s{surfaceIndex}")
data class Buildings(
val mapData: MapData,
val surfaceIndex: SurfaceIndex,
// a list of buildings... (Sealed interface, different types?)
)
}
@Serializable
@Resource("players")
data class Players(
val server: FactorioServer,
)
@Serializable
@Resource("{playerId}")
data class Player(
val server: FactorioServer,
val playerIndex: PlayerIndex,
)
}
private val FactorioServer.MapData.Tile.serverId: FactorioServerId
get() = mapData.server.factorioServerId

50
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/restApi.kt Normal file
View file

@ -0,0 +1,50 @@
package dev.adamko.kafkatorio.server.web.rest
import dev.adamko.kafkatorio.server.config.ApplicationProperties
import io.ktor.http.ContentType
import io.ktor.http.HttpStatusCode
import io.ktor.serialization.kotlinx.json.json
import io.ktor.server.application.Application
import io.ktor.server.application.install
import io.ktor.server.application.log
import io.ktor.server.plugins.autohead.AutoHeadResponse
import io.ktor.server.plugins.callloging.CallLogging
import io.ktor.server.plugins.compression.Compression
import io.ktor.server.plugins.compression.excludeContentType
import io.ktor.server.plugins.compression.minimumSize
import io.ktor.server.plugins.conditionalheaders.ConditionalHeaders
import io.ktor.server.plugins.contentnegotiation.ContentNegotiation
import io.ktor.server.plugins.partialcontent.PartialContent
import io.ktor.server.plugins.statuspages.StatusPages
import io.ktor.server.resources.Resources
import io.ktor.server.response.respondText
import io.ktor.server.routing.routing
fun Application.webmapTileServer(appProps: ApplicationProperties) {
install(AutoHeadResponse)
install(CallLogging)
install(Compression) {
default()
minimumSize(1024)
excludeContentType(ContentType.Video.Any)
}
install(ConditionalHeaders)
install(ContentNegotiation) {
json()
}
install(PartialContent)
install(Resources)
install(StatusPages) {
exception<Throwable> { call, cause ->
this@webmapTileServer.log.error("$call caused exception - returning 500", cause)
call.respondText(text = "500: $cause", status = HttpStatusCode.InternalServerError)
}
}
routing {
factorioServersRoutes(appProps)
}
}

27
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/webServer.kt Normal file
View file

@ -0,0 +1,27 @@
package dev.adamko.kafkatorio.server.web
import dev.adamko.kafkatorio.server.config.ApplicationProperties
import dev.adamko.kafkatorio.server.web.rest.webmapTileServer
import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer
import dev.adamko.kafkatorio.server.web.websocket.webmapWebsocketServer
import io.ktor.server.cio.CIO
import io.ktor.server.engine.embeddedServer
import kotlinx.coroutines.coroutineScope
suspend fun startWebServer(
appProps: ApplicationProperties,
wsServer: WebmapWebsocketServer,
): Unit = coroutineScope {
embeddedServer(
factory = CIO,
port = appProps.webPort.value,
) {
webmapWebsocketServer(wsServer)
webmapTileServer(appProps)
}.start(wait = true)
}

112
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/WebmapWebsocketServer.kt Normal file
View file

@ -0,0 +1,112 @@
package dev.adamko.kafkatorio.server.web.websocket
import io.ktor.websocket.DefaultWebSocketSession
import io.ktor.websocket.Frame
import io.ktor.websocket.send
import java.util.concurrent.atomic.AtomicInteger
import kotlin.time.Duration.Companion.seconds
import kotlin.time.TimeSource
import kotlinx.coroutines.awaitCancellation
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.receiveAsFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.isActive
import kotlinx.coroutines.job
import kotlinx.coroutines.launch
import kotlinx.coroutines.yield
class WebmapWebsocketServer {
private val startTime = TimeSource.Monotonic.markNow()
private val clients: MutableStateFlow<List<WSConnection>> = MutableStateFlow(emptyList())
private val messagesToClients: MutableSharedFlow<String> = MutableSharedFlow(
extraBufferCapacity = Int.MAX_VALUE,
onBufferOverflow = BufferOverflow.DROP_OLDEST
)
private val messagesFromClients: MutableSharedFlow<Frame> = MutableSharedFlow(
extraBufferCapacity = Int.MAX_VALUE,
onBufferOverflow = BufferOverflow.DROP_OLDEST
)
suspend fun bindClient(session: DefaultWebSocketSession): Unit = coroutineScope {
val client = WSConnection(session)
clients.update { clients -> clients + client }
client.coroutineContext.job.invokeOnCompletion { e ->
log("client ${client.name} completed ${e ?: "successfully"}")
clients.update { clients ->
(clients - client).filter { it.isActive }
}
}
messagesToClients.onEach { msg ->
runCatching {
if (client.isActive) {
client.outgoing.send(Frame.Text(msg))
}
}
}.launchIn(client)
client.launch {
client.incoming.receiveAsFlow().collect(messagesFromClients)
}
// client.launch {
while (isActive) {
runCatching {
client.send("elapsed ${startTime.elapsedNow()}")
}
delay(10.seconds)
yield()
}
// }
// awaitCancellation()
}
suspend fun sendMessageToClients(msg: String) {
log("sending message to clients: $msg")
messagesToClients.emit(msg)
}
private data class WSConnection(
private val session: DefaultWebSocketSession,
) : DefaultWebSocketSession by session {
val name: String = "user${lastId.getAndIncrement()}"
override suspend fun send(frame: Frame): Unit = runCatching {
super.send(frame)
}.fold(
onSuccess = { log("sent message to client $name: $frame") },
onFailure = { e -> log("failed to send message to client $name: $e") }
)
override fun equals(other: Any?): Boolean = name == other
override fun hashCode(): Int = name.hashCode()
companion object {
private val lastId = AtomicInteger(0)
}
}
companion object {
private fun log(msg: String) = println("[WebmapWebsocketServer] $msg")
}
}

34
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/wsServer.kt Normal file
View file

@ -0,0 +1,34 @@
package dev.adamko.kafkatorio.server.web.websocket
import io.ktor.server.application.Application
import io.ktor.server.application.install
import io.ktor.server.routing.routing
import io.ktor.server.websocket.WebSockets
import io.ktor.server.websocket.webSocket
import io.ktor.websocket.WebSocketDeflateExtension
import java.util.zip.Deflater
fun Application.webmapWebsocketServer(
wsServer: WebmapWebsocketServer,
) {
install(WebSockets) {
extensions {
install(WebSocketDeflateExtension) {
// Compression level to use for java.util.zip.Deflater
compressionLevel = Deflater.DEFAULT_COMPRESSION
// Prevent compressing small outgoing frames
compressIfBiggerThan(bytes = 4 * 1024)
}
}
}
routing {
// TODO only send a ws client message for a single Factorio-sever. Make the URL server specific.
// /kafkatorio/data/servers/{serverId}/ws
webSocket("kafkatorio/ws") {
wsServer.bindClient(this)
}
}
}

148
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/webmapTileServer.kt
View file

@ -1,148 +0,0 @@
package dev.adamko.kafkatorio.server
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kafkatorio.schema.common.PlayerIndex
import dev.adamko.kafkatorio.schema.common.SurfaceIndex
import dev.adamko.kafkatorio.schema.common.TilePngFilename
import io.ktor.http.ContentType
import io.ktor.http.HttpStatusCode
import io.ktor.http.content.EntityTagVersion
import io.ktor.http.content.versions
import io.ktor.resources.Resource
import io.ktor.serialization.kotlinx.json.json
import io.ktor.server.application.Application
import io.ktor.server.application.ApplicationCall
import io.ktor.server.application.call
import io.ktor.server.application.install
import io.ktor.server.http.content.LastModifiedVersion
import io.ktor.server.plugins.autohead.AutoHeadResponse
import io.ktor.server.plugins.compression.Compression
import io.ktor.server.plugins.compression.excludeContentType
import io.ktor.server.plugins.compression.minimumSize
import io.ktor.server.plugins.conditionalheaders.ConditionalHeaders
import io.ktor.server.plugins.contentnegotiation.ContentNegotiation
import io.ktor.server.plugins.partialcontent.PartialContent
import io.ktor.server.resources.Resources
import io.ktor.server.resources.get
import io.ktor.server.response.respond
import io.ktor.server.response.respondFile
import io.ktor.server.routing.routing
import io.ktor.util.combineSafe
import io.ktor.util.pipeline.PipelineContext
import kotlin.io.path.Path
import kotlinx.serialization.Serializable
fun Application.webmapTileServer(appProps: ApplicationProperties) {
install(AutoHeadResponse)
install(Compression) {
default()
minimumSize(1024)
excludeContentType(ContentType.Video.Any)
}
install(ConditionalHeaders)
install(ContentNegotiation) {
json()
}
install(PartialContent)
install(Resources)
val server = WebmapTileServer(appProps)
routing {
with(server) {
get<FactorioServers.FactorioServer.MapData.Tile> { tile ->
handleTileRequest(tile)
}
get<FactorioServers> {
}
}
}
}
private class WebmapTileServer(
private val appProps: ApplicationProperties
) {
private val serverDataDir by appProps::serverDataDir
suspend fun PipelineContext<Unit, ApplicationCall>.handleTileRequest(
tile: FactorioServers.FactorioServer.MapData.Tile
) {
val tileFilename = TilePngFilename(
tile.serverId,
tile.surfaceIndex,
tile.zoomLevel,
tile.chunkX,
tile.chunkY,
)
val tileFile = serverDataDir.combineSafe(Path(tileFilename.value))
if (tileFile.exists()) {
call.respondFile(tileFile) {
versions = versions + listOf(
EntityTagVersion(tileFile.lastModified().hashCode().toString()),
LastModifiedVersion(tileFile.lastModified()),
)
}
} else {
call.respond(HttpStatusCode.NotFound)
}
}
}
@Serializable
@Resource("servers")
private class FactorioServers {
@Serializable
@Resource("{factorioServerId}")
data class FactorioServer(
val factorioServerId: FactorioServerId,
) {
@Serializable
@Resource("map")
class MapData {
@Serializable
@Resource("tiles/s{surfaceIndex}/z{zoomLevel}/x{chunkX}/y{chunkY}.png")
data class Tile(
val parent: FactorioServer,
val surfaceIndex: SurfaceIndex,
val zoomLevel: Int,
val chunkX: Int,
val chunkY: Int,
) {
val serverId: FactorioServerId by parent::factorioServerId
}
@Serializable
@Resource("buildings/s{surfaceIndex}")
data class Buildings(
val surfaceIndex: SurfaceIndex,
// a list of buildings... (Sealed interface, different types?)
)
}
@Serializable
@Resource("players")
class Players {
@Serializable
@Resource("{playerId}")
data class Id(
val playerIndex: PlayerIndex,
)
}
}
}

130
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/webmapWebsocketServer.kt
View file

@ -1,130 +0,0 @@
package dev.adamko.kafkatorio.server
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import io.ktor.server.application.Application
import io.ktor.server.application.install
import io.ktor.server.routing.routing
import io.ktor.server.websocket.WebSockets
import io.ktor.server.websocket.webSocket
import io.ktor.websocket.DefaultWebSocketSession
import io.ktor.websocket.Frame
import io.ktor.websocket.WebSocketDeflateExtension
import io.ktor.websocket.send
import java.util.concurrent.atomic.AtomicInteger
import java.util.zip.Deflater
import kotlin.time.Duration.Companion.seconds
import kotlin.time.TimeSource
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.receiveAsFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.isActive
import kotlinx.coroutines.job
import kotlinx.coroutines.launch
fun Application.webmapWebsocketServer() {
install(WebSockets) {
extensions {
install(WebSocketDeflateExtension) {
// Compression level to use for java.util.zip.Deflater
compressionLevel = Deflater.DEFAULT_COMPRESSION
// Prevent compressing small outgoing frames
compressIfBiggerThan(bytes = 4 * 1024)
}
}
}
val wsServer = WebmapWebsocketServer()
routing {
webSocket("/ws") {
wsServer.bindClient(this)
}
}
}
private class WebmapWebsocketServer {
private val startTime = TimeSource.Monotonic.markNow()
private val clients: MutableStateFlow<Set<WSConnection>> = MutableStateFlow(emptySet())
private val messagesToClients: MutableSharedFlow<String> = MutableSharedFlow(
extraBufferCapacity = Int.MAX_VALUE,
onBufferOverflow = BufferOverflow.DROP_OLDEST
)
private val messagesFromClients: MutableSharedFlow<Frame> = MutableSharedFlow(
extraBufferCapacity = Int.MAX_VALUE,
onBufferOverflow = BufferOverflow.DROP_OLDEST
)
suspend fun bindClient(session: DefaultWebSocketSession) {
val client = WSConnection(session)
clients.update { clients -> clients + client }
messagesToClients.onEach { msg ->
runCatching {
client.send(msg)
}
}.launchIn(client)
client.launch {
client.incoming.receiveAsFlow()
.collect(messagesFromClients)
}
client.launch {
while (isActive) {
runCatching {
client.send("elapsed ${startTime.elapsedNow()}")
}
delay(10.seconds)
}
}
client.coroutineContext.job.invokeOnCompletion {
log("client ${client.name} completed")
clients.update { clients -> clients - client }
}
}
suspend fun sendMessageToClients(msg: String) {
log("sending message to clients: $msg")
messagesToClients.emit(msg)
}
}
private data class WSConnection(
private val session: DefaultWebSocketSession,
) : DefaultWebSocketSession by session {
val name: String = "user${lastId.getAndIncrement()}"
override suspend fun send(frame: Frame): Unit = runCatching {
super.send(frame)
}.fold(
onSuccess = { log("sent message to client $name: $frame") },
onFailure = { log("failed to send message to client $name: $frame") }
)
override fun equals(other: Any?): Boolean = name == other
override fun hashCode(): Int = name.hashCode()
companion object {
private val lastId = AtomicInteger(0)
}
}
private fun log(msg: String) = println("[WebmapWebsocketServer] $msg")

4
modules/events-server/src/main/resources/config.yml
View file

@ -1,11 +1,11 @@
# Factorio Events Processor config
web-port: ${SERVER_PORT:-8080}
servers-data-dir: ${SERVER_DATA_DIR:-./server-data}
server-data-dir: ${SERVER_DATA_DIR:-./server-data}
socket-server-host: ${SOCKET_SERVER_HOST:-127.0.0.1}
socket-server-port: ${SOCKET_SERVER_PORT:-9002}
socket-server-port: ${SOCKET_SERVER_PORT:-12092}
kafka-admin-config:

49
modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessorTest.kt
View file

@ -1,49 +0,0 @@
package dev.adamko.kafkatorio.processor.misc
import io.kotest.core.spec.style.FunSpec
import io.kotest.core.spec.style.Test
import io.mockk.every
import io.mockk.mockk
import kotlin.time.Duration.Companion.minutes
import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.Topology
import org.apache.kafka.streams.state.KeyValueIterator
import org.apache.kafka.streams.state.TimestampedKeyValueStore
class DebounceProcessorTest : FunSpec({
}) {
@Test
fun blah() {
// val topology = Topology()
//
// topology.addProcessor(
// "blah",
// debounceProcessor(
// 1.minutes,
// "asdasda",
// Serdes.String(),
// Serdes.String()
// )
// )
// val streamsBuilder = StreamsBuilder()
//
// streamsBuilder
// .stream<String, String>("input-topic")
// .process(
// debounceProcessor(
// 1.minutes,
// "input-topic-debounce-store",
// Serdes.String(),
// Serdes.String(),
// ),
// )
}
}

319
modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/KxsBinaryTests.kt
View file

@ -1,319 +0,0 @@
package dev.adamko.kafkatorio.processor.misc
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles
import dev.adamko.kafkatorio.schema.common.ChunkSize
import dev.adamko.kafkatorio.schema.common.ColourHex
import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kafkatorio.schema.common.MapChunkPosition
import dev.adamko.kafkatorio.schema.common.MapTilePosition
import dev.adamko.kafkatorio.schema.common.SurfaceIndex
import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio
import io.kotest.assertions.withClue
import io.kotest.core.spec.style.FunSpec
import io.kotest.matchers.collections.shouldContainExactly
import io.kotest.matchers.comparables.shouldBeEqualComparingTo
import io.kotest.matchers.equality.shouldBeEqualToComparingFields
import io.kotest.matchers.ints.shouldBeExactly
import io.kotest.matchers.longs.shouldBeExactly
import io.kotest.matchers.shouldBe
import io.kotest.property.Arb
import io.kotest.property.arbitrary.boolean
import io.kotest.property.arbitrary.booleanArray
import io.kotest.property.arbitrary.byte
import io.kotest.property.arbitrary.byteArray
import io.kotest.property.arbitrary.char
import io.kotest.property.arbitrary.charArray
import io.kotest.property.arbitrary.double
import io.kotest.property.arbitrary.doubleArray
import io.kotest.property.arbitrary.float
import io.kotest.property.arbitrary.floatArray
import io.kotest.property.arbitrary.int
import io.kotest.property.arbitrary.intArray
import io.kotest.property.arbitrary.long
import io.kotest.property.arbitrary.longArray
import io.kotest.property.arbitrary.positiveInt
import io.kotest.property.arbitrary.short
import io.kotest.property.arbitrary.shortArray
import io.kotest.property.arbitrary.string
import io.kotest.property.arbitrary.uByte
import io.kotest.property.arbitrary.uInt
import io.kotest.property.arbitrary.uLong
import io.kotest.property.arbitrary.uShort
import io.kotest.property.checkAll
import kotlinx.serialization.builtins.serializer
import kotlinx.serialization.decodeFromByteArray
import kotlinx.serialization.decodeFromString
import kotlinx.serialization.encodeToByteArray
import kotlinx.serialization.encodeToString
class KxsBinaryTests : FunSpec({
test("byte array") {
val ba = byteArrayOf(
4, 116, 101, 115, 116, 2, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 1, 0, 0, 0, 3
)
val result = kxsBinary.decodeFromByteArray(ServerMapChunkId.serializer(), ba)
withClue(ba.toString(Charsets.UTF_8)) {
result shouldBeEqualToComparingFields ServerMapChunkId(
serverId = FactorioServerId("test"),
chunkPosition = MapChunkPosition(1, 2),
surfaceIndex = SurfaceIndex(1u),
chunkSize = ChunkSize.CHUNK_064,
)
}
}
test("encoding") {
val ba = kxsBinary.encodeToByteArray(
ServerMapChunkId(
serverId = FactorioServerId("test"),
chunkPosition = MapChunkPosition(1, 2),
surfaceIndex = SurfaceIndex(1u),
chunkSize = ChunkSize.CHUNK_064,
)
)
withClue(ba.toString(Charsets.UTF_8)) {
ba shouldBe byteArrayOf(
4, 116, 101, 115, 116, 2, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 1, 0, 0, 0, 3
)
}
}
context("inline value class") {
val uuid = "fa858d73-6205-44cf-aeeb-6bbf3148c4a7"
//@formatter:off
val expectedBytes = byteArrayOf(
36, 102, 97, 56, 53, 56, 100, 55, 51, 45, 54, 50, 48, 53, 45, 52, 52, 99,
102, 45, 97, 101, 101, 98, 45, 54, 98, 98, 102, 51, 49, 52, 56, 99, 52, 97, 55,
)
//@formatter:on
test("encode") {
val ba = kxsBinary.encodeToByteArray(FactorioServerId(uuid))
withClue(ba.toString(Charsets.UTF_8)) {
ba shouldBe expectedBytes
}
}
test("decode") {
val ba = kxsBinary.encodeToByteArray(FactorioServerId(uuid))
val id: FactorioServerId = kxsBinary.decodeFromByteArray(ba)
id shouldBeEqualToComparingFields FactorioServerId(uuid)
}
}
context("ServerMapChunkTiles") {
val initial = ServerMapChunkTiles(
ServerMapChunkId(
FactorioServerId("server-id"),
MapChunkPosition(33, 44),
SurfaceIndex(4u),
ChunkSize.CHUNK_032,
),
mapOf(
MapTilePosition(1, 2) to ColourHex(1u, 2u, 3u, 4u),
)
)
test("!json round trip") {
val jsonEncoded = jsonMapperKafkatorio.encodeToString(initial)
val jsonDecoded =
jsonMapperKafkatorio.decodeFromString<ServerMapChunkTiles<ColourHex>>(jsonEncoded)
jsonDecoded shouldBeEqualToComparingFields initial
}
test("binary round trip") {
val binaryEncoded = kxsBinary.encodeToByteArray(initial)
val binaryDecoded =
kxsBinary.decodeFromByteArray<ServerMapChunkTiles<ColourHex>>(binaryEncoded)
binaryDecoded shouldBeEqualToComparingFields initial
}
}
context("primitives round trip") {
test("double") {
checkAll(Arb.double()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: Double = kxsBinary.decodeFromByteArray(encoded)
decoded shouldBeEqualComparingTo input
}
}
test("float") {
checkAll(Arb.float()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: Float = kxsBinary.decodeFromByteArray(encoded)
decoded shouldBeEqualComparingTo input
}
}
test("int") {
checkAll(Arb.int()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: Int = kxsBinary.decodeFromByteArray(encoded)
decoded shouldBeExactly input
}
}
test("long") {
checkAll(Arb.long()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: Long = kxsBinary.decodeFromByteArray(encoded)
decoded shouldBeExactly input
}
}
test("short") {
checkAll(Arb.short()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: Short = kxsBinary.decodeFromByteArray(encoded)
decoded shouldBeEqualComparingTo input
}
}
test("byte") {
checkAll(Arb.byte()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: Byte = kxsBinary.decodeFromByteArray(encoded)
decoded shouldBeEqualComparingTo input
}
}
test("boolean") {
checkAll(Arb.boolean()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: Boolean = kxsBinary.decodeFromByteArray(encoded)
decoded shouldBeEqualComparingTo input
}
}
test("char") {
checkAll(Arb.char()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: Char = kxsBinary.decodeFromByteArray(encoded)
decoded shouldBeEqualComparingTo input
}
}
test("string") {
checkAll(Arb.string()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: String = kxsBinary.decodeFromByteArray(encoded)
decoded shouldBeEqualComparingTo input
}
}
}
@OptIn(ExperimentalUnsignedTypes::class)
context("unsigned numbers round trip") {
test("uint") {
checkAll(Arb.uInt()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(UInt.serializer(), input)
val decoded: UInt = kxsBinary.decodeFromByteArray(UInt.serializer(), encoded)
decoded shouldBeEqualComparingTo input
}
}
test("ulong") {
checkAll(Arb.uLong()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(ULong.serializer(), input)
val decoded: ULong = kxsBinary.decodeFromByteArray(ULong.serializer(), encoded)
decoded shouldBeEqualComparingTo input
}
}
test("ushort") {
checkAll(Arb.uShort()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(UShort.serializer(), input)
val decoded: UShort = kxsBinary.decodeFromByteArray(UShort.serializer(), encoded)
decoded shouldBeEqualComparingTo input
}
}
test("ubyte") {
checkAll(Arb.uByte()) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(UByte.serializer(), input)
val decoded: UByte = kxsBinary.decodeFromByteArray(UByte.serializer(), encoded)
decoded shouldBeEqualComparingTo input
}
}
}
context("primitive arrays round trip") {
test("double array") {
checkAll(Arb.doubleArray(Arb.positiveInt(50), Arb.double())) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: DoubleArray = kxsBinary.decodeFromByteArray(encoded)
decoded.asList() shouldContainExactly input.asList()
}
}
test("float array") {
checkAll(Arb.floatArray(Arb.positiveInt(50), Arb.float())) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: FloatArray = kxsBinary.decodeFromByteArray(encoded)
decoded.asList() shouldContainExactly input.asList()
}
}
test("int array") {
checkAll(Arb.intArray(Arb.positiveInt(50), Arb.int())) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: IntArray = kxsBinary.decodeFromByteArray(encoded)
decoded.asList() shouldContainExactly input.asList()
}
}
test("long array") {
checkAll(Arb.longArray(Arb.positiveInt(50), Arb.long())) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: LongArray = kxsBinary.decodeFromByteArray(encoded)
decoded.asList() shouldContainExactly input.asList()
}
}
test("short array") {
checkAll(Arb.shortArray(Arb.positiveInt(50), Arb.short())) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: ShortArray = kxsBinary.decodeFromByteArray(encoded)
decoded.asList() shouldContainExactly input.asList()
}
}
test("byte array") {
checkAll(Arb.byteArray(Arb.positiveInt(50), Arb.byte())) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: ByteArray = kxsBinary.decodeFromByteArray(encoded)
decoded.asList() shouldContainExactly input.asList()
}
}
test("boolean array") {
checkAll(Arb.booleanArray(Arb.positiveInt(50), Arb.boolean())) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: BooleanArray = kxsBinary.decodeFromByteArray(encoded)
decoded.asList() shouldContainExactly input.asList()
}
}
test("char array") {
checkAll(Arb.charArray(Arb.positiveInt(50), Arb.char())) { input ->
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
val decoded: CharArray = kxsBinary.decodeFromByteArray(encoded)
decoded.asList() shouldContainExactly input.asList()
}
}
}
})

6
modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt
View file

@ -1,14 +1,14 @@
package dev.adamko.kafkatorio.processor.misc
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
import dev.adamko.kafkatorio.library.kxsBinary
import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles
import dev.adamko.kafkatorio.schema.common.ChunkSize
import dev.adamko.kafkatorio.schema.common.ColourHex
import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kafkatorio.schema.common.MapChunkPosition
import dev.adamko.kafkatorio.schema.common.MapTilePosition
import dev.adamko.kafkatorio.schema.common.SurfaceIndex
import dev.adamko.kafkatorio.server.processor.topology.ServerMapChunkTiles
import java.util.UUID
import kotlinx.serialization.encodeToByteArray
import org.junit.jupiter.api.Test
@ -24,7 +24,7 @@ class Blah {
serverId = FactorioServerId(UUID.randomUUID().toString()),
chunkPosition = MapChunkPosition(Int.MAX_VALUE, Int.MAX_VALUE),
surfaceIndex = SurfaceIndex(UInt.MAX_VALUE),
chunkSize = ChunkSize.MAX
chunkSize = ChunkSize.MAX,
),
buildMap {
repeat(512) { x ->

41
modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/serdes/MapTileDictionaryTest.kt
View file

@ -1,41 +0,0 @@
package dev.adamko.kafkatorio.processor.serdes
import dev.adamko.kafkatorio.schema.common.MapTileDictionary
import dev.adamko.kafkatorio.schema.common.PrototypeName
import io.kotest.core.spec.style.FunSpec
import io.kotest.matchers.equality.shouldBeEqualToComparingFields
import kotlinx.serialization.decodeFromByteArray
import kotlinx.serialization.encodeToByteArray
class MapTileDictionaryTest : FunSpec({
context("encode MapTileDictionary") {
val actual = MapTileDictionary(
tilesXY = mapOf(
"-192" to mapOf("-65" to MapTileDictionary.PrototypeKey(2)),
"161" to mapOf(
"-96" to MapTileDictionary.PrototypeKey(1),
"-95" to MapTileDictionary.PrototypeKey(2),
"94" to MapTileDictionary.PrototypeKey(2),
"93" to MapTileDictionary.PrototypeKey(32),
),
// unknown prototype
"44" to mapOf("1" to MapTileDictionary.PrototypeKey(44)),
),
protos = mapOf(
PrototypeName("water") to MapTileDictionary.PrototypeKey(1),
PrototypeName("grass-4") to MapTileDictionary.PrototypeKey(2),
PrototypeName("grass-3") to MapTileDictionary.PrototypeKey(32),
PrototypeName("not-used") to MapTileDictionary.PrototypeKey(99),
)
)
test("binary round trip") {
val encoded: ByteArray = kxsBinary.encodeToByteArray(actual)
val decoded: MapTileDictionary = kxsBinary.decodeFromByteArray(encoded)
decoded.shouldBeEqualToComparingFields(actual)
}
}
})

6
modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt
View file

@ -1,8 +1,8 @@
package dev.adamko.kafkatorio.processor.topology
//import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
//import dev.adamko.kafkatorio.processor.serdes.jsonMapper
//import dev.adamko.kafkatorio.processor.serdes.kxsBinary
//import dev.adamko.kafkatorio.server.processor.TOPIC_GROUPED_MAP_CHUNKS_STATE
//import dev.adamko.kafkatorio.processor.config.getJsonMapper
//import dev.adamko.kafkatorio.library.getKxsBinary
//import dev.adamko.kafkatorio.schema.common.Colour
//import dev.adamko.kafkatorio.schema.common.ColourHex
//import dev.adamko.kafkatorio.schema.common.MapChunkPosition

4
modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt → modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/processor/topology/SaveMapChunksTest.kt
View file

@ -1,6 +1,6 @@
package dev.adamko.kafkatorio.processor.topology
package dev.adamko.kafkatorio.server.processor.topology
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
import dev.adamko.kafkatorio.library.kxsBinary
import dev.adamko.kafkatorio.schema.common.ChunkSize
import dev.adamko.kafkatorio.schema.common.ColourHex
import dev.adamko.kafkatorio.schema.common.FactorioServerId