From 6035c2eaa038b5a2683dfbc3ef518de31ad9edc7 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sat, 8 Oct 2022 14:16:43 +0200 Subject: [PATCH] move kafka topics creator into events-processors --- .../dev/adamko/kafkatorio/processor/admin/main.kt | 10 ---------- .../kafkatorio/processor/core/topologyLauncher.kt | 2 +- .../dev/adamko/kafkatorio/processors/createTopics.kt | 11 +++++++++++ .../kafkatorio/processors/groupEntitiesMapChunks.kt | 4 +++- .../kafkatorio/processors/groupTilesMapChunks.kt | 11 +++++++++-- .../kotlin/dev/adamko/kafkatorio/processors/main.kt | 3 +++ .../dev/adamko/kafkatorio/processors/splitPackets.kt | 7 +++++-- 7 files changed, 32 insertions(+), 16 deletions(-) delete mode 100644 modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/main.kt create mode 100644 modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/createTopics.kt diff --git a/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/main.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/main.kt deleted file mode 100644 index 7819a59..0000000 --- a/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/main.kt +++ /dev/null @@ -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() -} diff --git a/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/topologyLauncher.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/topologyLauncher.kt index 03ff1cc..a597711 100644 --- a/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/topologyLauncher.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/topologyLauncher.kt @@ -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 = mapOf(), + appProps: ApplicationProperties, ): Unit = coroutineScope { val props: Properties = appProps.kafkaStreamsConfig.toProperties() diff --git a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/createTopics.kt b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/createTopics.kt new file mode 100644 index 0000000..535d2ef --- /dev/null +++ b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/createTopics.kt @@ -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() +} diff --git a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupEntitiesMapChunks.kt b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupEntitiesMapChunks.kt index 8783592..740b38f 100644 --- a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupEntitiesMapChunks.kt +++ b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupEntitiesMapChunks.kt @@ -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) } diff --git a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupTilesMapChunks.kt b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupTilesMapChunks.kt index fd65020..e95e026 100644 --- a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupTilesMapChunks.kt +++ b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupTilesMapChunks.kt @@ -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 = @@ -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 + ) } diff --git a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/main.kt b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/main.kt index 427903f..c046990 100644 --- a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/main.kt +++ b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/main.kt @@ -10,6 +10,9 @@ import kotlinx.coroutines.launch suspend fun main(): Unit = coroutineScope { + + createTopics() + launch { splitPackets() } diff --git a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/splitPackets.kt b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/splitPackets.kt index 9373bc2..7c4dd21 100644 --- a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/splitPackets.kt +++ b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/splitPackets.kt @@ -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) }