From ba03f761e0e67558e6e8c7a25d44c7a5c71d6335 Mon Sep 17 00:00:00 2001
From: Adam <897017+aSemy@users.noreply.github.com>
Date: Tue, 8 Feb 2022 17:56:34 +0100
Subject: [PATCH] zoom zoom faster - make 3 different Kafka consumers for
 player/tile/packet updates

---
 .../processor/KafkatorioTopology.kt           | 72 ++++++++++++++-----
 .../processor/config/ApplicationProperties.kt |  8 +--
 .../kafkatorio/processor/eventProcessor.kt    |  2 +-
 3 files changed, 60 insertions(+), 22 deletions(-)

diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt
index 06bf8ac..33d9a16 100644
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt
@@ -7,29 +7,38 @@ import dev.adamko.kafkatorio.processor.topology.saveMapTiles
 import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
 import dev.adamko.kafkatorio.processor.topology.tileProtoColourDictionary
 import java.time.Duration
+import kotlin.coroutines.CoroutineContext
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.job
+import kotlinx.coroutines.launch
 import org.apache.kafka.streams.KafkaStreams
 import org.apache.kafka.streams.StreamsBuilder
-import org.apache.kafka.streams.TopologyDescription
+import org.apache.kafka.streams.StreamsConfig
+import org.apache.kafka.streams.Topology
 
 class KafkatorioTopology(
   private val websocketServer: WebsocketServer,
-  private val builder: StreamsBuilder = StreamsBuilder(),
-) {
+) : CoroutineScope {
 
+  override val coroutineContext: CoroutineContext =
+    Dispatchers.Default + SupervisorJob() + CoroutineName("KafkatorioTopology")
 
   companion object {
     const val sourceTopic = "factorio-server-log"
   }
 
+  fun start() {
+    saveTiles()
+    splitPackets()
+    playerUpdates()
+  }
 
-  fun build() {
-
+  private fun saveTiles() {
+    val builder = StreamsBuilder()
     val packets = factorioServerPacketStream(builder)
-
-    splitFactorioServerPacketStream(packets)
-
-    playerUpdatesToWsServer(websocketServer, builder)
-
     val tileProtoColourDict = tileProtoColourDictionary(packets)
 
     val tilePrototypesTable = groupTilesIntoChunksWithColours(
@@ -40,17 +49,46 @@ class KafkatorioTopology(
 
     val topology = builder.build()
 
-    val streams = KafkaStreams(topology, appProps.kafkaConfig)
+    launchTopology("saveTiles", topology)
+  }
 
-//    streams.cleanUp()
+  private fun splitPackets() {
+
+    val builder = StreamsBuilder()
+    val packets = factorioServerPacketStream(builder)
+    splitFactorioServerPacketStream(packets)
+
+    val topology = builder.build()
+
+    launchTopology("splitPackets", topology)
+  }
+
+  private fun playerUpdates() {
+    val builder = StreamsBuilder()
+    playerUpdatesToWsServer(websocketServer, builder)
+
+    val topology = builder.build()
+
+    launchTopology("playerUpdates", topology)
+  }
+
+  private fun launchTopology(id: String, topology: Topology) {
+
+    val props =  appProps.kafkaConfig
+
+    val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String
+    props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)
+
+
+    val streams = KafkaStreams(topology, props)
     streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
-    Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(1)) })
 
-    val description: TopologyDescription = topology.describe()
-    println(description)
-
-    streams.start()
+    coroutineContext.job.invokeOnCompletion {
+      println("closing $id KafkaStreams")
+      streams.close(Duration.ofSeconds(1))
+    }
 
+    launch { streams.start() }
   }
 
 }
diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt
index f462adb..a7f523e 100644
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt
@@ -45,10 +45,10 @@ class ApplicationProperties(
   val kafkaConfig: Properties
     get() = EnvironmentKey
       .nonEmptyString()
-      .map {
-        it.reader().use { sr ->
-          Properties().also { p ->
-            p.load(sr)
+      .map { srcString ->
+        srcString.reader().use { stringReader ->
+          Properties().also { properties ->
+            properties.load(stringReader)
           }
         }
       }
diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt
index a8b487b..8e0c334 100644
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt
@@ -21,7 +21,7 @@ fun main(): Unit = runBlocking {
   val tileServer = WebMapTileServer()
 
   val topology = KafkatorioTopology(wsServer)
-  topology.build()
+  topology.start()
 
   val webServer = PolyHandler(
     tileServer.build(),