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

[events-processors] change where topo description is written

Browse source
This commit is contained in:
Adam 2022-10-22 19:52:02 +02:00
parent 657d5712f6
commit ce0d4b7378
2 changed files with 19 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -35,7 +35,7 @@ data class ApplicationProperties(
.addEnvironmentSource() .addEnvironmentSource()
.addResourceOrFileSource( .addResourceOrFileSource(
resourceOrFile = "${System.getProperty("user.home")}/.secret.kafkatorio-config.yml", resourceOrFile = "${System.getProperty("user.home")}/.secret.kafkatorio-config.yml",
optional = true optional = true,
) )
.addResourceOrFileSource("/.secret.config.yml", optional = true) .addResourceOrFileSource("/.secret.config.yml", optional = true)
.addResourceOrFileSource("/config.yml", optional = false) .addResourceOrFileSource("/config.yml", optional = false)

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

@ -4,7 +4,6 @@ import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import java.util.Properties import java.util.Properties
import kotlin.coroutines.resume import kotlin.coroutines.resume
import kotlin.coroutines.resumeWithException import kotlin.coroutines.resumeWithException
import kotlin.io.path.Path
import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.cancel import kotlinx.coroutines.cancel
import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.coroutineScope
@ -18,7 +17,6 @@ import org.apache.kafka.streams.Topology
import org.apache.kafka.streams.TopologyDescription import org.apache.kafka.streams.TopologyDescription
suspend fun launchTopology( suspend fun launchTopology(
id: String, id: String,
topology: Topology, topology: Topology,
@ -58,7 +56,7 @@ suspend fun launchTopology(
streams.close() streams.close()
} }
printTopologyDescription(appId, topology) printTopologyDescription(appId, topology, appProps)
streams.start() streams.start()
println("launched Topology $appId") println("launched Topology $appId")
} }
@ -68,21 +66,27 @@ suspend fun launchTopology(
} }
private fun printTopologyDescription(appId: String, topology: Topology) { private fun printTopologyDescription(
appId: String,
topology: Topology,
appProps: ApplicationProperties,
) {
val description: TopologyDescription = topology.describe() val description: TopologyDescription = topology.describe()
val descFile = Path("./build/$appId.txt").toFile().apply { val descFile = appProps.serverDataDir
parentFile.mkdirs() .resolve("kafka-streams/$appId.txt")
if (!exists()) createNewFile() .toFile().apply {
writeText(description.toString()) parentFile.mkdirs()
} if (!exists()) createNewFile()
writeText(description.toString())
}
println( println(
""" """
|---------------- |----------------
|$appId - ${descFile.canonicalPath} |$appId - ${descFile.canonicalPath}
|$description |$description
|---------------- |----------------
""".trimMargin() """.trimMargin()
) )
} }