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

move kafka topics creator into events-processors

Browse source
This commit is contained in:
Adam 2022-10-08 14:16:43 +02:00
parent 1c9a93b457
commit 6035c2eaa0
7 changed files with 32 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

10
modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/main.kt
View file

@ -1,10 +0,0 @@
package dev.adamko.kafkatorio.processor.admin
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
fun main() {
val appProps = ApplicationProperties.load()
val admin = KafkatorioKafkaAdmin(appProps)
admin.createKafkatorioTopics()
}

2
modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/topologyLauncher.kt
View file

@ -18,12 +18,12 @@ import org.apache.kafka.streams.Topology
import org.apache.kafka.streams.TopologyDescription
private val appProps = ApplicationProperties.load()
suspend fun launchTopology(
id: String,
topology: Topology,
additionalProperties: Map<String, String> = mapOf(),
appProps: ApplicationProperties,
): Unit = coroutineScope {
val props: Properties = appProps.kafkaStreamsConfig.toProperties()

11
modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/createTopics.kt Normal file
View file

@ -0,0 +1,11 @@
package dev.adamko.kafkatorio.processors
import dev.adamko.kafkatorio.processor.admin.KafkatorioKafkaAdmin
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
fun createTopics(
appProps: ApplicationProperties = ApplicationProperties.load(),
) {
val admin = KafkatorioKafkaAdmin(appProps)
admin.createKafkatorioTopics()
}

4
modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupEntitiesMapChunks.kt
View file

@ -1,5 +1,6 @@
package dev.adamko.kafkatorio.processors
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import dev.adamko.kafkatorio.processor.core.launchTopology
import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kafkatorio.schema.common.PrototypeHashCode
@ -22,6 +23,7 @@ suspend fun groupEntitiesMapChunks(
layer: ServerMapTileLayer,
colouredUpdatesStreamTopic: String,
colouredChunkStateTopic: String,
appProps: ApplicationProperties = ApplicationProperties.load(),
) {
val builder = StreamsBuilder()
@ -50,5 +52,5 @@ suspend fun groupEntitiesMapChunks(
colouredChunkStateTopic,
)
launchTopology("groupTilesMapChunks.${layer.dir}", topology)
launchTopology("groupTilesMapChunks.${layer.dir}", topology, appProps = appProps)
}

11
modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupTilesMapChunks.kt
View file

@ -1,5 +1,6 @@
package dev.adamko.kafkatorio.processors
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import dev.adamko.kafkatorio.processor.core.launchTopology
import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer
@ -11,7 +12,9 @@ import org.apache.kafka.streams.kstream.KStream
/** Terrain tiles */
suspend fun groupTilesMapChunks() {
suspend fun groupTilesMapChunks(
appProps: ApplicationProperties = ApplicationProperties.load(),
) {
val builder = StreamsBuilder()
val protosStream: KStream<FactorioServerId, PrototypesUpdate> =
@ -19,5 +22,9 @@ suspend fun groupTilesMapChunks() {
val topology = colourMapChunks(builder, protosStream)
launchTopology("groupTilesMapChunks.${ServerMapTileLayer.TERRAIN.dir}", topology)
launchTopology(
id = "groupTilesMapChunks.${ServerMapTileLayer.TERRAIN.dir}",
topology = topology,
appProps = appProps
)
}

3
modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/main.kt
View file

@ -10,6 +10,9 @@ import kotlinx.coroutines.launch
suspend fun main(): Unit = coroutineScope {
createTopics()
launch {
splitPackets()
}

7
modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/splitPackets.kt
View file

@ -1,17 +1,20 @@
package dev.adamko.kafkatorio.processors
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import dev.adamko.kafkatorio.processor.core.launchTopology
import dev.adamko.kafkatorio.server.processor.topology.factorioServerPacketStream
import dev.adamko.kafkatorio.server.processor.topology.splitFactorioServerPacketStream
import org.apache.kafka.streams.StreamsBuilder
suspend fun splitPackets() {
suspend fun splitPackets(
appProps: ApplicationProperties = ApplicationProperties.load(),
) {
val builder = StreamsBuilder()
val packets = factorioServerPacketStream(builder)
splitFactorioServerPacketStream(packets)
val topology = builder.build()
launchTopology("splitPackets", topology)
launchTopology("splitPackets", topology, appProps = appProps)
}