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

event-server: try sharing a job, so server shuts down at the same time

Browse source
This commit is contained in:
Adam 2022-05-01 15:45:23 +02:00
parent 153747e39c
commit fb265e39a6
2 changed files with 10 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

11
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt
View file

@ -15,6 +15,7 @@ import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancelChildren
import kotlinx.coroutines.job import kotlinx.coroutines.job
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import org.apache.kafka.streams.KafkaStreams import org.apache.kafka.streams.KafkaStreams
@ -30,7 +31,7 @@ internal class KafkatorioTopology(
) : CoroutineScope { ) : CoroutineScope {
override val coroutineContext: CoroutineContext = override val coroutineContext: CoroutineContext =
Dispatchers.Default + SupervisorJob() + CoroutineName("KafkatorioTopology") Dispatchers.Default + SupervisorJob(rootJob) + CoroutineName("KafkatorioTopology")
fun start() { fun start() {
splitPackets() splitPackets()
@ -76,7 +77,7 @@ internal class KafkatorioTopology(
private fun launchTopology( private fun launchTopology(
id: String, id: String,
topology: Topology, topology: Topology,
config: Map<String, String> = mapOf() config: Map<String, String> = mapOf(),
) { ) {
val props: Properties = appProps.kafkaConfig val props: Properties = appProps.kafkaConfig
@ -112,10 +113,16 @@ internal class KafkatorioTopology(
coroutineContext.job.invokeOnCompletion { coroutineContext.job.invokeOnCompletion {
println("closing $id KafkaStreams") println("closing $id KafkaStreams")
rootJob.cancelChildren()
rootJob.cancel()
streams.close(Duration.ofSeconds(1)) streams.close(Duration.ofSeconds(1))
} }
launch { streams.start() } launch { streams.start() }
} }
companion object {
val rootJob = SupervisorJob()
}
} }

2
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt
View file

@ -108,7 +108,7 @@ fun ColourHex.toRgbColor(): RGBColor {
private class ServerMapChunkHandler : CoroutineScope { private class ServerMapChunkHandler : CoroutineScope {
override val coroutineContext: CoroutineContext = override val coroutineContext: CoroutineContext =
Dispatchers.Default + SupervisorJob() + CoroutineName("ServerMapChunkHandler") Dispatchers.Default + SupervisorJob(rootJob) + CoroutineName("ServerMapChunkHandler")
/** The inbox for this handler. */ /** The inbox for this handler. */