From 4ca40afa090660cc701d2e18cd8d968e0fd53e7c Mon Sep 17 00:00:00 2001
From: Adam <897017+asemy@users.noreply.github.com>
Date: Fri, 29 Oct 2021 19:15:15 +0200
Subject: [PATCH] streams processor work. Stuck because of schema clashes. I
 want to get protoc working but it's a massive PIA.

---
 .../processor/StreamsExceptionHandler.kt      |  2 +-
 .../adamko/factorioevents/processor/main.kt   | 13 ++--
 .../src/main/resources/logback.xml            | 21 ++++++
 .../model/FactorioServerLogRecordTest.kt      | 68 +++++++++++++++++++
 4 files changed, 97 insertions(+), 7 deletions(-)
 create mode 100644 modules/factorio-events-processor/src/main/resources/logback.xml
 create mode 100644 modules/factorio-events-processor/src/test/kotlin/dev/adamko/factorioevents/processor/model/FactorioServerLogRecordTest.kt

diff --git a/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/StreamsExceptionHandler.kt b/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/StreamsExceptionHandler.kt
index bf05679..107e6e5 100644
--- a/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/StreamsExceptionHandler.kt
+++ b/modules/factorio-events-processor/src/main/kotlin/dev/adamko/factorioevents/processor/StreamsExceptionHandler.kt
@@ -6,6 +6,6 @@ import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThr
 class StreamsExceptionHandler : StreamsUncaughtExceptionHandler {
   override fun handle(exception: Throwable): StreamThreadExceptionResponse {
     println("${exception.message} - ${exception.stackTrace}")
-    return StreamThreadExceptionResponse.REPLACE_THREAD
+    return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION
   }
 }
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 3336a95..4932637 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
@@ -2,15 +2,12 @@ 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.TopologyDescription
 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
 
@@ -20,7 +17,6 @@ fun main() {
 
   val builder = StreamsBuilder()
 
-
   builder.stream(
     "factorio-server-log",
     Consumed.with(Serdes.String(), Serdes.String())
@@ -40,7 +36,12 @@ fun main() {
 
   streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
 
-  Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(10)) })
+  Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(1)) })
+
+//  streams.cleanUp()
+
+  val description: TopologyDescription = topology.describe()
+  println(description)
 
   streams.start()
 
diff --git a/modules/factorio-events-processor/src/main/resources/logback.xml b/modules/factorio-events-processor/src/main/resources/logback.xml
new file mode 100644
index 0000000..658a6ae
--- /dev/null
+++ b/modules/factorio-events-processor/src/main/resources/logback.xml
@@ -0,0 +1,21 @@
+<configuration>
+
+  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+    <withJansi>true</withJansi>
+    <!-- encoders are assigned the type
+         ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+    <encoder>
+      <pattern>%cyan(%d{HH:mm:ss.SSS}) %gray([%thread]) %highlight(%-5level) %magenta(%logger{36}) - %msg%n
+      </pattern>
+    </encoder>
+  </appender>
+
+  <logger name="org.apache.kafka" level="info" additivity="false">
+    <appender-ref ref="CONSOLE"/>
+  </logger>
+
+  <root level="info">
+    <appender-ref ref="STDOUT"/>
+  </root>
+
+</configuration>
\ No newline at end of file
diff --git a/modules/factorio-events-processor/src/test/kotlin/dev/adamko/factorioevents/processor/model/FactorioServerLogRecordTest.kt b/modules/factorio-events-processor/src/test/kotlin/dev/adamko/factorioevents/processor/model/FactorioServerLogRecordTest.kt
new file mode 100644
index 0000000..d672635
--- /dev/null
+++ b/modules/factorio-events-processor/src/test/kotlin/dev/adamko/factorioevents/processor/model/FactorioServerLogRecordTest.kt
@@ -0,0 +1,68 @@
+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)
+  }
+
+})