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

- serialization improvements (the tests work at least)

Browse source 
- fixing some version mis-alignments
This commit is contained in:
Adam 2021-12-01 13:35:32 +01:00
parent babe0161ae
commit 53862ef1b5
19 changed files with 305 additions and 260 deletions
Download patch file Download diff file Expand all files Collapse all files

21
modules/factorio-events-processor/build.gradle.kts
View file

@ -4,8 +4,7 @@ plugins {
application
}
description =
"""
description = """
Receive raw Factorio Events from Kafka and process them into targeted topics or data formats.
Factorio events -> Mod -> Kafka -> ***Factorio Events Processor*** -> processed events
@ -19,13 +18,23 @@ dependencies {
implementation("org.apache.kafka:kafka-streams:3.0.0")
implementation(libs.kotlinx.serialization)
implementation(projects.modules.factorioEventsDataModel)
// implementation("com.ionspin.kotlin:bignum:0.3.3")
implementation(enforcedPlatform(libs.kotlinx.serialization.bom))
implementation(libs.kotlinx.serialization.core)
implementation(libs.kotlinx.serialization.json)
implementation(platform(libs.http4k.bom))
implementation(libs.http4k.core)
// implementation(libs.http4k.core)
implementation(libs.http4k.cloudnative)
implementation(libs.http4k.format.json)
implementation(libs.http4k.format.yaml)
// implementation(libs.http4k.format.json)
implementation(libs.http4k.format.yaml) {
because("parsing yaml properties files")
}
implementation(libs.http4k.format.kotlinx)
implementation(libs.bundles.logging)

1
modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/config/ApplicationProperties.kt
View file

@ -9,7 +9,6 @@ import org.http4k.cloudnative.env.Environment
import org.http4k.cloudnative.env.EnvironmentKey
import org.http4k.core.Credentials
import org.http4k.core.Uri
import org.http4k.format.Jackson
import org.http4k.format.Json
import org.http4k.lens.boolean
import org.http4k.lens.map

8
modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/main.kt
View file

@ -1,7 +1,8 @@
package dev.adamko.factorioevents.processor
import dev.adamko.factorioevents.model.FactorioEvent
import dev.adamko.factorioevents.model.jsonMapper
import dev.adamko.factorioevents.processor.config.ApplicationProperties
import dev.adamko.factorioevents.processor.model.FactorioServerLogRecord
import java.time.Duration
import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.streams.KafkaStreams
@ -9,7 +10,7 @@ import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.TopologyDescription
import org.apache.kafka.streams.kstream.Consumed
import org.apache.kafka.streams.processor.TopicNameExtractor
import org.http4k.format.Jackson
import org.http4k.format.KotlinxSerialization
fun main() {
@ -23,7 +24,8 @@ fun main() {
)
.mapValues { readOnlyKey, value ->
println("Mapping $readOnlyKey:$value")
Jackson.asA<FactorioServerLogRecord<*>>(value)
// jsonMapper.decodeFromString<FactorioEvent<*>>(value)
KotlinxSerialization.asA<FactorioEvent<*>>(value)
}
.to(TopicNameExtractor { key, value, recordContext ->
value.data.objectName

61
modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/model/FactorioServerLogRecord.kt
View file

@ -1,61 +0,0 @@
package dev.adamko.factorioevents.processor.model
import com.fasterxml.jackson.annotation.JsonProperty
import com.fasterxml.jackson.annotation.JsonTypeInfo
import com.fasterxml.jackson.annotation.JsonTypeName
import com.fasterxml.jackson.annotation.JsonValue
import java.math.BigDecimal
import java.math.BigInteger
import kotlin.reflect.KClass
data class FactorioServerLogRecord<DataType : FactorioObjectData>(
@JsonProperty("mod_version")
val modVersion: String,
@JsonProperty("tick")
val tick: BigInteger,
@JsonProperty("event_type")
val eventType: String,
@JsonProperty("data")
val data: DataType,
)
@JsonTypeInfo(
use = JsonTypeInfo.Id.NAME,
include = JsonTypeInfo.As.EXISTING_PROPERTY,
property = "object_name",
defaultImpl = UnknownFactorioObject::class,
visible = true,
)
sealed class FactorioObjectData(
@JsonProperty("object_name")
val objectName: String
)
object UnknownFactorioObject : FactorioObjectData("unknown")
@JsonTypeName("LuaSurface")
class LuaSurfaceData(
@JsonProperty("name")
val name: String,
@JsonProperty("index")
val index: Int,
@JsonProperty("daytime")
val daytime: BigDecimal,
) : FactorioObjectData("LuaSurface")
/*
{
"tick": 1144860,
"object_name": "LuaSurface",
"event_type": "on_tick",
"data": {
"name": "nauvis",
"index": 1,
"daytime": 0.4942800000457983
}
}
*/

68
modules/factorio-events-processor/src/test/kotlin/dev/adamko/factorioevents/processor/model/FactorioServerLogRecordTest.kt
View file

@ -1,68 +0,0 @@
package dev.adamko.factorioevents.processor.model
import io.kotest.assertions.json.shouldEqualJson
import io.kotest.core.spec.style.FunSpec
import io.kotest.matchers.shouldBe
import io.kotest.matchers.shouldNotBe
import org.http4k.format.Jackson
import org.intellij.lang.annotations.Language
class FactorioServerLogRecordTest : FunSpec({
test("player obj") {
@Language("JSON")
val example = """
{
"mod_version":"0.0.3",
"tick":1256529,
"event_type":"on_player_joined_game",
"data":{
"object_name":"LuaPlayer",
"name":"fredthedeadhead",
"character_unit_number":1,
"associated_characters_unit_numbers":{
},
"position":{
"x":3.48828125,
"y":23.43359375
}
}
}
""".trimIndent()
val record: FactorioServerLogRecord<*> = Jackson.asA(example)
record shouldBe null
}
test("surface obj") {
@Language("JSON")
val example = """
{
"mod_version":"0.0.3",
"tick":1256529,
"event_type":"on_player_joined_game",
"data":{
"object_name":"LuaEntity",
"name":"character",
"type":"character",
"active":true,
"health":250,
"surface_index":1,
"unit_number":1,
"position":{
"x":3.48828125,
"y":23.43359375
}
}
}
""".trimIndent()
val record: FactorioServerLogRecord<LuaSurfaceData> = Jackson.asA(example)
record shouldNotBe null
val str = Jackson.asFormatString(record)
str.shouldEqualJson(example)
}
})