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 @@
+
+
+
+ true
+
+
+ %cyan(%d{HH:mm:ss.SSS}) %gray([%thread]) %highlight(%-5level) %magenta(%logger{36}) - %msg%n
+
+
+
+
+
+
+
+
+
+
+
+
+
\ 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 = Jackson.asA(example)
+
+ record shouldNotBe null
+
+ val str = Jackson.asFormatString(record)
+ str.shouldEqualJson(example)
+ }
+
+})