streams processor work. Stuck because of schema clashes. I want to get protoc working but it's a massive PIA.

This commit is contained in:
Adam 2021-10-29 19:15:15 +02:00
parent 106d7a7923
commit 4ca40afa09
4 changed files with 97 additions and 7 deletions

View file

@ -6,6 +6,6 @@ import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThr
class StreamsExceptionHandler : StreamsUncaughtExceptionHandler { class StreamsExceptionHandler : StreamsUncaughtExceptionHandler {
override fun handle(exception: Throwable): StreamThreadExceptionResponse { override fun handle(exception: Throwable): StreamThreadExceptionResponse {
println("${exception.message} - ${exception.stackTrace}") println("${exception.message} - ${exception.stackTrace}")
return StreamThreadExceptionResponse.REPLACE_THREAD return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION
} }
} }

View file

@ -2,15 +2,12 @@ package dev.adamko.factorioevents.processor
import dev.adamko.factorioevents.processor.config.ApplicationProperties import dev.adamko.factorioevents.processor.config.ApplicationProperties
import dev.adamko.factorioevents.processor.model.FactorioServerLogRecord import dev.adamko.factorioevents.processor.model.FactorioServerLogRecord
import dev.adamko.factorioevents.processor.model.LuaSurfaceData
import java.time.Duration import java.time.Duration
import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.streams.KafkaStreams import org.apache.kafka.streams.KafkaStreams
import org.apache.kafka.streams.KeyValue
import org.apache.kafka.streams.StreamsBuilder 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.Consumed
import org.apache.kafka.streams.kstream.Named
import org.apache.kafka.streams.processor.TopicNameExtractor import org.apache.kafka.streams.processor.TopicNameExtractor
import org.http4k.format.Jackson import org.http4k.format.Jackson
@ -20,7 +17,6 @@ fun main() {
val builder = StreamsBuilder() val builder = StreamsBuilder()
builder.stream( builder.stream(
"factorio-server-log", "factorio-server-log",
Consumed.with(Serdes.String(), Serdes.String()) Consumed.with(Serdes.String(), Serdes.String())
@ -40,7 +36,12 @@ fun main() {
streams.setUncaughtExceptionHandler(StreamsExceptionHandler()) 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() streams.start()

View file

@ -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>

View file

@ -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)
}
})