From da9cc6398fb0e93af7d458b1db6cc5f71536af98 Mon Sep 17 00:00:00 2001
From: Adam <897017+asemy@users.noreply.github.com>
Date: Mon, 25 Oct 2021 23:35:47 +0200
Subject: [PATCH] mucking about with the kafka processor

---
 .../build.gradle.kts                          |  4 +-
 .../adamko/factorioevents/processor/main.kt   | 32 +++++++++-
 .../model/FactorioServerLogRecord.kt          | 62 +++++++++++++++++++
 3 files changed, 95 insertions(+), 3 deletions(-)
 create mode 100644 modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/model/FactorioServerLogRecord.kt

diff --git a/modules/factorio-events-processor/build.gradle.kts b/modules/factorio-events-processor/build.gradle.kts
index 9a44560..9229e4d 100644
--- a/modules/factorio-events-processor/build.gradle.kts
+++ b/modules/factorio-events-processor/build.gradle.kts
@@ -1,6 +1,6 @@
-
 plugins {
   id("dev.adamko.factoriowebmap.archetype.kotlin-jvm")
+  kotlin("plugin.serialization")
   application
 }
 
@@ -19,6 +19,8 @@ dependencies {
 
   implementation("org.apache.kafka:kafka-streams:3.0.0")
 
+  implementation(libs.kotlinx.serialization)
+
   implementation(platform(libs.http4k.bom))
   implementation(libs.http4k.core)
   implementation(libs.http4k.cloudnative)
diff --git a/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/main.kt b/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/main.kt
index f0df841..3336a95 100644
--- a/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/main.kt
+++ b/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/main.kt
@@ -1,19 +1,47 @@
 package dev.adamko.factorioevents.processor
 
 import dev.adamko.factorioevents.processor.config.ApplicationProperties
+import dev.adamko.factorioevents.processor.model.FactorioServerLogRecord
+import dev.adamko.factorioevents.processor.model.LuaSurfaceData
+import java.time.Duration
+import org.apache.kafka.common.serialization.Serdes
 import org.apache.kafka.streams.KafkaStreams
+import org.apache.kafka.streams.KeyValue
 import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.kstream.Branched
+import org.apache.kafka.streams.kstream.Consumed
+import org.apache.kafka.streams.kstream.Named
+import org.apache.kafka.streams.processor.TopicNameExtractor
+import org.http4k.format.Jackson
 
 fun main() {
 
   val appProps = ApplicationProperties()
 
-  val topology = StreamsBuilder().build()
+  val builder = StreamsBuilder()
 
+
+  builder.stream(
+    "factorio-server-log",
+    Consumed.with(Serdes.String(), Serdes.String())
+  )
+    .mapValues { readOnlyKey, value ->
+      println("Mapping $readOnlyKey:$value")
+      Jackson.asA<FactorioServerLogRecord<*>>(value)
+    }
+    .to(TopicNameExtractor { key, value, recordContext ->
+      value.data.objectName
+    })
+
+  val topology = builder.build()
   val properties = appProps.kafkaConfig
 
   val streams = KafkaStreams(topology, properties)
 
   streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
 
-}
\ No newline at end of file
+  Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(10)) })
+
+  streams.start()
+
+}
diff --git a/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/model/FactorioServerLogRecord.kt b/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/model/FactorioServerLogRecord.kt
new file mode 100644
index 0000000..c8a8182
--- /dev/null
+++ b/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/model/FactorioServerLogRecord.kt
@@ -0,0 +1,62 @@
+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.PROPERTY,
+  property = "object_name",
+  defaultImpl = Void::class, // equivalent to null
+  visible = true,
+)
+sealed class FactorioObjectData(
+  @JsonProperty("object_name")
+  val objectName: String
+)
+
+@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
+  }
+}
+
+ */
\ No newline at end of file