lazzexus/Kafkatorio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

factorio -> event -> mod -> console -> kafka pipe -> kafka -> event processor working!!!!!!!!!!!!!!!!!!

Browse source
This commit is contained in:
Adam 2021-12-03 09:04:18 +01:00
parent 7091f7648d
commit 42b60ac0c3
19 changed files with 152 additions and 87 deletions
Download patch file Download diff file Expand all files Collapse all files

43
modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt
View file

@ -5,13 +5,40 @@ import dev.adamko.kafkatorio.events.schema.FactorioObjectData
import dev.adamko.kafkatorio.events.schema.jsonMapper
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import java.time.Duration
import kotlinx.serialization.decodeFromString
import kotlinx.serialization.encodeToString
import kotlinx.serialization.modules.SerializersModule
import org.apache.kafka.common.serialization.Deserializer
import org.apache.kafka.common.serialization.Serde
import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.common.serialization.Serializer
import org.apache.kafka.streams.KafkaStreams
import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.TopologyDescription
import org.apache.kafka.streams.kstream.Consumed
import org.apache.kafka.streams.processor.TopicNameExtractor
import org.http4k.format.KotlinxSerialization
import org.apache.kafka.streams.kstream.Produced
import org.http4k.format.ConfigurableKotlinxSerialization
import org.http4k.format.asConfigurable
import org.http4k.format.withStandardMappings
object KSX : ConfigurableKotlinxSerialization({
ignoreUnknownKeys = true
prettyPrint = true
prettyPrintIndent = " "
asConfigurable().withStandardMappings().done()
serializersModule = SerializersModule { }
})
object JsonSerdes : Serde<FactorioEvent<FactorioObjectData>> {
override fun serializer() = Serializer<FactorioEvent<FactorioObjectData>> { _, message ->
jsonMapper.encodeToString<FactorioEvent<FactorioObjectData>>(message).encodeToByteArray()
}
override fun deserializer() = Deserializer<FactorioEvent<FactorioObjectData>> { _, bytes ->
jsonMapper.decodeFromString(bytes.decodeToString())
}
}
fun main() {
@ -25,16 +52,18 @@ fun main() {
)
.mapValues { readOnlyKey, value ->
println("Mapping $readOnlyKey:$value")
// jsonMapper.decodeFromString<FactorioEvent<*>>(value)
KotlinxSerialization.asA<FactorioEvent<FactorioObjectData>>(value)
jsonMapper.decodeFromString<FactorioEvent<FactorioObjectData>>(value)
// KSX.asA<FactorioEvent<FactorioObjectData>>(value)
}
.to(TopicNameExtractor { key, value, recordContext ->
value.data.objectName
})
.to(
{ key, value, recordContext -> value.data.objectName },
Produced.with(Serdes.String(), JsonSerdes)
)
val topology = builder.build()
val properties = appProps.kafkaConfig
val streams = KafkaStreams(topology, properties)
streams.setUncaughtExceptionHandler(StreamsExceptionHandler())