From fb265e39a655fdce4fde519249bf7a556a6033e8 Mon Sep 17 00:00:00 2001
From: Adam <897017+aSemy@users.noreply.github.com>
Date: Sun, 1 May 2022 15:45:23 +0200
Subject: [PATCH] event-server: try sharing a job, so server shuts down at the
 same time

---
 .../adamko/kafkatorio/processor/KafkatorioTopology.kt | 11 +++++++++--
 .../kafkatorio/processor/topology/flowMapTiles.kt     |  2 +-
 2 files changed, 10 insertions(+), 3 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 acb614b..d69b557 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
@@ -15,6 +15,7 @@ import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancelChildren
 import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
 import org.apache.kafka.streams.KafkaStreams
@@ -30,7 +31,7 @@ internal class KafkatorioTopology(
 ) : CoroutineScope {
 
   override val coroutineContext: CoroutineContext =
-    Dispatchers.Default + SupervisorJob() + CoroutineName("KafkatorioTopology")
+    Dispatchers.Default + SupervisorJob(rootJob) + CoroutineName("KafkatorioTopology")
 
   fun start() {
     splitPackets()
@@ -76,7 +77,7 @@ internal class KafkatorioTopology(
   private fun launchTopology(
     id: String,
     topology: Topology,
-    config: Map<String, String> = mapOf()
+    config: Map<String, String> = mapOf(),
   ) {
 
     val props: Properties = appProps.kafkaConfig
@@ -112,10 +113,16 @@ internal class KafkatorioTopology(
 
     coroutineContext.job.invokeOnCompletion {
       println("closing $id KafkaStreams")
+      rootJob.cancelChildren()
+      rootJob.cancel()
       streams.close(Duration.ofSeconds(1))
     }
 
     launch { streams.start() }
   }
 
+  companion object {
+    val rootJob = SupervisorJob()
+  }
+
 }
diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt
index ede9b1e..cd4b72e 100644
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt
@@ -108,7 +108,7 @@ fun ColourHex.toRgbColor(): RGBColor {
 private class ServerMapChunkHandler : CoroutineScope {
 
   override val coroutineContext: CoroutineContext =
-    Dispatchers.Default + SupervisorJob() + CoroutineName("ServerMapChunkHandler")
+    Dispatchers.Default + SupervisorJob(rootJob) + CoroutineName("ServerMapChunkHandler")
 
 
   /** The inbox for this handler. */