From 0c6f2a4bfee44db855373de2371b8731badc0be5 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sun, 19 Jun 2022 10:23:11 +0200 Subject: [PATCH] refactor websocket/tile server --- modules/events-server/.gitignore | 1 + .../processor/KafkatorioTopology.kt | 103 +++++++----- .../kafkatorio/processor/WebsocketServer.kt | 93 ----------- .../processor/admin/KafkatorioKafkaAdmin.kt | 17 +- .../processor/config/ApplicationProperties.kt | 144 ++++++++++------- .../kafkatorio/processor/config/Lenses.kt | 73 ++++----- .../processor/config/environmentExt.kt | 148 +++++++++--------- .../kafkatorio/processor/eventProcessor.kt | 62 +++++--- .../processor/misc/FileEventsFlow.kt | 2 - .../processor/tileserver/WebMapTileServer.kt | 100 +++--------- .../processor/tileserver/WebServer.kt | 16 ++ .../processor/tileserver/WebsocketServer.kt | 148 ++++++++++++++++++ .../topology/factorioServerPackets.kt | 5 +- .../processor/topology/playerUpdates.kt | 57 +++++-- .../src/main/resources/.gitignore | 1 - 15 files changed, 556 insertions(+), 414 deletions(-) delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebServer.kt create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebsocketServer.kt delete mode 100644 modules/events-server/src/main/resources/.gitignore diff --git a/modules/events-server/.gitignore b/modules/events-server/.gitignore index e7abcfb..fe98be7 100644 --- a/modules/events-server/.gitignore +++ b/modules/events-server/.gitignore @@ -1 +1,2 @@ kafka-state/ +kafkatorio-tiles/ 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 179863e..8da6383 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 @@ -1,23 +1,26 @@ package dev.adamko.kafkatorio.processor import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer +import dev.adamko.kafkatorio.processor.topology.broadcastToWebsocket import dev.adamko.kafkatorio.processor.topology.colourMapChunks import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream -import dev.adamko.kafkatorio.processor.topology.broadcastToWebsocket import dev.adamko.kafkatorio.processor.topology.saveMapTiles import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream -import java.time.Duration import java.util.Properties -import kotlin.coroutines.CoroutineContext -import kotlin.coroutines.cancellation.CancellationException +import kotlin.coroutines.resume +import kotlin.coroutines.resumeWithException import kotlin.io.path.Path +import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.Duration.Companion.seconds import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.cancelChildren +import kotlinx.coroutines.cancel +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.job import kotlinx.coroutines.launch +import kotlinx.coroutines.suspendCancellableCoroutine +import org.apache.kafka.clients.producer.ProducerConfig import org.apache.kafka.streams.KafkaStreams import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.StreamsConfig @@ -27,13 +30,12 @@ import org.apache.kafka.streams.TopologyDescription internal class KafkatorioTopology( private val websocketServer: WebsocketServer, - private val appProps: ApplicationProperties = ApplicationProperties(), -) : CoroutineScope { + private val appProps: ApplicationProperties, +) { - override val coroutineContext: CoroutineContext = - Dispatchers.Default + rootJob + CoroutineName("KafkatorioTopology") + private val builderScope = CoroutineScope(CoroutineName("KafkaTopology")) - fun start() { + suspend fun start() = builderScope.launch { splitPackets() groupTilesMapChunks() @@ -42,7 +44,7 @@ internal class KafkatorioTopology( websocketBroadcaster() } - private fun splitPackets() { + private suspend fun splitPackets() { val builder = StreamsBuilder() val packets = factorioServerPacketStream(builder) splitFactorioServerPacketStream(packets) @@ -52,45 +54,77 @@ internal class KafkatorioTopology( launchTopology("splitPackets", topology) } - private fun groupTilesMapChunks() { + private suspend fun groupTilesMapChunks() { val builder = StreamsBuilder() val topology = colourMapChunks(builder) launchTopology("groupTilesMapChunks", topology) } - private fun saveTiles() { + private suspend fun saveTiles() { val builder = StreamsBuilder() - val topology = saveMapTiles(builder, appProps.webmapTileDir) + val topology = saveMapTiles(builder, appProps.tileDir) launchTopology("saveTiles", topology) } - private fun websocketBroadcaster() { + private suspend fun websocketBroadcaster() { val builder = StreamsBuilder() broadcastToWebsocket(websocketServer, builder) val topology = builder.build() - launchTopology("websocketBroadcaster", topology) + launchTopology( + "websocketBroadcaster", topology, mapOf( + StreamsConfig.COMMIT_INTERVAL_MS_CONFIG to "${1.seconds.inWholeMilliseconds}", + StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG) to "${10.milliseconds.inWholeMilliseconds}" + ) + ) } - private fun launchTopology( + private suspend fun launchTopology( id: String, topology: Topology, - config: Map = mapOf(), + additionalProperties: Map = mapOf(), ) { - val props: Properties = appProps.kafkaConfig + val props: Properties = appProps.kafkaStreamsConfig.toProperties() - config.forEach { (k, v) -> props.setProperty(k, v) } + additionalProperties.forEach { (k, v) -> props.setProperty(k, v) } - val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String + 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 { - coroutineContext.job.cancel(CancellationException(cause = it)) - }) + builderScope.launch(CoroutineName(appId)) { + val streamsState: KafkaStreams.State = suspendCancellableCoroutine { cont -> + + val streams = KafkaStreams(topology, props) + streams.setUncaughtExceptionHandler(StreamsExceptionHandler { e -> + cont.resumeWithException(e) + }) + + streams.setStateListener { newState: KafkaStreams.State, _ -> + when (newState) { + KafkaStreams.State.NOT_RUNNING, + KafkaStreams.State.ERROR -> cont.resume(newState) + else -> Unit // do nothing + } + } + + cont.invokeOnCancellation { + println("[$appId] coroutine cancelled -> closing. cause:$it") + streams.close() + } + + printTopologyDescription(appId, topology) + streams.start() + println("launched Topology $appId") + } + println("exiting Topology $appId: $streamsState") + currentCoroutineContext().job.cancel("$appId: $streamsState") + } + } + + private fun printTopologyDescription(appId: String, topology: Topology) { val description: TopologyDescription = topology.describe() val descFile = Path("./build/$appId.txt").toFile().apply { @@ -107,19 +141,12 @@ internal class KafkatorioTopology( |---------------- """.trimMargin() ) - - coroutineContext.job.invokeOnCompletion { - println("closing $id KafkaStreams") - rootJob.cancelChildren() - rootJob.cancel() - streams.close(Duration.ofSeconds(1)) - } - - launch { streams.start() } } companion object { - val rootJob = SupervisorJob() + private fun Map.toProperties(): Properties = + entries.fold(Properties()) { props, (k, v) -> + props.apply { setProperty(k, v) } + } } - } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt deleted file mode 100644 index aa03266..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt +++ /dev/null @@ -1,93 +0,0 @@ -package dev.adamko.kafkatorio.processor - -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.channels.BufferOverflow -import kotlinx.coroutines.flow.MutableSharedFlow -import kotlinx.coroutines.flow.launchIn -import kotlinx.coroutines.flow.onEach -import kotlinx.coroutines.flow.withIndex -import kotlinx.coroutines.job -import kotlinx.coroutines.runBlocking -import kotlinx.coroutines.supervisorScope -import org.http4k.routing.bind -import org.http4k.routing.websockets -import org.http4k.websocket.Websocket -import org.http4k.websocket.WsHandler -import org.http4k.websocket.WsMessage -import org.http4k.websocket.WsStatus - - -class WebsocketServer { - - // create a scope+job for the server - private val serverJob = KafkatorioTopology.rootJob - private val serverScope = CoroutineScope(serverJob) - - private val clients: MutableSharedFlow = MutableSharedFlow( - replay = 100, - extraBufferCapacity = 100, - BufferOverflow.DROP_OLDEST, - ) - - private val messages: MutableSharedFlow = MutableSharedFlow( - replay = 100, - extraBufferCapacity = 100, - BufferOverflow.DROP_OLDEST, - ) - - private fun bindClient(ws: Websocket) = runBlocking(serverScope.coroutineContext) { - clients.emit(ws) - } - - fun sendMessage(msg: String) = runBlocking(serverScope.coroutineContext) { - messages.emit(msg) - } - - suspend fun build(): WsHandler = supervisorScope builder@{ - - coroutineContext.job.invokeOnCompletion { e -> - e?.let { serverJob.completeExceptionally(it) } ?: serverJob.complete() - } - - clients - .withIndex() - .onEach { (id, ws) -> - // create a scope+job for each client - val clientJob = SupervisorJob(serverJob) - val clientScope = CoroutineScope(clientJob) - - clientJob.invokeOnCompletion { e -> - println("[clientJob.invokeOnCompletion] closing client ws $e") - when (e) { - null -> ws.close(WsStatus.NORMAL) - else -> ws.close(WsStatus.ABNORMAL_CLOSE) - } - } - - ws.onError { e -> - sendMessage("[ws.onError] client $id error [${e}]") - clientJob.completeExceptionally(e) - } - ws.onClose { status -> - sendMessage("[ws.onClose] client $id disconnected [${status}]") - clientJob.complete() - } - ws.onMessage { wsMsg -> - sendMessage("$id says: ${wsMsg.bodyString()}") - } - - sendMessage("new client $id!") - - messages - .onEach { msg -> ws.send(WsMessage(msg)) } - .launchIn(clientScope) - - } - .launchIn(serverScope) - - - return@builder websockets("/ws" bind ::bindClient) - - } -} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/KafkatorioKafkaAdmin.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/KafkatorioKafkaAdmin.kt index b0fbb20..02471bb 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/KafkatorioKafkaAdmin.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/KafkatorioKafkaAdmin.kt @@ -1,5 +1,6 @@ package dev.adamko.kafkatorio.processor.admin +import dev.adamko.kafkatorio.processor.config.ApplicationProperties import java.util.concurrent.TimeUnit import mu.KotlinLogging import org.apache.kafka.clients.admin.Admin @@ -8,17 +9,17 @@ import org.apache.kafka.clients.admin.NewTopic import org.apache.kafka.clients.admin.TopicListing -object KafkatorioKafkaAdmin { +class KafkatorioKafkaAdmin( + appProps: ApplicationProperties +) { private val logger = KotlinLogging.logger {} - private val kafkaAdmin: Admin = - Admin.create( - mapOf( - "bootstrap.servers" to "http://localhost:9092", - "client.id" to "kafkatorio.setup-admin", - ) - ) + init { + println("Creating KafkatorioKafkaAdmin with config: ${appProps.kafkaAdminConfig}") + } + + private val kafkaAdmin: Admin = Admin.create(appProps.kafkaAdminConfig) fun createKafkatorioTopics() { 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 e9ec6e3..6881ecb 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 @@ -1,64 +1,94 @@ package dev.adamko.kafkatorio.processor.config -import dev.adamko.kafkatorio.processor.config.PortVal.Companion.portVal +import com.sksamuel.hoplite.ConfigLoaderBuilder +import com.sksamuel.hoplite.addEnvironmentSource +import com.sksamuel.hoplite.addResourceOrFileSource import java.nio.file.Path -import java.time.Clock -import java.util.Properties -import kotlin.io.path.Path -import org.http4k.cloudnative.env.Environment -import org.http4k.cloudnative.env.EnvironmentKey -import org.http4k.lens.nonEmptyString -/** - * Load properties, in descending priority order, from - * - * 1. Environment variables - * 2. This app's env file (as defined by the environment variable [Lenses.APPLICATION_PROPERTIES_FILE_PATH]) - * # 3. A resources-dir yaml file - * - */ -internal class ApplicationProperties( - vararg environments: Environment = arrayOf( - Environment.env(), - Environment.fromEnvVarPath(), - Environment.fromResourcesFile(".secret.config.yml") - ) + +// loaded by hoplite +data class ApplicationProperties( + val webPort: PortVal, + val tileDir: Path, + val kafkaAdminConfig: Map, + val kafkaStreamsConfig: Map, ) { - - private val appProps: Environment = - environments.reduce { acc, environment -> acc.overrides(environment) } - - - fun clock(): Clock = Clock.systemUTC() - - - /** - * The port of the server - * - * Set by GCP ([docs](https://cloud.google.com/run/docs/reference/container-contract#env-vars)). - */ - val serverPort: PortVal - get() = EnvironmentKey.portVal().defaulted("PORT", PortVal.DEFAULT)[appProps] - - - val kafkaConfig: Properties - get() = EnvironmentKey - .nonEmptyString() - .map { srcString -> - srcString.reader().use { stringReader -> - Properties().also { properties -> - properties.load(stringReader) - } - } - } - .required("kafka.config")[appProps] - - - val webmapTileDir: Path - get() = EnvironmentKey - .nonEmptyString() - .map(::Path) - .required("tileServer.webmapTileDir")[appProps] - + companion object { + /** use Hoplite to load config from environment variables or properties files */ + fun load(): ApplicationProperties = + ConfigLoaderBuilder.default() + .addDefaults() + .allowUnresolvedSubstitutions() + .allowEmptyTree() + .addEnvironmentSource() + .addResourceOrFileSource("/.secret.config.yml", optional = true) + .addResourceOrFileSource("/config.yml", optional = false) + .build() + .loadConfigOrThrow() + } } + + +//import dev.adamko.kafkatorio.processor.config.PortVal.Companion.portVal +//import java.nio.file.Path +//import java.time.Clock +//import java.util.Properties +//import kotlin.io.path.Path +//import org.http4k.cloudnative.env.Environment +//import org.http4k.cloudnative.env.EnvironmentKey +//import org.http4k.lens.nonEmptyString +// +///** +// * Load properties, in descending priority order, from +// * +// * 1. Environment variables +// * 2. This app's env file (as defined by the environment variable [Lenses.APPLICATION_PROPERTIES_FILE_PATH]) +// * # 3. A resources-dir yaml file +// * +// */ +//internal class ApplicationProperties( +// vararg environments: Environment = arrayOf( +// Environment.env(), +// Environment.fromEnvVarPath(), +// Environment.fromResourcesFile(".secret.config.yml") +// ) +//) { +// +// +// private val appProps: Environment = +// environments.reduce { acc, environment -> acc.overrides(environment) } +// +// +// fun clock(): Clock = Clock.systemUTC() +// +// +// /** +// * The port of the server +// * +// * Set by GCP ([docs](https://cloud.google.com/run/docs/reference/container-contract#env-vars)). +// */ +// val serverPort: PortVal +// get() = EnvironmentKey.portVal().defaulted("PORT", PortVal.DEFAULT)[appProps] +// +// +// val kafkaConfig: Properties +// get() = EnvironmentKey +// .nonEmptyString() +// .map { srcString -> +// srcString.reader().use { stringReader -> +// Properties().also { properties -> +// properties.load(stringReader) +// } +// } +// } +// .required("kafka.config")[appProps] +// +// +// val webmapTileDir: Path +// get() = EnvironmentKey +// .nonEmptyString() +// .map(::Path) +// .required("tileServer.webmapTileDir")[appProps] +// +//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/Lenses.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/Lenses.kt index fec46a8..7faeb9c 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/Lenses.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/Lenses.kt @@ -1,38 +1,39 @@ package dev.adamko.kafkatorio.processor.config -import java.io.File -import org.http4k.cloudnative.env.Environment -import org.http4k.cloudnative.env.EnvironmentKey -import org.http4k.core.Credentials -import org.http4k.lens.BiDiLens -import org.http4k.lens.BiDiMapping - - -internal object Lenses { - - /** The location of this app's env file. */ - val APPLICATION_PROPERTIES_FILE_PATH: BiDiLens = - EnvironmentKey - .map( - { path: String -> File(path) }, - { file: File -> file.canonicalPath }, - ) - .optional("APPLICATION_PROPERTIES_FILE_PATH") - - /** Bidirectional [String] and [Credentials] mapper. */ - val credentialsLens: BiDiMapping = run { - val separator = ":" - BiDiMapping( - // from String to Credentials - { property -> - property.split(separator, limit = 2) - .takeIf { it.size == 2 } - ?.let { (user, pass) -> Credentials(user, pass) } - ?: throw IllegalArgumentException("Credentials are invalid") - }, - // from Credentials to String - { it.user + separator + it.password } - ) - } - -} +//import java.io.File +//import org.http4k.cloudnative.env.Environment +//import org.http4k.cloudnative.env.EnvironmentKey +//import org.http4k.core.Credentials +//import org.http4k.lens.BiDiLens +//import org.http4k.lens.BiDiMapping +// +// +//internal object Lenses { +// +// /** The location of this app's env file. */ +// val APPLICATION_PROPERTIES_FILE_PATH: BiDiLens = +// EnvironmentKey +// .map( +// { path: String -> File(path) }, +// { file: File -> file.canonicalPath }, +// ) +// .optional("APPLICATION_PROPERTIES_FILE_PATH") +// +// +// /** Bidirectional [String] and [Credentials] mapper. */ +// val credentialsLens: BiDiMapping = run { +// val separator = ":" +// BiDiMapping( +// // from String to Credentials +// { property -> +// property.split(separator, limit = 2) +// .takeIf { it.size == 2 } +// ?.let { (user, pass) -> Credentials(user, pass) } +// ?: throw IllegalArgumentException("Credentials are invalid") +// }, +// // from Credentials to String +// { it.user + separator + it.password } +// ) +// } +// +//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/environmentExt.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/environmentExt.kt index 4bfeec7..3ff3eee 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/environmentExt.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/environmentExt.kt @@ -1,74 +1,78 @@ package dev.adamko.kafkatorio.processor.config -import java.io.File -import java.io.FileNotFoundException -import org.http4k.cloudnative.env.Environment -import org.http4k.cloudnative.env.MapEnvironment -import org.http4k.cloudnative.env.fromYaml -import org.http4k.lens.Lens - - -/** - * Load properties from either a `.yaml` or `.env` file. - * - * (An improved version of [Environment.fromResource]) - */ -fun Environment.Companion.fromResourcesFile( - resource: String, - default: Environment = EMPTY -): Environment { - return try { - val file = - this::class.java - .getResource("/${resource.removePrefix("/")}") - ?.file - ?.let { File(it) } - fromFile(file, default) - } catch (e: FileNotFoundException) { - default - } -} - -/** - * Try fetching a properties file the location provided by the - * [`JVM`][jvmProperties] or [`System`][env] environment variable - * [Lenses.APPLICATION_PROPERTIES_FILE_PATH], if it exists. - * - * If it does not exist, return [default]. - */ -fun Environment.Companion.fromEnvVarPath( - location: Lens = Lenses.APPLICATION_PROPERTIES_FILE_PATH, - default: Environment = EMPTY, -): Environment { - val file = (jvmProperties() overrides env())[location] - - return fromFile(file, default) -} - -/** - * Load properties from either a `.yaml` or `.env` file. - * - * (An improved version of [Environment.from]) - */ -internal fun Environment.Companion.fromFile( - file: File?, - default: Environment = EMPTY, -): Environment { - return file - ?.takeIf { it.isFile } - .let { - when (it?.extension) { - "yml", "yaml" -> fromYaml(it) - "env" -> from(it) - else -> default - } - } -} - -/** The same as [Environment.ENV], but refreshes on each invocation (which is useful for testing). */ -fun Environment.Companion.env(): Environment = - MapEnvironment.from(System.getenv().toProperties()) - -/** The same as [Environment.JVM_PROPERTIES], but refreshes on each invocation (which is useful for testing). */ -internal fun Environment.Companion.jvmProperties(): Environment = - MapEnvironment.from(System.getProperties()) +//import java.io.File +//import java.io.FileNotFoundException +//import org.http4k.cloudnative.env.Environment +//import org.http4k.cloudnative.env.MapEnvironment +//import org.http4k.cloudnative.env.fromYaml +//import org.http4k.lens.Lens +// +// +///** +// * Load properties from either a `.yaml` or `.env` file. +// * +// * (An improved version of [Environment.fromResource]) +// */ +//fun Environment.Companion.fromResourcesFile( +// resource: String, +// default: Environment = EMPTY +//): Environment { +// return try { +// val file = +// this::class.java +// .getResource("/${resource.removePrefix("/")}") +// ?.file +// ?.let { File(it) } +// fromFile(file, default) +// } catch (e: FileNotFoundException) { +// default +// } +//} +// +// +///** +// * Try fetching a properties file the location provided by the +// * [`JVM`][jvmProperties] or [`System`][env] environment variable +// * [Lenses.APPLICATION_PROPERTIES_FILE_PATH], if it exists. +// * +// * If it does not exist, return [default]. +// */ +//fun Environment.Companion.fromEnvVarPath( +// location: Lens = Lenses.APPLICATION_PROPERTIES_FILE_PATH, +// default: Environment = EMPTY, +//): Environment { +// val file = (jvmProperties() overrides env())[location] +// +// return fromFile(file, default) +//} +// +// +///** +// * Load properties from either a `.yaml` or `.env` file. +// * +// * (An improved version of [Environment.from]) +// */ +//internal fun Environment.Companion.fromFile( +// file: File?, +// default: Environment = EMPTY, +//): Environment { +// return file +// ?.takeIf { it.isFile } +// .let { +// when (it?.extension) { +// "yml", "yaml" -> fromYaml(it) +// "env" -> from(it) +// else -> default +// } +// } +//} +// +// +///** The same as [Environment.ENV], but refreshes on each invocation (which is useful for testing). */ +//fun Environment.Companion.env(): Environment = +// MapEnvironment.from(System.getenv().toProperties()) +// +// +///** The same as [Environment.JVM_PROPERTIES], but refreshes on each invocation (which is useful for testing). */ +//internal fun Environment.Companion.jvmProperties(): Environment = +// MapEnvironment.from(System.getProperties()) 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 8d9c6e2..3eceec0 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 @@ -2,33 +2,57 @@ package dev.adamko.kafkatorio.processor import dev.adamko.kafkatorio.processor.admin.KafkatorioKafkaAdmin +import dev.adamko.kafkatorio.processor.config.ApplicationProperties import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer -import kotlinx.coroutines.runBlocking -import org.http4k.server.PolyHandler -import org.http4k.server.Undertow -import org.http4k.server.asServer +import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer +import dev.adamko.kafkatorio.processor.tileserver.webServer +import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.job +import kotlinx.coroutines.launch +import kotlinx.serialization.Serializable +import org.http4k.server.Http4kServer -fun main(): Unit = runBlocking { +suspend fun main() { - KafkatorioKafkaAdmin.createKafkatorioTopics() + val appProps = ApplicationProperties.load() + + val admin = KafkatorioKafkaAdmin(appProps) + admin.createKafkatorioTopics() val wsServer = WebsocketServer() - val tileServer = WebMapTileServer() + val tileServer = WebMapTileServer(appProps) - val topology = KafkatorioTopology(wsServer) - topology.start() + coroutineScope { + launch { + val topology = KafkatorioTopology(wsServer, appProps) + topology.start() - val webServer = PolyHandler( - tileServer.build(), - wsServer.build(), - ).asServer(Undertow(9073)) + println("launched KafkatorioTopology") - Runtime.getRuntime().addShutdownHook(Thread { - webServer.stop() - }) -// launch { - webServer.start().block() -// } + awaitCancellation() + } + launch { + val webServer: Http4kServer = webServer(tileServer, wsServer, appProps) + + currentCoroutineContext().job.invokeOnCompletion { + println("webserver stopping. Cause: $it") + webServer.stop() + } + + webServer.start() + println("launched webserver port:${webServer.port()}") + + awaitCancellation() + } + } } +// Runtime.getRuntime().addShutdownHook(Thread { +// webServer.stop() +// }) +// launch { +// } +//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/FileEventsFlow.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/FileEventsFlow.kt index 6b467d3..05b79c1 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/FileEventsFlow.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/FileEventsFlow.kt @@ -1,6 +1,5 @@ package dev.adamko.kafkatorio.processor.misc -import dev.adamko.kafkatorio.processor.KafkatorioTopology import java.io.File import java.nio.file.FileSystems import java.nio.file.Files @@ -56,7 +55,6 @@ class FileEventsFlow private constructor( override val coroutineContext: CoroutineContext = CoroutineName("FileEventsFlow") + - KafkatorioTopology.rootJob + Dispatchers.Unconfined + CoroutineExceptionHandler { _, e -> println("handling exception in FileEventsFlow") diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebMapTileServer.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebMapTileServer.kt index 0519684..a599cef 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebMapTileServer.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebMapTileServer.kt @@ -18,18 +18,17 @@ import org.http4k.routing.routes import org.http4k.routing.static internal class WebMapTileServer( - appProps: ApplicationProperties = ApplicationProperties() + appProps: ApplicationProperties ) { - private val tilesLoader = ResourceLoader.Directory(appProps.webmapTileDir.absolutePathString()) + private val tilesLoader = ResourceLoader.Directory(appProps.tileDir.toFile().canonicalPath) private val filters = listOf( - ETagChecker()(), + ETagChecker(), ).reduce(Filter::then) private val routes: RoutingHttpHandler = routes( - "/tiles" bind GET to - static(tilesLoader).withFilter(filters) + "/tiles" bind GET to static(tilesLoader).withFilter(filters) ) fun build(): HttpHandler { @@ -37,87 +36,40 @@ internal class WebMapTileServer( } - inner class ETagChecker -// : CoroutineScope - { + private class ETagChecker : Filter { -// private val tagState: MutableStateFlow> = -// MutableStateFlow(mutableMapOf()) -// -// override val coroutineContext: CoroutineContext = -// CoroutineName("ETagChecker") + -// Dispatchers.Unconfined + -// KafkatorioTopology.rootJob + override fun invoke(next: HttpHandler): HttpHandler = { request: Request -> + val requestedETag = request.header("If-None-Match") -// init { -// launch { -// appProps.webmapTileDir -// .fileEventsFlow() -// .filterIsInstance() -// .filter { it.path.isRegularFile() } -//// .filter { it.path.extension == "png" } -// .runningFold(mutableMapOf()) { tags, event -> -//// println("[WebMapTileServer] handling file update ${event.path}") -// -// val relativePath = event.path -// .relativeTo(appProps.webmapTileDir) -// .invariantSeparatorsPathString -// -// when (event) { -// is KWatchEvent.Deleted -> tags.remove(relativePath) -// is KWatchEvent.Created, -// is KWatchEvent.Modified -> { -// val md5 = event.path.md5() -// if (md5 != null) { -// tags[relativePath] = md5 -// } else { -// println("null md5 for ${event.path}") -// } -// } -// } -// -// tags -// }.cancellable() -// .collect(tagState) -// } -// } + var response = next(request) + val responseBytes = response.body.stream.use { it.readAllBytes() } + val currentETag = responseBytes.md5() + response = response.body(MemoryBody(responseBytes)) - private val Request.etagKey: String get() = uri.path.removePrefix("/tiles/") - - - operator fun invoke() = Filter { next -> - { request -> - val requestedETag = request.header("If-None-Match") - - var response = next(request) - val responseBytes = response.body.stream.use { it.readAllBytes() } - val currentETag = responseBytes.md5() - - response = response.body(MemoryBody(responseBytes)) - - when { - currentETag == null -> { + when { + currentETag == null -> { // println("[${request.etagKey}] currentETag is null (requested: $requestedETag), executing request") - response - .header("Cache-Control", "no-cache, max-age=0, must-revalidate") - } - requestedETag == currentETag -> { + response + .header("Cache-Control", "no-cache, max-age=0, must-revalidate") + } + requestedETag == currentETag -> { // println("[${request.etagKey}] currentETag == requested, returning 304 $requestedETag") - Response(Status.NOT_MODIFIED) - .header("Cache-Control", "no-cache, max-age=0, must-revalidate") - } - else -> { + Response(Status.NOT_MODIFIED) + .header("Cache-Control", "no-cache, max-age=0, must-revalidate") + } + else -> { // println("[${request.etagKey}] currentETag != requested, executing request and adding currentETag:$currentETag (requested:$requestedETag)") - response - .header("ETag", currentETag) + response + .header("ETag", currentETag) // .header("Cache-Control", "no-store") - .header("Cache-Control", "no-cache, max-age=0, must-revalidate") - } + .header("Cache-Control", "no-cache, max-age=0, must-revalidate") } } } + } + companion object { private fun ByteArray.md5(): String? { return if (isEmpty()) { null diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebServer.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebServer.kt new file mode 100644 index 0000000..e3f5f4c --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebServer.kt @@ -0,0 +1,16 @@ +package dev.adamko.kafkatorio.processor.tileserver + +import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import org.http4k.server.Http4kServer +import org.http4k.server.PolyHandler +import org.http4k.server.Undertow +import org.http4k.server.asServer + +internal suspend fun webServer( + tileServer: WebMapTileServer, + wsServer: WebsocketServer, + appProps: ApplicationProperties, +): Http4kServer = PolyHandler( + tileServer.build(), + wsServer.build(), +).asServer(Undertow(appProps.webPort())) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebsocketServer.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebsocketServer.kt new file mode 100644 index 0000000..47ca3de --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebsocketServer.kt @@ -0,0 +1,148 @@ +package dev.adamko.kafkatorio.processor.tileserver + +import kotlin.coroutines.resume +import kotlin.coroutines.resumeWithException +import kotlin.time.Duration.Companion.seconds +import kotlin.time.TimeSource +import kotlinx.coroutines.CoroutineExceptionHandler +import kotlinx.coroutines.CoroutineName +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.launchIn +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.flow.withIndex +import kotlinx.coroutines.isActive +import kotlinx.coroutines.job +import kotlinx.coroutines.launch +import kotlinx.coroutines.supervisorScope +import kotlinx.coroutines.suspendCancellableCoroutine +import org.http4k.routing.bind +import org.http4k.routing.websockets +import org.http4k.websocket.Websocket +import org.http4k.websocket.WsHandler +import org.http4k.websocket.WsMessage +import org.http4k.websocket.WsStatus + + +class WebsocketServer { + + private val serverScope = CoroutineScope( + CoroutineName("WsServer") + + Dispatchers.IO + + SupervisorJob() + + CoroutineExceptionHandler { _, e -> + println("[ws.WsServer] handling exception ${e.message}") + } + ) + + + private val clients: MutableSharedFlow = MutableSharedFlow( + replay = 100, + extraBufferCapacity = 100, + BufferOverflow.DROP_OLDEST, + ) + + + private val messages: MutableSharedFlow = MutableSharedFlow( + replay = 100, + extraBufferCapacity = 100, + BufferOverflow.DROP_OLDEST, + ) + + + private fun bindClient(ws: Websocket) = serverScope.launch { + println("[ws.bindClient] ${ws.upgradeRequest.toMessage()}") + clients.emit(ws) + } + + + fun sendMessage(msg: String) = serverScope.launch { + println("[ws.sendMessage] $msg") + messages.emit(msg) + } + + + fun build(): WsHandler { + val startTime = TimeSource.Monotonic.markNow() + + serverScope.launch { + supervisorScope { + clients + .withIndex() + .onEach { (id, ws) -> + createClient(id, ws) + }.launchIn(serverScope) + } + } + + serverScope.launch { + while (isActive) { + delay(10.seconds) + sendMessage("elapsed ${startTime.elapsedNow()}") + } + } + + return websockets( + "/ws/foo" bind { ws: Websocket -> + serverScope.launch { bindClient(ws) } + } + ) + } + + + private fun createClient(id: Int, ws: Websocket) { + val clientJob = SupervisorJob(serverScope.coroutineContext.job) + + serverScope.launch(CoroutineName("ws-client-$id") + clientJob) { +// return serverScope.launch(CoroutineName("ws-client-$id")) { +// supervisorScope { + launch { + + suspendCancellableCoroutine { cont -> + println("[ws-client] entering suspendCancellableCoroutine") + + cont.invokeOnCancellation { e: Throwable? -> + launch { sendMessage("[ws-client.invokeOnCompletion] closing client ws ${e?.message}") } + when (e) { + null -> ws.close(WsStatus.NORMAL) + else -> ws.close(WsStatus.ABNORMAL_CLOSE) + } + } + + ws.onError { e: Throwable -> + launch { sendMessage("[ws-client.onError] client $id error [${e.message}]") } + if (!cont.isCompleted) { + cont.resumeWithException(e) + } + } + + ws.onClose { status: WsStatus -> + launch { sendMessage("[ws-client.onClose] client $id disconnected [${status}]") } + if (!cont.isCompleted) { + cont.resume(status) + } + } + + ws.onMessage { wsMsg -> + launch { sendMessage("$id says: ${wsMsg.bodyString()}") } + } + } + } + + println("[ws-client] $id listening to messages") + messages + .onEach { msg -> + println("[ws-client] sending message to $id") + ws.send(WsMessage(msg)) + } + .launchIn(this) + } + + +// } + } +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt index df4423e..f19cfc5 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt @@ -6,6 +6,8 @@ import dev.adamko.kafkatorio.processor.admin.topicName import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kafkatorio.schema.common.FactorioServerId import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket +import dev.adamko.kotka.extensions.component1 +import dev.adamko.kotka.extensions.component2 import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.producedAs import dev.adamko.kotka.extensions.streams.map @@ -15,7 +17,6 @@ import kotlinx.serialization.decodeFromString import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.kstream.KStream -import org.apache.kafka.streams.processor.RecordContext fun factorioServerPacketStream( @@ -46,7 +47,7 @@ fun splitFactorioServerPacketStream( jsonMapper.serde(), jsonMapper.serde(), ) - ) { _: FactorioServerId, value: KafkatorioPacket, _: RecordContext -> + ) { (_: FactorioServerId, value: KafkatorioPacket) -> // println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}") value.data.topicName } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt index 968a924..d80cb71 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt @@ -1,18 +1,22 @@ package dev.adamko.kafkatorio.processor.topology -import dev.adamko.kafkatorio.processor.WebsocketServer import dev.adamko.kafkatorio.processor.admin.TOPIC_BROADCAST_TO_WEBSOCKET import dev.adamko.kafkatorio.processor.admin.topicName +import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer import dev.adamko.kafkatorio.schema.common.FactorioServerId import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio import dev.adamko.kafkatorio.schema.packets.EventServerPacket import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket import dev.adamko.kafkatorio.schema.packets.PlayerUpdate +import dev.adamko.kafkatorio.schema.packets.SurfaceUpdate import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.producedAs +import dev.adamko.kotka.extensions.stream import dev.adamko.kotka.extensions.streams.forEach import dev.adamko.kotka.extensions.streams.map +import dev.adamko.kotka.extensions.streams.to import dev.adamko.kotka.kxs.serde +import kotlinx.coroutines.runBlocking import kotlinx.serialization.encodeToString import kotlinx.serialization.json.Json import org.apache.kafka.streams.StreamsBuilder @@ -32,7 +36,7 @@ fun broadcastToWebsocket( builder: StreamsBuilder = StreamsBuilder(), ) { - builder.preparePlayerUpdatesBroadcast() + builder.broadcastKafkatorioPackets() builder.stream( TOPIC_BROADCAST_TO_WEBSOCKET, @@ -40,25 +44,54 @@ fun broadcastToWebsocket( ).forEach( "$pid.websocket-server.send-message" ) { _, esPacket -> - websocketServer.sendMessage(jsonMapper.encodeToString(esPacket)) + runBlocking { + websocketServer.sendMessage(jsonMapper.encodeToString(esPacket)) + } } } -private fun StreamsBuilder.preparePlayerUpdatesBroadcast() { - stream( +private fun StreamsBuilder.broadcastKafkatorioPackets() { + val playerUpdatesStream = stream( + consumedAs( + "$pid.player-updates", + jsonMapper.serde(FactorioServerId.serializer()), + jsonMapper.serde(KafkatorioPacket.serializer()), + ), PlayerUpdate::class.topicName, - consumedAs("$pid.player-updates", jsonMapper.serde(), jsonMapper.serde()), - ).map( - "$pid.player-updates.convert" - ) { serverId, playerPacket -> - serverId to EventServerPacket.Kafkatorio(playerPacket) + SurfaceUpdate::class.topicName, + ) + + playerUpdatesStream.map( + "$pid.packet-updates.convert" + ) { serverId, packet -> + serverId to EventServerPacket.Kafkatorio(packet) + }.to( + producedAs( + "$pid.event-server-packets.produce", + jsonMapper.serde(FactorioServerId.serializer()), + jsonMapper.serde(EventServerPacket.serializer()), + ) + ) { TOPIC_BROADCAST_TO_WEBSOCKET } +} + + +private fun StreamsBuilder.prepareServerUpdateBroadcast() { + + val surfaceUpdatesStream = stream( + SurfaceUpdate::class.topicName, + consumedAs("$pid.surface-updates", jsonMapper.serde(), jsonMapper.serde()), + ) + + surfaceUpdatesStream.map( + "$pid.surface-updates.convert" + ) { serverId, serverUpdatePacket -> + serverId to EventServerPacket.Kafkatorio(serverUpdatePacket) }.to( - TOPIC_BROADCAST_TO_WEBSOCKET, producedAs( "$pid.player-updates.stream", jsonMapper.serde(FactorioServerId.serializer()), jsonMapper.serde(EventServerPacket.serializer()), ) - ) + ) { TOPIC_BROADCAST_TO_WEBSOCKET } } diff --git a/modules/events-server/src/main/resources/.gitignore b/modules/events-server/src/main/resources/.gitignore deleted file mode 100644 index 46d00ec..0000000 --- a/modules/events-server/src/main/resources/.gitignore +++ /dev/null @@ -1 +0,0 @@ -kafkatorio-web-map/**