mucking about with the kafka processor

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())
 
-}
+  Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(10)) })
+
+  streams.start()
+
+}

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
+  }
+}
+
+ */