diff --git a/.gitignore b/.gitignore index a9942d5..fc1afcd 100644 --- a/.gitignore +++ b/.gitignore @@ -2,10 +2,12 @@ # ignore any files that start with '.secret' .secret.* -.idea +.state/ + +.idea/ !.idea/icon.svg -node_modules +node_modules/ *.hprof @@ -16,108 +18,13 @@ node_modules # Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio, WebStorm and Rider # Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839 -# User-specific stuff -.idea/**/workspace.xml -.idea/**/tasks.xml -.idea/**/usage.statistics.xml -.idea/**/dictionaries -.idea/**/shelf - -# AWS User-specific -.idea/**/aws.xml - -# Generated files -.idea/**/contentModel.xml - -# Sensitive or high-churn files -.idea/**/dataSources/ -.idea/**/dataSources.ids -.idea/**/dataSources.local.xml -.idea/**/sqlDataSources.xml -.idea/**/dynamic.xml -.idea/**/uiDesigner.xml -.idea/**/dbnavigator.xml - -# Gradle -.idea/**/gradle.xml -.idea/**/libraries - -# Gradle and Maven with auto-import -# When using Gradle or Maven with auto-import, you should exclude module files, -# since they will be recreated, and may cause churn. Uncomment if using -# auto-import. -.idea/artifacts -.idea/compiler.xml -.idea/jarRepositories.xml -.idea/modules.xml -.idea/*.iml -.idea/modules +# File-based project format +*.iws *.iml *.ipr -# CMake -cmake-build-*/ - -# Mongo Explorer plugin -.idea/**/mongoSettings.xml - -# File-based project format -*.iws - -# IntelliJ out/ -# mpeltonen/sbt-idea plugin -.idea_modules/ - -# JIRA plugin -atlassian-ide-plugin.xml - -# Cursive Clojure plugin -.idea/replstate.xml - -# Crashlytics plugin (for Android Studio and IntelliJ) -com_crashlytics_export_strings.xml -crashlytics.properties -crashlytics-build.properties -fabric.properties - -# Editor-based Rest Client -.idea/httpRequests - -# Android studio 3.1+ serialized cache file -.idea/caches/build_file_checksums.ser - -### Intellij Patch ### -# Comment Reason: https://github.com/joeblau/gitignore.io/issues/186#issuecomment-215987721 - -# *.iml -# modules.xml -# .idea/misc.xml -# *.ipr - -# Sonarlint plugin -# https://plugins.jetbrains.com/plugin/7973-sonarlint -.idea/**/sonarlint/ - -# SonarQube Plugin -# https://plugins.jetbrains.com/plugin/7238-sonarqube-community-plugin -.idea/**/sonarIssues.xml - -# Markdown Navigator plugin -# https://plugins.jetbrains.com/plugin/7896-markdown-navigator-enhanced -.idea/**/markdown-navigator.xml -.idea/**/markdown-navigator-enh.xml -.idea/**/markdown-navigator/ - -# Cache file creation bug -# See https://youtrack.jetbrains.com/issue/JBR-2257 -.idea/$CACHE_FILE$ - -# CodeStream plugin -# https://plugins.jetbrains.com/plugin/12206-codestream -.idea/codestream.xml - ### Kotlin ### # Compiled class file *.class @@ -125,12 +32,6 @@ fabric.properties # Log file *.log -# BlueJ files -*.ctxt - -# Mobile Tools for Java (J2ME) -.mtj.tmp/ - # Package Files # *.jar *.war @@ -198,7 +99,6 @@ Temporary Items # Local History for Visual Studio Code .history/ -### VisualStudioCode Patch ### # Ignore all local history of files .history .ionide diff --git a/build.gradle.kts b/build.gradle.kts index da1e252..51241c3 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -41,17 +41,17 @@ tasks.wrapper { } val startInfra by tasks.registering { - group = project.name + group = rootProject.name dependsOn( // ":modules:infra-kafka-pipe:processRun", - runKafkaCluster +// runKafkaCluster // ":modules:infra-kafka-connect:processRun", ) } val runKafkatorio by tasks.registering { - group = project.name + group = rootProject.name dependsOn( ":modules:infra-factorio-client:processRestart", @@ -60,44 +60,44 @@ val runKafkatorio by tasks.registering { ) } -val runEventsServer by tasks.registering { - group = project.name +//val runEventsServer by tasks.registering { +// group = project.name +// +// dependsOn(":modules:events-server:run") +//} - dependsOn(":modules:events-server:run") -} +//val runWebMap by tasks.registering { +// group = project.name +// +// dependsOn(":modules:web-map:browserDevelopmentRun") +//} -val runWebMap by tasks.registering { - group = project.name - - dependsOn(":modules:web-map:browserDevelopmentRun") -} +// +//val runKafkaCluster by tasks.registering { +// group = project.name +// +// dependsOn( +// ":modules:infra-kafka-cluster:processRun", +// ) +//} -val runKafkaCluster by tasks.registering { - group = project.name - - dependsOn( - ":modules:infra-kafka-cluster:processRun", - ) -} +//val runFactorioServer by tasks.registering { +// group = project.name +// +// dependsOn( +// ":modules:infra-factorio-server:processRun", +// ) +//} -val runFactorioServer by tasks.registering { - group = project.name - - dependsOn( - ":modules:infra-factorio-server:processRun", - ) -} - - -val runFactorioClient by tasks.registering { - group = project.name - - dependsOn( - ":modules:infra-factorio-client:processRun", - ) -} +//val runFactorioClient by tasks.registering { +// group = project.name +// +// dependsOn( +// ":modules:infra-factorio-client:processRun", +// ) +//} apply(from = "$projectDir/kt52647.gradle.kts") diff --git a/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/base.gradle.kts b/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/base.gradle.kts index 1352a4e..19385e5 100644 --- a/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/base.gradle.kts +++ b/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/base.gradle.kts @@ -1,7 +1,5 @@ package dev.adamko.kafkatorio -import org.gradle.api.internal.tasks.userinput.UserInputHandler -import org.gradle.kotlin.dsl.support.serviceOf plugins { base diff --git a/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/gradle/extensions.kt b/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/gradle/extensions.kt index 7693f66..f3c0743 100644 --- a/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/gradle/extensions.kt +++ b/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/gradle/extensions.kt @@ -7,9 +7,13 @@ import org.gradle.api.file.Directory import org.gradle.api.file.FileCopyDetails import org.gradle.api.file.ProjectLayout import org.gradle.api.file.RelativePath +import org.gradle.api.plugins.JavaPlugin import org.gradle.api.provider.Provider import org.gradle.api.specs.NotSpec import org.gradle.api.specs.Spec +import org.gradle.api.tasks.TaskProvider +import org.gradle.api.tasks.application.CreateStartScripts +import org.gradle.kotlin.dsl.register import org.gradle.plugins.ide.idea.model.IdeaModule import org.gradle.process.ExecSpec import org.jetbrains.kotlin.gradle.targets.js.yarn.YarnRootExtension @@ -94,3 +98,21 @@ fun Project.taskProvider(taskName: String): Provider = providers.provider const val DOCKER_COMPOSE_TASK_GROUP = "docker-compose" + + +fun Project.registerStartScriptTask( + appName: String, + configureStartScripts: CreateStartScripts.() -> Unit = {}, +): TaskProvider = + tasks.register("startScript$appName") { + + val jarTask = tasks.getByName(JavaPlugin.JAR_TASK_NAME) + val runtimeClasspath = configurations.getByName(JavaPlugin.RUNTIME_CLASSPATH_CONFIGURATION_NAME) + dependsOn(jarTask) + dependsOn(runtimeClasspath) + + applicationName = appName + outputDir = layout.buildDirectory.dir("scripts").get().asFile + classpath = jarTask.outputs.files + runtimeClasspath + configureStartScripts() + } diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 2833d4e..8233f44 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -11,7 +11,7 @@ kotlin = "1.7.10" ## libs ## camel-kafkaConnectors = "0.11.0" http4k = "4.25.16.2" -kafka = "3.2.1" # https://kafka.apache.org/downloads +kafka = "3.2.2" # https://kafka.apache.org/downloads kvision = "5.14.0" # https://github.com/rjaros/kvision/releases log4j2 = "2.17.2" okio = "3.2.0" # https://search.maven.org/artifact/com.squareup.okio/okio diff --git a/modules/events-processor-core/build.gradle.kts b/modules/events-processor-core/build.gradle.kts index 7b92be6..f681852 100644 --- a/modules/events-processor-core/build.gradle.kts +++ b/modules/events-processor-core/build.gradle.kts @@ -4,15 +4,15 @@ import org.jetbrains.kotlin.gradle.tasks.KotlinCompile plugins { dev.adamko.kafkatorio.lang.`kotlin-jvm` kotlin("plugin.serialization") - application +// application } -//description = "serves Kafkatorio content over the web" +description = "Shared Kafkatorio processors functionality" val projectId: String by project.extra -val kafkaStateDir: Directory = layout.projectDirectory.dir("kafka-state") +val kafkaStateDir: Directory = layout.projectDirectory.dir(".state/kafka") dependencies { @@ -37,12 +37,17 @@ dependencies { implementation(libs.simpleSyslogParser) implementation(libs.kafka.kotkaStreams) + + implementation(libs.bundles.hoplite) + + implementation(libs.ktorServer.authJwt) +// implementation(libs.nimbusJoseJwt) } -application { - mainClass.set("dev.adamko.kafkatorio.server.EventsServerKt") -} +//application { +// mainClass.set("dev.adamko.kafkatorio.processor.admin.MainKt") +//} tasks.withType().configureEach { @@ -52,3 +57,10 @@ tasks.withType().configureEach { "-opt-in=kotlinx.serialization.ExperimentalSerializationApi", ) } + + +val createKafkaTopics by tasks.registering(JavaExec::class) { + group = rootProject.name + classpath = sourceSets.main.get().runtimeClasspath + mainClass.set("dev.adamko.kafkatorio.processor.admin.MainKt") +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/admin/KafkatorioKafkaAdmin.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/KafkatorioKafkaAdmin.kt similarity index 90% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/admin/KafkatorioKafkaAdmin.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/KafkatorioKafkaAdmin.kt index a052945..889ab1f 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/admin/KafkatorioKafkaAdmin.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/KafkatorioKafkaAdmin.kt @@ -1,7 +1,7 @@ -package dev.adamko.kafkatorio.server.processor.admin +package dev.adamko.kafkatorio.processor.admin -import dev.adamko.kafkatorio.server.config.ApplicationProperties -import dev.adamko.kafkatorio.server.processor.allTopics +import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import dev.adamko.kafkatorio.processor.config.allTopics import java.util.concurrent.TimeUnit import mu.KotlinLogging import org.apache.kafka.clients.admin.Admin @@ -11,7 +11,7 @@ import org.apache.kafka.clients.admin.TopicListing class KafkatorioKafkaAdmin( - appProps: ApplicationProperties + appProps: ApplicationProperties, ) { private val logger = KotlinLogging.logger {} diff --git a/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/main.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/main.kt new file mode 100644 index 0000000..7819a59 --- /dev/null +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/main.kt @@ -0,0 +1,10 @@ +package dev.adamko.kafkatorio.processor.admin + +import dev.adamko.kafkatorio.processor.config.ApplicationProperties + + +fun main() { + val appProps = ApplicationProperties.load() + val admin = KafkatorioKafkaAdmin(appProps) + admin.createKafkatorioTopics() +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/ApplicationProperties.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt similarity index 76% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/ApplicationProperties.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt index 9b7d7a4..7d8a020 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/ApplicationProperties.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt @@ -1,15 +1,16 @@ -package dev.adamko.kafkatorio.server.config +package dev.adamko.kafkatorio.processor.config import com.sksamuel.hoplite.ConfigLoaderBuilder import com.sksamuel.hoplite.addEnvironmentSource import com.sksamuel.hoplite.addResourceOrFileSource +import com.sksamuel.hoplite.sources.UserSettingsPropertySource import dev.adamko.kafkatorio.schema.common.FactorioServerId import dev.adamko.kafkatorio.schema.common.FactorioServerToken import dev.adamko.kafkatorio.schema.common.MaskedValue import java.nio.file.Path +val appProps = ApplicationProperties.load() -// loaded by hoplite data class ApplicationProperties( val webPort: PortVal, val serverDataDir: Path, @@ -32,8 +33,13 @@ data class ApplicationProperties( .allowUnresolvedSubstitutions() .allowEmptySources() .addEnvironmentSource() + .addResourceOrFileSource( + resourceOrFile = "${System.getProperty("user.home")}/.secret.kafkatorio-config.yml", + optional = true + ) .addResourceOrFileSource("/.secret.config.yml", optional = true) .addResourceOrFileSource("/config.yml", optional = false) + .addPropertySource(UserSettingsPropertySource) .build() .loadConfigOrThrow() } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/PortVal.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/PortVal.kt similarity index 90% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/PortVal.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/PortVal.kt index a4c48c8..29ebb4f 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/PortVal.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/PortVal.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.server.config +package dev.adamko.kafkatorio.processor.config /** Improved (slightly) version of http4k's `Port`. */ diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/kafkatorioTopics.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/kafkatorioTopics.kt similarity index 98% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/kafkatorioTopics.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/kafkatorioTopics.kt index b3af92d..36d70ba 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/kafkatorioTopics.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/kafkatorioTopics.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.server.processor +package dev.adamko.kafkatorio.processor.config import dev.adamko.kafkatorio.schema.packets.ConfigurationUpdate import dev.adamko.kafkatorio.schema.packets.ConsoleChatUpdate diff --git a/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/kxsJson.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/kxsJson.kt new file mode 100644 index 0000000..364d089 --- /dev/null +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/config/kxsJson.kt @@ -0,0 +1,12 @@ +package dev.adamko.kafkatorio.processor.config + +import dev.adamko.kafkatorio.library.jsonMapperKafkatorio +import kotlinx.serialization.json.Json + + +@PublishedApi +internal val jsonMapper: Json = Json { + prettyPrint = true + prettyPrintIndent = " " + serializersModule = jsonMapperKafkatorio.serializersModule +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/security/Authenticator.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/Authenticator.kt similarity index 97% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/security/Authenticator.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/Authenticator.kt index 7c95183..dd842d5 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/security/Authenticator.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/Authenticator.kt @@ -1,10 +1,10 @@ -package dev.adamko.kafkatorio.server.security +package dev.adamko.kafkatorio.processor.core import com.auth0.jwt.JWT import com.auth0.jwt.algorithms.Algorithm import com.auth0.jwt.interfaces.DecodedJWT +import dev.adamko.kafkatorio.processor.config.ApplicationProperties import dev.adamko.kafkatorio.schema.common.FactorioServerToken -import dev.adamko.kafkatorio.server.config.ApplicationProperties import kotlin.reflect.jvm.jvmName import kotlinx.coroutines.CancellationException import kotlinx.serialization.SerialName diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/DebounceProcessor.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/DebounceProcessor.kt similarity index 98% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/DebounceProcessor.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/DebounceProcessor.kt index b3cd20f..e7aefda 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/DebounceProcessor.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/DebounceProcessor.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.server.processor +package dev.adamko.kafkatorio.processor.core import dev.adamko.kotka.extensions.component1 import dev.adamko.kotka.extensions.component2 diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/StreamsExceptionHandler.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/StreamsExceptionHandler.kt similarity index 91% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/StreamsExceptionHandler.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/StreamsExceptionHandler.kt index 03f0cc3..f29e953 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/StreamsExceptionHandler.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/StreamsExceptionHandler.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.server.processor +package dev.adamko.kafkatorio.processor.core import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse diff --git a/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/topologyLauncher.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/topologyLauncher.kt new file mode 100644 index 0000000..03ff1cc --- /dev/null +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/processor/core/topologyLauncher.kt @@ -0,0 +1,88 @@ +package dev.adamko.kafkatorio.processor.core + +import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import java.util.Properties +import kotlin.coroutines.resume +import kotlin.coroutines.resumeWithException +import kotlin.io.path.Path +import kotlinx.coroutines.CoroutineName +import kotlinx.coroutines.cancel +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.job +import kotlinx.coroutines.launch +import kotlinx.coroutines.suspendCancellableCoroutine +import org.apache.kafka.streams.KafkaStreams +import org.apache.kafka.streams.StreamsConfig +import org.apache.kafka.streams.Topology +import org.apache.kafka.streams.TopologyDescription + + +private val appProps = ApplicationProperties.load() + +suspend fun launchTopology( + id: String, + topology: Topology, + additionalProperties: Map = mapOf(), +): Unit = coroutineScope { + + val props: Properties = appProps.kafkaStreamsConfig.toProperties() + + additionalProperties.forEach { (k, v) -> props.setProperty(k, v) } + + val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as String + props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId) + + 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.PENDING_SHUTDOWN, + KafkaStreams.State.NOT_RUNNING, + KafkaStreams.State.PENDING_ERROR, + KafkaStreams.State.ERROR -> if (cont.isActive) 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 { + parentFile.mkdirs() + if (!exists()) createNewFile() + writeText(description.toString()) + } + + println( + """ + |---------------- + |$appId - ${descFile.canonicalPath} + |$description + |---------------- + """.trimMargin() + ) +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourMapChunks.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourMapChunks.kt similarity index 98% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourMapChunks.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourMapChunks.kt index 9bcf517..8675fec 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourMapChunks.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourMapChunks.kt @@ -3,6 +3,7 @@ package dev.adamko.kafkatorio.server.processor.topology import dev.adamko.kafkatorio.library.kxsBinary import dev.adamko.kafkatorio.library.resize import dev.adamko.kafkatorio.library.toMapChunkPosition +import dev.adamko.kafkatorio.processor.core.DebounceProcessor.Companion.addDebounceProcessor import dev.adamko.kafkatorio.schema.common.ChunkSize import dev.adamko.kafkatorio.schema.common.ColourHex import dev.adamko.kafkatorio.schema.common.FactorioServerId @@ -12,9 +13,8 @@ import dev.adamko.kafkatorio.schema.common.ServerMapChunkId import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer import dev.adamko.kafkatorio.schema.packets.MapChunkTileUpdate import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate -import dev.adamko.kafkatorio.server.processor.DebounceProcessor.Companion.addDebounceProcessor -import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_TERRAIN_COLOURED_032_UPDATES -import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE +import dev.adamko.kafkatorio.processor.config.TOPIC_MAP_CHUNK_TERRAIN_COLOURED_032_UPDATES +import dev.adamko.kafkatorio.processor.config.TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.groupedAs import dev.adamko.kotka.extensions.materializedAs diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourResourceChunks.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourResourceChunks.kt similarity index 97% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourResourceChunks.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourResourceChunks.kt index 96d1835..965a47c 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourResourceChunks.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/colourResourceChunks.kt @@ -2,6 +2,7 @@ package dev.adamko.kafkatorio.server.processor.topology import dev.adamko.kafkatorio.library.kxsBinary import dev.adamko.kafkatorio.library.toMapTilePosition +import dev.adamko.kafkatorio.processor.core.DebounceProcessor.Companion.addDebounceProcessor import dev.adamko.kafkatorio.schema.common.ChunkSize import dev.adamko.kafkatorio.schema.common.ColourHex import dev.adamko.kafkatorio.schema.common.FactorioPrototype @@ -16,7 +17,6 @@ import dev.adamko.kafkatorio.schema.packets.MapChunkEntityUpdate import dev.adamko.kafkatorio.schema.packets.MapChunkResourceUpdate import dev.adamko.kafkatorio.schema.packets.MapChunkTileUpdate import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate -import dev.adamko.kafkatorio.server.processor.DebounceProcessor.Companion.addDebounceProcessor import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.materializedAs import dev.adamko.kotka.extensions.stream @@ -27,7 +27,6 @@ import dev.adamko.kotka.extensions.tables.join import dev.adamko.kotka.kxs.serde import kotlin.collections.component1 import kotlin.collections.component2 -import kotlin.math.ceil import kotlin.math.floor import kotlin.time.Duration.Companion.seconds import org.apache.kafka.streams.StreamsBuilder @@ -232,6 +231,9 @@ private fun KTable>.enr private fun MapBoundingBox.iterator(origin: MapTilePosition): Iterator { + // check if tile width/height should be divided by 2. I think I read that it's not divided by 2 + // if it's an odd number? + val xOffset: Int = floor(tileWidth.toDouble() / 2).toInt() val startX = origin.x - xOffset val endX = origin.x + xOffset diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt similarity index 93% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt index 20ecdab..42999ab 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt @@ -4,9 +4,9 @@ import dev.adamko.kafkatorio.schema.common.FactorioServerId import dev.adamko.kafkatorio.schema.common.Tick import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketDataError -import dev.adamko.kafkatorio.server.config.jsonMapper -import dev.adamko.kafkatorio.server.processor.TOPIC_SRC_SERVER_LOG -import dev.adamko.kafkatorio.server.processor.topicName +import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG +import dev.adamko.kafkatorio.processor.config.jsonMapper +import dev.adamko.kafkatorio.processor.config.topicName import dev.adamko.kotka.extensions.component1 import dev.adamko.kotka.extensions.component2 import dev.adamko.kotka.extensions.consumedAs diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/model.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/model.kt similarity index 100% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/model.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/model.kt diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/packetDataStreams.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/packetDataStreams.kt similarity index 91% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/packetDataStreams.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/packetDataStreams.kt index f21625f..3adc34c 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/packetDataStreams.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/packetDataStreams.kt @@ -1,10 +1,10 @@ package dev.adamko.kafkatorio.server.processor.topology +import dev.adamko.kafkatorio.processor.config.jsonMapper import dev.adamko.kafkatorio.schema.common.FactorioServerId import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData -import dev.adamko.kafkatorio.server.config.jsonMapper -import dev.adamko.kafkatorio.server.processor.topicName +import dev.adamko.kafkatorio.processor.config.topicName import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.streams.filter import dev.adamko.kotka.extensions.streams.mapValues diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/prototypes.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/prototypes.kt similarity index 100% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/prototypes.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/prototypes.kt diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/saveMapTiles.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/saveMapTiles.kt similarity index 94% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/saveMapTiles.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/saveMapTiles.kt index bff04b3..d0364a5 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/saveMapTiles.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/saveMapTiles.kt @@ -10,11 +10,11 @@ import dev.adamko.kafkatorio.schema.common.FactorioServerId import dev.adamko.kafkatorio.schema.common.ServerMapChunkId import dev.adamko.kafkatorio.schema.common.ServerMapTilePngFilename import dev.adamko.kafkatorio.schema.packets.EventServerPacket -import dev.adamko.kafkatorio.server.config.jsonMapper -import dev.adamko.kafkatorio.server.processor.TOPIC_BROADCAST_TO_WEBSOCKET -import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE -import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE -import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE +import dev.adamko.kafkatorio.processor.config.TOPIC_BROADCAST_TO_WEBSOCKET +import dev.adamko.kafkatorio.processor.config.TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE +import dev.adamko.kafkatorio.processor.config.TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE +import dev.adamko.kafkatorio.processor.config.TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE +import dev.adamko.kafkatorio.processor.config.jsonMapper import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.producedAs import dev.adamko.kotka.extensions.stream @@ -44,7 +44,7 @@ private const val WEB_MAP_TILE_SIZE_PX = 256 fun saveMapTiles( builder: StreamsBuilder, serverDataDir: Path, -): Topology { +) { val subdividedMapChunkTilesDebounced: KStream> = builder.stream( @@ -64,8 +64,6 @@ fun saveMapTiles( subdividedMapChunkTilesDebounced.saveChunkAsImage(serverDataDir) savedChunkFilenames.broadcastSavedTileUpdates() - - return builder.build() } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/websocketBroadcaster.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/websocketBroadcaster.kt similarity index 89% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/websocketBroadcaster.kt rename to modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/websocketBroadcaster.kt index a173a26..5fc2d85 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/websocketBroadcaster.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/websocketBroadcaster.kt @@ -6,9 +6,8 @@ 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.kafkatorio.server.processor.TOPIC_BROADCAST_TO_WEBSOCKET -import dev.adamko.kafkatorio.server.processor.topicName -import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer +import dev.adamko.kafkatorio.processor.config.TOPIC_BROADCAST_TO_WEBSOCKET +import dev.adamko.kafkatorio.processor.config.topicName import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.producedAs import dev.adamko.kotka.extensions.stream @@ -31,8 +30,13 @@ private val jsonMapper: Json = Json { private const val pid = "websocketBroadcast" +interface WebsocketBroadcaster { + suspend fun sendMessageToClients(message: String) +} + + fun broadcastToWebsocket( - websocketServer: WebmapWebsocketServer, + websocketBroadcaster: WebsocketBroadcaster, builder: StreamsBuilder, ) { @@ -45,7 +49,7 @@ fun broadcastToWebsocket( "$pid.websocket-server.send-message" ) { _, esPacket -> runBlocking { - websocketServer.sendMessageToClients(jsonMapper.encodeToString(esPacket)) + websocketBroadcaster.sendMessageToClients(jsonMapper.encodeToString(esPacket)) } } } diff --git a/modules/events-server/src/main/resources/config.yml b/modules/events-processor-core/src/main/resources/config.yml similarity index 89% rename from modules/events-server/src/main/resources/config.yml rename to modules/events-processor-core/src/main/resources/config.yml index c377231..b9f2858 100644 --- a/modules/events-server/src/main/resources/config.yml +++ b/modules/events-processor-core/src/main/resources/config.yml @@ -5,7 +5,7 @@ server-data-dir: ${SERVER_DATA_DIR:-./server-data} socket-server-host: ${SOCKET_SERVER_HOST:-127.0.0.1} -#socket-server-port: ${SOCKET_SERVER_PORT:-12092} +socket-server-port: ${SOCKET_SERVER_PORT:-12092} kafka-admin-config: @@ -16,7 +16,7 @@ kafka-admin-config: kafka-streams-config: application.id: kafkatorio-events-processor bootstrap.servers: ${KAFKA_BOOTSTRAP_SERVERS:-http://localhost:9092} - state.dir: ${KAFKA_STREAMS_STATE_DIR:-./kafka-state} + state.dir: ${KAFKA_STREAMS_STATE_DIR:-.state/kafka} num.stream.threads: 1 processing.guarantee: exactly_once_v2 auto.offset.reset: earliest @@ -65,5 +65,3 @@ kafka-streams-config: # # fetch.message.max.bytes=31457280 # # 100mb 104857600 # -#tileServer: -# webmapTileDir: 'D:\Users\Adam\Projects\games\kafkatorio\modules\events-server\kafkatorio-tiles' diff --git a/modules/events-server/src/main/resources/external-annotations/org/apache/kafka/streams/state/annotations.xml b/modules/events-processor-core/src/main/resources/external-annotations/org/apache/kafka/streams/state/annotations.xml similarity index 100% rename from modules/events-server/src/main/resources/external-annotations/org/apache/kafka/streams/state/annotations.xml rename to modules/events-processor-core/src/main/resources/external-annotations/org/apache/kafka/streams/state/annotations.xml diff --git a/modules/events-server/src/main/resources/logback.xml b/modules/events-processor-core/src/main/resources/logback.xml similarity index 100% rename from modules/events-server/src/main/resources/logback.xml rename to modules/events-processor-core/src/main/resources/logback.xml diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/security/AuthenticatorTest.kt b/modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/AuthenticatorTest.kt similarity index 97% rename from modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/security/AuthenticatorTest.kt rename to modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/AuthenticatorTest.kt index 0241fdb..5fb2d4f 100644 --- a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/security/AuthenticatorTest.kt +++ b/modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/AuthenticatorTest.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.server.security +package dev.adamko.kafkatorio.processor.core import dev.adamko.kafkatorio.schema.common.MaskedValue import dev.adamko.kafkatorio.server.config.ApplicationProperties diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt b/modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/misc/serdeSize.kt similarity index 100% rename from modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt rename to modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/misc/serdeSize.kt diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/processor/topology/FactorioServerPacketsTest.kt b/modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/topology/FactorioServerPacketsTest.kt similarity index 96% rename from modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/processor/topology/FactorioServerPacketsTest.kt rename to modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/topology/FactorioServerPacketsTest.kt index 0cadcec..b219310 100644 --- a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/processor/topology/FactorioServerPacketsTest.kt +++ b/modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/topology/FactorioServerPacketsTest.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.server.processor.topology +package dev.adamko.kafkatorio.processor.core.topology import dev.adamko.kafkatorio.library.kxsBinary import dev.adamko.kafkatorio.schema.common.ColourHex diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt b/modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/topology/GroupMapChunksTest.kt similarity index 99% rename from modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt rename to modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/topology/GroupMapChunksTest.kt index 56cb643..5dc33de 100644 --- a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt +++ b/modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/topology/GroupMapChunksTest.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.processor.topology +package dev.adamko.kafkatorio.processor.core.topology //import dev.adamko.kafkatorio.server.processor.TOPIC_GROUPED_MAP_CHUNKS_STATE //import dev.adamko.kafkatorio.processor.config.getJsonMapper diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/processor/topology/SaveMapChunksTest.kt b/modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/topology/SaveMapChunksTest.kt similarity index 99% rename from modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/processor/topology/SaveMapChunksTest.kt rename to modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/topology/SaveMapChunksTest.kt index 919bef3..a82ce0e 100644 --- a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/processor/topology/SaveMapChunksTest.kt +++ b/modules/events-processor-core/src/test/kotlin/dev/adamko/kafkatorio/processor/core/topology/SaveMapChunksTest.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.server.processor.topology +package dev.adamko.kafkatorio.processor.core.topology import dev.adamko.kafkatorio.library.iterateTiles import dev.adamko.kafkatorio.library.kxsBinary diff --git a/modules/events-processor-entities/build.gradle.kts b/modules/events-processor-entities/build.gradle.kts deleted file mode 100644 index 7b92be6..0000000 --- a/modules/events-processor-entities/build.gradle.kts +++ /dev/null @@ -1,54 +0,0 @@ -import org.jetbrains.kotlin.gradle.tasks.KotlinCompile - - -plugins { - dev.adamko.kafkatorio.lang.`kotlin-jvm` - kotlin("plugin.serialization") - application -} - - -//description = "serves Kafkatorio content over the web" - -val projectId: String by project.extra - -val kafkaStateDir: Directory = layout.projectDirectory.dir("kafka-state") - - -dependencies { - implementation(platform(projects.modules.versionsPlatform)) - - implementation(libs.kafka.streams) - testImplementation(libs.kafka.streamsTestUtils) - - implementation(projects.modules.eventsLibrary) - - implementation(libs.kotlinxSerialization.core) - implementation(libs.kotlinxSerialization.json) - - implementation(libs.kotlinx.coroutines.core) - - implementation(libs.bundles.logging) - - implementation(libs.scrimage.core) - - implementation(libs.okio.core) - - implementation(libs.simpleSyslogParser) - - implementation(libs.kafka.kotkaStreams) -} - - -application { - mainClass.set("dev.adamko.kafkatorio.server.EventsServerKt") -} - - -tasks.withType().configureEach { - kotlinOptions.freeCompilerArgs += listOf( - "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", - "-opt-in=kotlinx.coroutines.FlowPreview", - "-opt-in=kotlinx.serialization.ExperimentalSerializationApi", - ) -} diff --git a/modules/events-processor-packets/build.gradle.kts b/modules/events-processor-packets/build.gradle.kts deleted file mode 100644 index 7b92be6..0000000 --- a/modules/events-processor-packets/build.gradle.kts +++ /dev/null @@ -1,54 +0,0 @@ -import org.jetbrains.kotlin.gradle.tasks.KotlinCompile - - -plugins { - dev.adamko.kafkatorio.lang.`kotlin-jvm` - kotlin("plugin.serialization") - application -} - - -//description = "serves Kafkatorio content over the web" - -val projectId: String by project.extra - -val kafkaStateDir: Directory = layout.projectDirectory.dir("kafka-state") - - -dependencies { - implementation(platform(projects.modules.versionsPlatform)) - - implementation(libs.kafka.streams) - testImplementation(libs.kafka.streamsTestUtils) - - implementation(projects.modules.eventsLibrary) - - implementation(libs.kotlinxSerialization.core) - implementation(libs.kotlinxSerialization.json) - - implementation(libs.kotlinx.coroutines.core) - - implementation(libs.bundles.logging) - - implementation(libs.scrimage.core) - - implementation(libs.okio.core) - - implementation(libs.simpleSyslogParser) - - implementation(libs.kafka.kotkaStreams) -} - - -application { - mainClass.set("dev.adamko.kafkatorio.server.EventsServerKt") -} - - -tasks.withType().configureEach { - kotlinOptions.freeCompilerArgs += listOf( - "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", - "-opt-in=kotlinx.coroutines.FlowPreview", - "-opt-in=kotlinx.serialization.ExperimentalSerializationApi", - ) -} diff --git a/modules/events-processor-tiles/build.gradle.kts b/modules/events-processor-tiles/build.gradle.kts deleted file mode 100644 index 7b92be6..0000000 --- a/modules/events-processor-tiles/build.gradle.kts +++ /dev/null @@ -1,54 +0,0 @@ -import org.jetbrains.kotlin.gradle.tasks.KotlinCompile - - -plugins { - dev.adamko.kafkatorio.lang.`kotlin-jvm` - kotlin("plugin.serialization") - application -} - - -//description = "serves Kafkatorio content over the web" - -val projectId: String by project.extra - -val kafkaStateDir: Directory = layout.projectDirectory.dir("kafka-state") - - -dependencies { - implementation(platform(projects.modules.versionsPlatform)) - - implementation(libs.kafka.streams) - testImplementation(libs.kafka.streamsTestUtils) - - implementation(projects.modules.eventsLibrary) - - implementation(libs.kotlinxSerialization.core) - implementation(libs.kotlinxSerialization.json) - - implementation(libs.kotlinx.coroutines.core) - - implementation(libs.bundles.logging) - - implementation(libs.scrimage.core) - - implementation(libs.okio.core) - - implementation(libs.simpleSyslogParser) - - implementation(libs.kafka.kotkaStreams) -} - - -application { - mainClass.set("dev.adamko.kafkatorio.server.EventsServerKt") -} - - -tasks.withType().configureEach { - kotlinOptions.freeCompilerArgs += listOf( - "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", - "-opt-in=kotlinx.coroutines.FlowPreview", - "-opt-in=kotlinx.serialization.ExperimentalSerializationApi", - ) -} diff --git a/modules/events-processors/build.gradle.kts b/modules/events-processors/build.gradle.kts new file mode 100644 index 0000000..01ae8d4 --- /dev/null +++ b/modules/events-processors/build.gradle.kts @@ -0,0 +1,91 @@ +import org.jetbrains.kotlin.gradle.tasks.KotlinCompile + +plugins { + dev.adamko.kafkatorio.lang.`kotlin-jvm` + kotlin("plugin.serialization") + application +} + + +description = "processors for Kafkatorio packets" + +val projectId: String by project.extra + + +dependencies { + implementation(platform(projects.modules.versionsPlatform)) + + implementation(libs.kafka.streams) + + testImplementation(libs.kafka.streamsTestUtils) + + implementation(projects.modules.eventsLibrary) + implementation(projects.modules.eventsProcessorCore) + + implementation(libs.kotlinxSerialization.core) + implementation(libs.kotlinxSerialization.json) + + implementation(libs.kotlinx.coroutines.core) + + implementation(libs.bundles.logging) + + implementation(libs.scrimage.core) + + implementation(libs.okio.core) + + implementation(libs.simpleSyslogParser) + + implementation(libs.kafka.kotkaStreams) +} + + +application { + mainClass.set("dev.adamko.kafkatorio.processors.MainKt") +} + + +tasks.withType().configureEach { + kotlinOptions.freeCompilerArgs += listOf( + "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", + "-opt-in=kotlinx.coroutines.FlowPreview", + "-opt-in=kotlinx.serialization.ExperimentalSerializationApi", + ) +} + + +//val startScriptEntitiesProcessor = registerStartScriptTask("EntitiesProcessor") { +// mainClass.set("dev.adamko.kafkatorio.processors.EntitiesProcessorKt") +//} +// +//val startScriptPacketsProcessor = registerStartScriptTask("PacketsProcessor") { +// mainClass.set("dev.adamko.kafkatorio.processors.PacketsProcessorKt") +//} +// +//val startScriptTilesProcessor = registerStartScriptTask("TilesProcessor") { +// mainClass.set("dev.adamko.kafkatorio.processors.TilesProcessorKt") +//} +// +// +//application.applicationDistribution.into("bin") { +// from(startScriptEntitiesProcessor) +// from(startScriptPacketsProcessor) +// from(startScriptTilesProcessor) +// fileMode = 0b000_111_101_101 // 0755 +// duplicatesStrategy = DuplicatesStrategy.EXCLUDE +//} + +tasks.named("run") { + + val kafkaStateDir: Directory = layout.projectDirectory.dir(".state/kafka/") + + doFirst { + kafkaStateDir.asFile.mkdirs() + } +} + + +val runEventsProcessors by tasks.registering { + group = rootProject.name + + dependsOn(tasks.run) +} diff --git a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupEntitiesMapChunks.kt b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupEntitiesMapChunks.kt new file mode 100644 index 0000000..8783592 --- /dev/null +++ b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupEntitiesMapChunks.kt @@ -0,0 +1,54 @@ +package dev.adamko.kafkatorio.processors + +import dev.adamko.kafkatorio.processor.core.launchTopology +import dev.adamko.kafkatorio.schema.common.FactorioServerId +import dev.adamko.kafkatorio.schema.common.PrototypeHashCode +import dev.adamko.kafkatorio.schema.common.ServerMapChunkId +import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer +import dev.adamko.kafkatorio.schema.packets.MapChunkEntityUpdate +import dev.adamko.kafkatorio.schema.packets.MapChunkResourceUpdate +import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate +import dev.adamko.kafkatorio.server.processor.topology.ServerMapChunkTiles +import dev.adamko.kafkatorio.server.processor.topology.colourEntityChunks +import dev.adamko.kafkatorio.server.processor.topology.convertEntityUpdateToServerMapChunkTiles +import dev.adamko.kafkatorio.server.processor.topology.convertResourceUpdateToServerMapChunkTiles +import dev.adamko.kafkatorio.server.processor.topology.streamPacketData +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.kstream.KStream + + +/** Building and resource tiles */ +suspend fun groupEntitiesMapChunks( + layer: ServerMapTileLayer, + colouredUpdatesStreamTopic: String, + colouredChunkStateTopic: String, +) { + val builder = StreamsBuilder() + + val entityChunksStream: KStream> = + when (layer) { + ServerMapTileLayer.TERRAIN -> return + + ServerMapTileLayer.RESOURCE -> + builder.streamPacketData() + .convertResourceUpdateToServerMapChunkTiles("groupEntitiesMapChunks.${layer.dir}") + + ServerMapTileLayer.BUILDING -> + builder.streamPacketData() + .convertEntityUpdateToServerMapChunkTiles("groupEntitiesMapChunks.${layer.dir}") + } + + val protosStream: KStream = + builder.streamPacketData() + + val topology = colourEntityChunks( + builder, + entityChunksStream, + protosStream, + layer, + colouredUpdatesStreamTopic, + colouredChunkStateTopic, + ) + + launchTopology("groupTilesMapChunks.${layer.dir}", topology) +} diff --git a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupTilesMapChunks.kt b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupTilesMapChunks.kt new file mode 100644 index 0000000..fd65020 --- /dev/null +++ b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/groupTilesMapChunks.kt @@ -0,0 +1,23 @@ +package dev.adamko.kafkatorio.processors + +import dev.adamko.kafkatorio.processor.core.launchTopology +import dev.adamko.kafkatorio.schema.common.FactorioServerId +import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer +import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate +import dev.adamko.kafkatorio.server.processor.topology.colourMapChunks +import dev.adamko.kafkatorio.server.processor.topology.streamPacketData +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.kstream.KStream + + +/** Terrain tiles */ +suspend fun groupTilesMapChunks() { + val builder = StreamsBuilder() + + val protosStream: KStream = + builder.streamPacketData() + + val topology = colourMapChunks(builder, protosStream) + + launchTopology("groupTilesMapChunks.${ServerMapTileLayer.TERRAIN.dir}", topology) +} diff --git a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/main.kt b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/main.kt new file mode 100644 index 0000000..427903f --- /dev/null +++ b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/main.kt @@ -0,0 +1,36 @@ +package dev.adamko.kafkatorio.processors + +import dev.adamko.kafkatorio.processor.config.TOPIC_MAP_CHUNK_BUILDING_COLOURED_032_UPDATES +import dev.adamko.kafkatorio.processor.config.TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE +import dev.adamko.kafkatorio.processor.config.TOPIC_MAP_CHUNK_RESOURCE_COLOURED_032_UPDATES +import dev.adamko.kafkatorio.processor.config.TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE +import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch + + +suspend fun main(): Unit = coroutineScope { + launch { + splitPackets() + } + + launch { + groupTilesMapChunks() + } + + launch { + groupEntitiesMapChunks( + layer = ServerMapTileLayer.BUILDING, + colouredUpdatesStreamTopic = TOPIC_MAP_CHUNK_BUILDING_COLOURED_032_UPDATES, + colouredChunkStateTopic = TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE, + ) + } + + launch { + groupEntitiesMapChunks( + layer = ServerMapTileLayer.RESOURCE, + colouredUpdatesStreamTopic = TOPIC_MAP_CHUNK_RESOURCE_COLOURED_032_UPDATES, + colouredChunkStateTopic = TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE, + ) + } +} diff --git a/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/splitPackets.kt b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/splitPackets.kt new file mode 100644 index 0000000..9373bc2 --- /dev/null +++ b/modules/events-processors/src/main/kotlin/dev/adamko/kafkatorio/processors/splitPackets.kt @@ -0,0 +1,17 @@ +package dev.adamko.kafkatorio.processors + +import dev.adamko.kafkatorio.processor.core.launchTopology +import dev.adamko.kafkatorio.server.processor.topology.factorioServerPacketStream +import dev.adamko.kafkatorio.server.processor.topology.splitFactorioServerPacketStream +import org.apache.kafka.streams.StreamsBuilder + + +suspend fun splitPackets() { + val builder = StreamsBuilder() + val packets = factorioServerPacketStream(builder) + splitFactorioServerPacketStream(packets) + + val topology = builder.build() + + launchTopology("splitPackets", topology) +} diff --git a/modules/events-server-core/build.gradle.kts b/modules/events-server-core/build.gradle.kts deleted file mode 100644 index 1172be0..0000000 --- a/modules/events-server-core/build.gradle.kts +++ /dev/null @@ -1,79 +0,0 @@ -import org.jetbrains.kotlin.gradle.tasks.KotlinCompile - - -plugins { - dev.adamko.kafkatorio.lang.`kotlin-jvm` - kotlin("plugin.serialization") - application -} - - -//description = "serves Kafkatorio content over the web" - -val projectId: String by project.extra - - -dependencies { - implementation(platform(projects.modules.versionsPlatform)) - - implementation(libs.kafka.streams) - testImplementation(libs.kafka.streamsTestUtils) - - implementation(projects.modules.eventsLibrary) - - implementation(libs.kotlinxSerialization.core) - implementation(libs.kotlinxSerialization.json) - - implementation(libs.kotlinx.coroutines.core) - - implementation(libs.bundles.logging) - - implementation(libs.scrimage.core) - - implementation(libs.okio.core) - - // - implementation(libs.ktorSerialization.kotlinxJson) - - with(libs.ktorServer) { - implementation(auth) - implementation(authJwt) - implementation(core) - implementation(callId) - implementation(callLogging) - implementation(contentNegotiaion) - implementation(cio) - implementation(autoHeadResponse) - implementation(partialContent) - implementation(resources) - implementation(statusPages) - implementation(websockets) - implementation(compression) - } - - with(libs.ktorNetwork) { - implementation(core) - implementation(tls) - } - // - - implementation(libs.simpleSyslogParser) - -// implementation(libs.nimbusJoseJwt) - - implementation(libs.kafka.kotkaStreams) -} - - -application { - mainClass.set("dev.adamko.kafkatorio.server.EventsServerKt") -} - - -tasks.withType().configureEach { - kotlinOptions.freeCompilerArgs += listOf( - "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", - "-opt-in=kotlinx.coroutines.FlowPreview", - "-opt-in=kotlinx.serialization.ExperimentalSerializationApi", - ) -} diff --git a/modules/events-server-syslog/build.gradle.kts b/modules/events-server-syslog/build.gradle.kts index 1172be0..d746d17 100644 --- a/modules/events-server-syslog/build.gradle.kts +++ b/modules/events-server-syslog/build.gradle.kts @@ -8,7 +8,7 @@ plugins { } -//description = "serves Kafkatorio content over the web" +description = "Consumes Syslog Kafkatorio packets and sends them to Kafka" val projectId: String by project.extra @@ -20,6 +20,7 @@ dependencies { testImplementation(libs.kafka.streamsTestUtils) implementation(projects.modules.eventsLibrary) + implementation(projects.modules.eventsProcessorCore) implementation(libs.kotlinxSerialization.core) implementation(libs.kotlinxSerialization.json) @@ -62,11 +63,13 @@ dependencies { // implementation(libs.nimbusJoseJwt) implementation(libs.kafka.kotkaStreams) + + implementation(libs.bundles.hoplite) } application { - mainClass.set("dev.adamko.kafkatorio.server.EventsServerKt") + mainClass.set("dev.adamko.kafkatorio.processor.syslog.SyslogMainKt") } @@ -77,3 +80,9 @@ tasks.withType().configureEach { "-opt-in=kotlinx.serialization.ExperimentalSerializationApi", ) } + + +val runSyslogServer by tasks.registering { + group = rootProject.name + dependsOn(tasks.run) +} diff --git a/modules/events-server-syslog/docker/Dockerfile b/modules/events-server-syslog/docker/Dockerfile new file mode 100644 index 0000000..a196e24 --- /dev/null +++ b/modules/events-server-syslog/docker/Dockerfile @@ -0,0 +1,11 @@ +FROM gradle:7-jdk11 AS build +COPY --chown=gradle:gradle . /home/gradle/src +WORKDIR /home/gradle/src +RUN gradle buildFatJar --no-daemon + + +FROM openjdk:11 +EXPOSE 8080:8080 +RUN mkdir /app +COPY --from=build /home/gradle/src/build/libs/*.jar /app/ktor-docker-sample.jar +ENTRYPOINT ["java","-jar","/app/ktor-docker-sample.jar"] diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/KafkatorioPacketKafkaProducer.kt b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/KafkatorioPacketKafkaProducer.kt similarity index 92% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/KafkatorioPacketKafkaProducer.kt rename to modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/KafkatorioPacketKafkaProducer.kt index 1b0cfa3..d60a436 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/KafkatorioPacketKafkaProducer.kt +++ b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/KafkatorioPacketKafkaProducer.kt @@ -1,12 +1,12 @@ -package dev.adamko.kafkatorio.server.processor +package dev.adamko.kafkatorio.processor.syslog import com.github.palindromicity.syslog.dsl.SyslogFieldKeys +import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import dev.adamko.kafkatorio.processor.core.Authenticator +import dev.adamko.kafkatorio.processor.syslog.config.jsonMapper import dev.adamko.kafkatorio.schema.common.FactorioServerId -import dev.adamko.kafkatorio.server.config.ApplicationProperties -import dev.adamko.kafkatorio.server.config.jsonMapper -import dev.adamko.kafkatorio.server.security.Authenticator -import dev.adamko.kafkatorio.server.socket.SyslogMsg -import dev.adamko.kafkatorio.server.socket.SyslogSocketServer +import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG +import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG_DLQ import dev.adamko.kotka.kxs.kafkaSerializer import java.util.Properties import kotlin.coroutines.coroutineContext diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/socket/SyslogMsg.kt b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogMsg.kt similarity index 96% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/socket/SyslogMsg.kt rename to modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogMsg.kt index d329352..42bb7ea 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/socket/SyslogMsg.kt +++ b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogMsg.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.server.socket +package dev.adamko.kafkatorio.processor.syslog import com.github.palindromicity.syslog.dsl.SyslogFieldKeys diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/socket/SyslogSocketServer.kt b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogSocketServer.kt similarity index 96% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/socket/SyslogSocketServer.kt rename to modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogSocketServer.kt index 666d477..12cac26 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/socket/SyslogSocketServer.kt +++ b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogSocketServer.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.server.socket +package dev.adamko.kafkatorio.processor.syslog import com.github.palindromicity.syslog.AllowableDeviations import com.github.palindromicity.syslog.DefaultKeyProvider @@ -7,7 +7,7 @@ import com.github.palindromicity.syslog.NilPolicy import com.github.palindromicity.syslog.SyslogParser import com.github.palindromicity.syslog.SyslogParserBuilder import com.github.palindromicity.syslog.SyslogSpecification -import dev.adamko.kafkatorio.server.config.ApplicationProperties +import dev.adamko.kafkatorio.processor.config.ApplicationProperties import io.ktor.network.selector.SelectorManager import io.ktor.network.sockets.Socket import io.ktor.network.sockets.aSocket diff --git a/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/config/kxsJson.kt b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/config/kxsJson.kt new file mode 100644 index 0000000..6b8797b --- /dev/null +++ b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/config/kxsJson.kt @@ -0,0 +1,12 @@ +package dev.adamko.kafkatorio.processor.syslog.config + +import dev.adamko.kafkatorio.library.jsonMapperKafkatorio +import kotlinx.serialization.json.Json + + +@PublishedApi +internal val jsonMapper: Json = Json { + prettyPrint = true + prettyPrintIndent = " " + serializersModule = jsonMapperKafkatorio.serializersModule +} diff --git a/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/syslogMain.kt b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/syslogMain.kt new file mode 100644 index 0000000..6667561 --- /dev/null +++ b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/syslogMain.kt @@ -0,0 +1,20 @@ +package dev.adamko.kafkatorio.processor.syslog + +import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import dev.adamko.kafkatorio.processor.core.Authenticator +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch + + +/** Receive log messages from the syslog socket, and send them to a Kafka topic */ +suspend fun main(): Unit = coroutineScope { + + val appProps = ApplicationProperties.load() + val syslogServer = SyslogSocketServer(appProps) + val authenticator = Authenticator(appProps) + + val producer = KafkatorioPacketKafkaProducer(appProps, syslogServer, authenticator) + + launch { syslogServer.start() } + launch { producer.launch() } +} diff --git a/modules/events-server-api/build.gradle.kts b/modules/events-server-web/build.gradle.kts similarity index 81% rename from modules/events-server-api/build.gradle.kts rename to modules/events-server-web/build.gradle.kts index 63de5d8..f65b0ae 100644 --- a/modules/events-server-api/build.gradle.kts +++ b/modules/events-server-web/build.gradle.kts @@ -8,11 +8,11 @@ plugins { } -//description = "serves Kafkatorio content over the web" +description = "serves Kafkatorio content over web APIs" val projectId: String by project.extra -val kafkaStateDir: Directory = layout.projectDirectory.dir("kafka-state") +val kafkaStateDir: Directory = layout.projectDirectory.dir(".state/kafka") dependencies { @@ -22,6 +22,7 @@ dependencies { testImplementation(libs.kafka.streamsTestUtils) implementation(projects.modules.eventsLibrary) + implementation(projects.modules.eventsProcessorCore) implementation(libs.kotlinxSerialization.core) implementation(libs.kotlinxSerialization.json) @@ -64,11 +65,13 @@ dependencies { // implementation(libs.nimbusJoseJwt) implementation(libs.kafka.kotkaStreams) + + implementation(libs.bundles.hoplite) } application { - mainClass.set("dev.adamko.kafkatorio.server.EventsServerKt") + mainClass.set("dev.adamko.kafkatorio.server.web.MainKt") } @@ -79,3 +82,10 @@ tasks.withType().configureEach { "-opt-in=kotlinx.serialization.ExperimentalSerializationApi", ) } + + +val runEventsWebServer by tasks.registering { + group = rootProject.name + + dependsOn(tasks.run) +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/kxsJson.kt b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/config/kxsJson.kt similarity index 70% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/kxsJson.kt rename to modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/config/kxsJson.kt index 6dab1ed..da63668 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/kxsJson.kt +++ b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/config/kxsJson.kt @@ -1,10 +1,10 @@ -package dev.adamko.kafkatorio.server.config +package dev.adamko.kafkatorio.server.web.config import dev.adamko.kafkatorio.library.jsonMapperKafkatorio import kotlinx.serialization.json.Json -val jsonMapper: Json = Json { +internal val jsonMapper: Json = Json { prettyPrint = true prettyPrintIndent = " " serializersModule = jsonMapperKafkatorio.serializersModule diff --git a/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/main.kt b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/main.kt new file mode 100644 index 0000000..4ce5ae0 --- /dev/null +++ b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/main.kt @@ -0,0 +1,61 @@ +package dev.adamko.kafkatorio.server.web + +import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import dev.adamko.kafkatorio.processor.core.launchTopology +import dev.adamko.kafkatorio.server.processor.topology.broadcastToWebsocket +import dev.adamko.kafkatorio.server.processor.topology.saveMapTiles +import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer +import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.Duration.Companion.seconds +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch +import org.apache.kafka.clients.producer.ProducerConfig +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.StreamsConfig + +suspend fun main(): Unit = coroutineScope { + val appProps = ApplicationProperties.load() + val wsServer = WebmapWebsocketServer() + + launch { + websocketBroadcaster(wsServer) + } + +// launch { +// broadcastToWebsocket(wsServer, StreamsBuilder()) +// } + + launch { + saveTiles(appProps) + } + + launch { + startWebServer( + appProps = appProps, + wsServer = wsServer, + ) + } +} + +private suspend fun websocketBroadcaster(wsServer: WebmapWebsocketServer) { + val builder = StreamsBuilder() + + broadcastToWebsocket(wsServer, builder) + + val topology = builder.build() + + launchTopology( + "websocketBroadcaster", topology, mapOf( + StreamsConfig.COMMIT_INTERVAL_MS_CONFIG to "${1.seconds.inWholeMilliseconds}", + StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG) to "${1.milliseconds.inWholeMilliseconds}" + ) + ) +} + +private suspend fun saveTiles( + appProps: ApplicationProperties +) { + val builder = StreamsBuilder() + saveMapTiles(builder, appProps.serverDataDir) + launchTopology("saveTiles", builder.build()) +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/factorioServersRoutes.kt b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/factorioServersRoutes.kt similarity index 95% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/factorioServersRoutes.kt rename to modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/factorioServersRoutes.kt index 3925a33..082fcde 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/factorioServersRoutes.kt +++ b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/factorioServersRoutes.kt @@ -5,8 +5,8 @@ import dev.adamko.kafkatorio.schema.common.PlayerIndex import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer import dev.adamko.kafkatorio.schema.common.ServerMapTilePngFilename import dev.adamko.kafkatorio.schema.common.SurfaceIndex -import dev.adamko.kafkatorio.server.config.ApplicationProperties -import dev.adamko.kafkatorio.server.config.jsonMapper +import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import dev.adamko.kafkatorio.server.web.config.jsonMapper import io.ktor.http.HttpStatusCode import io.ktor.http.content.EntityTagVersion import io.ktor.http.content.versions @@ -25,7 +25,7 @@ import kotlinx.serialization.Serializable import kotlinx.serialization.json.encodeToJsonElement -fun Routing.factorioServersRoutes( +internal fun Routing.factorioServersRoutes( appProps: ApplicationProperties, ) { diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/restApi.kt b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/restApi.kt similarity index 92% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/restApi.kt rename to modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/restApi.kt index 0fe7f45..22f8e3e 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/restApi.kt +++ b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/restApi.kt @@ -1,6 +1,6 @@ package dev.adamko.kafkatorio.server.web.rest -import dev.adamko.kafkatorio.server.config.ApplicationProperties +import dev.adamko.kafkatorio.processor.config.ApplicationProperties import io.ktor.http.ContentType import io.ktor.http.HttpStatusCode import io.ktor.http.isSuccess @@ -22,7 +22,7 @@ import io.ktor.server.response.respondText import io.ktor.server.routing.routing -fun Application.webmapTileServer(appProps: ApplicationProperties) { +internal fun Application.webmapTileServer(appProps: ApplicationProperties) { install(AutoHeadResponse) install(CallLogging) { diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/webServer.kt b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/webServer.kt similarity index 85% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/webServer.kt rename to modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/webServer.kt index 4fe626e..6792e02 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/webServer.kt +++ b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/webServer.kt @@ -1,6 +1,7 @@ package dev.adamko.kafkatorio.server.web -import dev.adamko.kafkatorio.server.config.ApplicationProperties + +import dev.adamko.kafkatorio.processor.config.ApplicationProperties import dev.adamko.kafkatorio.server.web.rest.webmapTileServer import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer import dev.adamko.kafkatorio.server.web.websocket.webmapWebsocketServer @@ -9,7 +10,7 @@ import io.ktor.server.engine.embeddedServer import kotlinx.coroutines.coroutineScope -suspend fun startWebServer( +internal suspend fun startWebServer( appProps: ApplicationProperties, wsServer: WebmapWebsocketServer, ): Unit = coroutineScope { diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/WebmapWebsocketServer.kt b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/WebmapWebsocketServer.kt similarity index 92% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/WebmapWebsocketServer.kt rename to modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/WebmapWebsocketServer.kt index e672196..c704411 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/WebmapWebsocketServer.kt +++ b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/WebmapWebsocketServer.kt @@ -1,5 +1,6 @@ package dev.adamko.kafkatorio.server.web.websocket +import dev.adamko.kafkatorio.server.processor.topology.WebsocketBroadcaster import io.ktor.websocket.CloseReason import io.ktor.websocket.DefaultWebSocketSession import io.ktor.websocket.Frame @@ -24,7 +25,7 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.yield -class WebmapWebsocketServer { +class WebmapWebsocketServer : WebsocketBroadcaster { private val startTime = TimeSource.Monotonic.markNow() @@ -88,9 +89,9 @@ class WebmapWebsocketServer { } - suspend fun sendMessageToClients(msg: String) { - log("sending message to clients: $msg") - messagesToClients.emit(msg) + override suspend fun sendMessageToClients(message: String) { + log("sending message to clients: $message") + messagesToClients.emit(message) } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/wsServer.kt b/modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/wsServer.kt similarity index 100% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/wsServer.kt rename to modules/events-server-web/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/wsServer.kt diff --git a/modules/events-server/build.gradle.kts b/modules/events-server/build.gradle.kts index 8613f5c..c865d61 100644 --- a/modules/events-server/build.gradle.kts +++ b/modules/events-server/build.gradle.kts @@ -10,11 +10,11 @@ plugins { } -description = "serves Kafkatorio content over the web" +description = "deprecated - this was split up into separate modules" val projectId: String by project.extra -val kafkaStateDir: Directory = layout.projectDirectory.dir("kafka-state") +val kafkaStateDir: Directory = layout.projectDirectory.dir(".state/kafka") dependencies { diff --git a/modules/events-server/docker/Dockerfile b/modules/events-server/docker/Dockerfile index 48e4647..3f7ac14 100644 --- a/modules/events-server/docker/Dockerfile +++ b/modules/events-server/docker/Dockerfile @@ -13,7 +13,7 @@ ENV SOCKET_SERVER_PORT=9002 EXPOSE $SOCKET_SERVER_PORT ENV SERVER_DATA_DIR="/app/data" -ENV KAFKA_STREAMS_STATE_DIR="/app/kafka-state" +ENV KAFKA_STREAMS_STATE_DIR="/app/state/kafka" VOLUME ${SERVER_DATA_DIR} \ ${KAFKA_STREAMS_STATE_DIR} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/eventsServer.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/eventsServer.kt index ab6d89a..bc3c1d6 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/eventsServer.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/eventsServer.kt @@ -1,32 +1,32 @@ package dev.adamko.kafkatorio.server -import dev.adamko.kafkatorio.server.config.ApplicationProperties -import dev.adamko.kafkatorio.server.processor.eventsProcessor -import dev.adamko.kafkatorio.server.security.Authenticator -import dev.adamko.kafkatorio.server.socket.SyslogSocketServer -import dev.adamko.kafkatorio.server.web.startWebServer -import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer -import kotlinx.coroutines.coroutineScope -import kotlinx.coroutines.launch +//import dev.adamko.kafkatorio.server.config.ApplicationProperties +//import dev.adamko.kafkatorio.server.processor.eventsProcessor +//import dev.adamko.kafkatorio.server.security.Authenticator +//import dev.adamko.kafkatorio.server.socket.SyslogSocketServer +//import dev.adamko.kafkatorio.server.web.startWebServer +//import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer +//import kotlinx.coroutines.coroutineScope +//import kotlinx.coroutines.launch -suspend fun main(): Unit = coroutineScope { - - val appProps = ApplicationProperties.load() - - val websocketServer = WebmapWebsocketServer() - val syslogSocketServer = SyslogSocketServer(appProps) - val authenticator = Authenticator(appProps) - - launch { - startWebServer(appProps, websocketServer) - } - - launch { - syslogSocketServer.start() - } - - launch { - eventsProcessor(appProps, syslogSocketServer, websocketServer, authenticator) - } -} +//suspend fun main(): Unit = coroutineScope { +// +// val appProps = ApplicationProperties.load() +// +// val websocketServer = WebmapWebsocketServer() +// val syslogSocketServer = SyslogSocketServer(appProps) +// val authenticator = Authenticator(appProps) +// +// launch { +// startWebServer(appProps, websocketServer) +// } +// +// launch { +// syslogSocketServer.start() +// } +// +// launch { +// eventsProcessor(appProps, syslogSocketServer, websocketServer, authenticator) +// } +//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/KafkatorioTopology.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/KafkatorioTopology.kt index 56c8e25..abdc507 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/KafkatorioTopology.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/KafkatorioTopology.kt @@ -1,243 +1,243 @@ package dev.adamko.kafkatorio.server.processor -import dev.adamko.kafkatorio.schema.common.FactorioServerId -import dev.adamko.kafkatorio.schema.common.PrototypeHashCode -import dev.adamko.kafkatorio.schema.common.ServerMapChunkId -import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer -import dev.adamko.kafkatorio.schema.packets.MapChunkEntityUpdate -import dev.adamko.kafkatorio.schema.packets.MapChunkResourceUpdate -import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate -import dev.adamko.kafkatorio.server.config.ApplicationProperties -import dev.adamko.kafkatorio.server.processor.topology.ServerMapChunkTiles -import dev.adamko.kafkatorio.server.processor.topology.broadcastToWebsocket -import dev.adamko.kafkatorio.server.processor.topology.colourEntityChunks -import dev.adamko.kafkatorio.server.processor.topology.colourMapChunks -import dev.adamko.kafkatorio.server.processor.topology.convertEntityUpdateToServerMapChunkTiles -import dev.adamko.kafkatorio.server.processor.topology.convertResourceUpdateToServerMapChunkTiles -import dev.adamko.kafkatorio.server.processor.topology.factorioServerPacketStream -import dev.adamko.kafkatorio.server.processor.topology.saveMapTiles -import dev.adamko.kafkatorio.server.processor.topology.splitFactorioServerPacketStream -import dev.adamko.kafkatorio.server.processor.topology.streamPacketData -import dev.adamko.kafkatorio.server.socket.SyslogSocketServer -import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer -import java.util.Properties -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.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 -import org.apache.kafka.streams.Topology -import org.apache.kafka.streams.TopologyDescription -import org.apache.kafka.streams.kstream.KStream +//import dev.adamko.kafkatorio.schema.common.FactorioServerId +//import dev.adamko.kafkatorio.schema.common.PrototypeHashCode +//import dev.adamko.kafkatorio.schema.common.ServerMapChunkId +//import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer +//import dev.adamko.kafkatorio.schema.packets.MapChunkEntityUpdate +//import dev.adamko.kafkatorio.schema.packets.MapChunkResourceUpdate +//import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate +//import dev.adamko.kafkatorio.server.config.ApplicationProperties +//import dev.adamko.kafkatorio.server.processor.topology.ServerMapChunkTiles +//import dev.adamko.kafkatorio.server.processor.topology.broadcastToWebsocket +//import dev.adamko.kafkatorio.server.processor.topology.colourEntityChunks +//import dev.adamko.kafkatorio.server.processor.topology.colourMapChunks +//import dev.adamko.kafkatorio.server.processor.topology.convertEntityUpdateToServerMapChunkTiles +//import dev.adamko.kafkatorio.server.processor.topology.convertResourceUpdateToServerMapChunkTiles +//import dev.adamko.kafkatorio.server.processor.topology.factorioServerPacketStream +//import dev.adamko.kafkatorio.server.processor.topology.saveMapTiles +//import dev.adamko.kafkatorio.server.processor.topology.splitFactorioServerPacketStream +//import dev.adamko.kafkatorio.server.processor.topology.streamPacketData +//import dev.adamko.kafkatorio.server.socket.SyslogSocketServer +//import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer +//import java.util.Properties +//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.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 +//import org.apache.kafka.streams.Topology +//import org.apache.kafka.streams.TopologyDescription +//import org.apache.kafka.streams.kstream.KStream internal class KafkatorioTopology( - private val appProps: ApplicationProperties, - private val wsServer: WebmapWebsocketServer, - private val syslogServer: SyslogSocketServer, - private val kafkatorioPacketKafkaProducer: KafkatorioPacketKafkaProducer, +// private val appProps: ApplicationProperties, +// private val wsServer: WebmapWebsocketServer, +// private val syslogServer: SyslogSocketServer, +// private val kafkatorioPacketKafkaProducer: KafkatorioPacketKafkaProducer, ) { - private val builderScope = CoroutineScope(CoroutineName("KafkaTopology")) +// private val builderScope = CoroutineScope(CoroutineName("KafkaTopology")) - suspend fun start() = builderScope.launch { - splitPackets() - - groupTilesMapChunks() - - groupEntitiesMapChunks( - layer = ServerMapTileLayer.BUILDING, - colouredUpdatesStreamTopic = TOPIC_MAP_CHUNK_BUILDING_COLOURED_032_UPDATES, - colouredChunkStateTopic = TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE, - ) - - groupEntitiesMapChunks( - layer = ServerMapTileLayer.RESOURCE, - colouredUpdatesStreamTopic = TOPIC_MAP_CHUNK_RESOURCE_COLOURED_032_UPDATES, - colouredChunkStateTopic = TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE, - ) - - saveTiles() - - websocketBroadcaster() - - syslogProducer() - } +// suspend fun start() = builderScope.launch { +// splitPackets() +// +// groupTilesMapChunks() +// +//// groupEntitiesMapChunks( +//// layer = ServerMapTileLayer.BUILDING, +//// colouredUpdatesStreamTopic = TOPIC_MAP_CHUNK_BUILDING_COLOURED_032_UPDATES, +//// colouredChunkStateTopic = TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE, +//// ) +// +//// groupEntitiesMapChunks( +//// layer = ServerMapTileLayer.RESOURCE, +//// colouredUpdatesStreamTopic = TOPIC_MAP_CHUNK_RESOURCE_COLOURED_032_UPDATES, +//// colouredChunkStateTopic = TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE, +//// ) +// +// saveTiles() +// +// websocketBroadcaster() +// +// syslogProducer() +// } - private suspend fun splitPackets() { - val builder = StreamsBuilder() - val packets = factorioServerPacketStream(builder) - splitFactorioServerPacketStream(packets) - - val topology = builder.build() - - launchTopology("splitPackets", topology) - } +// private suspend fun splitPackets() { +// val builder = StreamsBuilder() +// val packets = factorioServerPacketStream(builder) +// splitFactorioServerPacketStream(packets) +// +// val topology = builder.build() +// +// launchTopology("splitPackets", topology) +// } - /** Terrain tiles */ - private suspend fun groupTilesMapChunks() { - val builder = StreamsBuilder() +// /** Terrain tiles */ +// private suspend fun groupTilesMapChunks() { +// val builder = StreamsBuilder() +// +// val protosStream: KStream = +// builder.streamPacketData() +// +// val topology = colourMapChunks(builder, protosStream) +// +// launchTopology("groupTilesMapChunks.${ServerMapTileLayer.TERRAIN.dir}", topology) +// } - val protosStream: KStream = - builder.streamPacketData() - - val topology = colourMapChunks(builder, protosStream) - - launchTopology("groupTilesMapChunks.${ServerMapTileLayer.TERRAIN.dir}", topology) - } +// +// /** Building and resource tiles */ +// private suspend fun groupEntitiesMapChunks( +// layer: ServerMapTileLayer, +// colouredUpdatesStreamTopic: String, +// colouredChunkStateTopic: String, +// ) { +// val builder = StreamsBuilder() +// +// val entityChunksStream: KStream> = +// when (layer) { +// ServerMapTileLayer.TERRAIN -> return +// +// ServerMapTileLayer.RESOURCE -> +// builder.streamPacketData() +// .convertResourceUpdateToServerMapChunkTiles("groupEntitiesMapChunks.${layer.dir}") +// +// ServerMapTileLayer.BUILDING -> +// builder.streamPacketData() +// .convertEntityUpdateToServerMapChunkTiles("groupEntitiesMapChunks.${layer.dir}") +// } +// +// val protosStream: KStream = +// builder.streamPacketData() +// +// val topology = colourEntityChunks( +// builder, +// entityChunksStream, +// protosStream, +// layer, +// colouredUpdatesStreamTopic, +// colouredChunkStateTopic, +// ) +// +// launchTopology("groupTilesMapChunks.${layer.dir}", topology) +// } - /** Building and resource tiles */ - private suspend fun groupEntitiesMapChunks( - layer: ServerMapTileLayer, - colouredUpdatesStreamTopic: String, - colouredChunkStateTopic: String, - ) { - val builder = StreamsBuilder() - - val entityChunksStream: KStream> = - when (layer) { - ServerMapTileLayer.TERRAIN -> return - - ServerMapTileLayer.RESOURCE -> - builder.streamPacketData() - .convertResourceUpdateToServerMapChunkTiles("groupEntitiesMapChunks.${layer.dir}") - - ServerMapTileLayer.BUILDING -> - builder.streamPacketData() - .convertEntityUpdateToServerMapChunkTiles("groupEntitiesMapChunks.${layer.dir}") - } - - val protosStream: KStream = - builder.streamPacketData() - - val topology = colourEntityChunks( - builder, - entityChunksStream, - protosStream, - layer, - colouredUpdatesStreamTopic, - colouredChunkStateTopic, - ) - - launchTopology("groupTilesMapChunks.${layer.dir}", topology) - } +// private suspend fun saveTiles() { +// val builder = StreamsBuilder() +// val topology = saveMapTiles(builder, appProps.serverDataDir) +// launchTopology("saveTiles", topology) +// } - private suspend fun saveTiles() { - val builder = StreamsBuilder() - val topology = saveMapTiles(builder, appProps.serverDataDir) - launchTopology("saveTiles", topology) - } +// private suspend fun websocketBroadcaster() { +// val builder = StreamsBuilder() +// broadcastToWebsocket(wsServer, builder) +// +// val topology = builder.build() +// +// launchTopology( +// "websocketBroadcaster", topology, mapOf( +// StreamsConfig.COMMIT_INTERVAL_MS_CONFIG to "${1.seconds.inWholeMilliseconds}", +// StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG) to "${1.milliseconds.inWholeMilliseconds}" +// ) +// ) +// } - private suspend fun websocketBroadcaster() { - val builder = StreamsBuilder() - broadcastToWebsocket(wsServer, builder) - - val topology = builder.build() - - launchTopology( - "websocketBroadcaster", topology, mapOf( - StreamsConfig.COMMIT_INTERVAL_MS_CONFIG to "${1.seconds.inWholeMilliseconds}", - StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG) to "${1.milliseconds.inWholeMilliseconds}" - ) - ) - } +// /** Receive log messages from the syslog socket, and send them to a Kafka topic */ +// private fun syslogProducer() { +// builderScope.launch { +// kafkatorioPacketKafkaProducer.launch() +// } +// } - /** Receive log messages from the syslog socket, and send them to a Kafka topic */ - private fun syslogProducer() { - builderScope.launch { - kafkatorioPacketKafkaProducer.launch() - } - } +// private suspend fun launchTopology( +// id: String, +// topology: Topology, +// additionalProperties: Map = mapOf(), +// ) { +// +// val props: Properties = appProps.kafkaStreamsConfig.toProperties() +// +// additionalProperties.forEach { (k, v) -> props.setProperty(k, v) } +// +// val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as String +// props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId) +// +// 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 -> if (cont.isActive) 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 suspend fun launchTopology( - id: String, - topology: Topology, - additionalProperties: Map = mapOf(), - ) { - - val props: Properties = appProps.kafkaStreamsConfig.toProperties() - - additionalProperties.forEach { (k, v) -> props.setProperty(k, v) } - - val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as String - props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId) - - 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 -> if (cont.isActive) 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 { +// parentFile.mkdirs() +// if (!exists()) createNewFile() +// writeText(description.toString()) +// } +// +// println( +// """ +// |---------------- +// |$appId - ${descFile.canonicalPath} +// |$description +// |---------------- +// """.trimMargin() +// ) +// } - private fun printTopologyDescription(appId: String, topology: Topology) { - val description: TopologyDescription = topology.describe() - - val descFile = Path("./build/$appId.txt").toFile().apply { - parentFile.mkdirs() - if (!exists()) createNewFile() - writeText(description.toString()) - } - - println( - """ - |---------------- - |$appId - ${descFile.canonicalPath} - |$description - |---------------- - """.trimMargin() - ) - } - - - companion object { - private fun Map.toProperties(): Properties = - entries.fold(Properties()) { props, (k, v) -> - props.apply { setProperty(k, v) } - } - } +// companion object { +// 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/server/processor/eventProcessor.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/eventProcessor.kt index 9c80d9c..7e0b2a3 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/eventProcessor.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/eventProcessor.kt @@ -1,35 +1,35 @@ package dev.adamko.kafkatorio.server.processor -import dev.adamko.kafkatorio.server.config.ApplicationProperties -import dev.adamko.kafkatorio.server.processor.admin.KafkatorioKafkaAdmin -import dev.adamko.kafkatorio.server.security.Authenticator -import dev.adamko.kafkatorio.server.socket.SyslogSocketServer -import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer -import kotlinx.coroutines.awaitCancellation -import kotlinx.coroutines.coroutineScope -import kotlinx.coroutines.launch - - -suspend fun eventsProcessor( - appProps: ApplicationProperties, - syslogServer: SyslogSocketServer, - wsServer: WebmapWebsocketServer, - authenticator: Authenticator, -): Unit = coroutineScope { - - val admin = KafkatorioKafkaAdmin(appProps) - admin.createKafkatorioTopics() - - val kafkatorioPacketKafkaProducer = KafkatorioPacketKafkaProducer(appProps, syslogServer, authenticator) - - val topology = KafkatorioTopology(appProps, wsServer, syslogServer, kafkatorioPacketKafkaProducer) - - launch { - topology.start() - - println("launched KafkatorioTopology") - - awaitCancellation() - } -} +//import dev.adamko.kafkatorio.server.config.ApplicationProperties +//import dev.adamko.kafkatorio.server.processor.admin.KafkatorioKafkaAdmin +//import dev.adamko.kafkatorio.server.security.Authenticator +//import dev.adamko.kafkatorio.server.socket.SyslogSocketServer +//import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer +//import kotlinx.coroutines.awaitCancellation +//import kotlinx.coroutines.coroutineScope +//import kotlinx.coroutines.launch +// +// +//suspend fun eventsProcessor( +// appProps: ApplicationProperties, +// syslogServer: SyslogSocketServer, +// wsServer: WebmapWebsocketServer, +// authenticator: Authenticator, +//): Unit = coroutineScope { +// +// val admin = KafkatorioKafkaAdmin(appProps) +// admin.createKafkatorioTopics() +// +// val kafkatorioPacketKafkaProducer = KafkatorioPacketKafkaProducer(appProps, syslogServer, authenticator) +// +// val topology = KafkatorioTopology(appProps, wsServer, syslogServer, kafkatorioPacketKafkaProducer) +// +// launch { +// topology.start() +// +// println("launched KafkatorioTopology") +// +// awaitCancellation() +// } +//} diff --git a/modules/infra-factorio-client/build.gradle.kts b/modules/infra-factorio-client/build.gradle.kts index 5ecf3ec..4c694a9 100644 --- a/modules/infra-factorio-client/build.gradle.kts +++ b/modules/infra-factorio-client/build.gradle.kts @@ -109,3 +109,10 @@ tasks.processRun { tasks.processKill { dependsOn(clientKill) } + + +val runFactorioClient by tasks.registering { + group = rootProject.name + + dependsOn(tasks.processRun) +} diff --git a/modules/infra-factorio-server/build.gradle.kts b/modules/infra-factorio-server/build.gradle.kts index 4380489..a97e76b 100644 --- a/modules/infra-factorio-server/build.gradle.kts +++ b/modules/infra-factorio-server/build.gradle.kts @@ -35,7 +35,7 @@ val deployModToLocalServer by tasks.registering(Copy::class) { into(factorioServerDataDir.dir("mods")) doLast { - logger.lifecycle("Copying mods ${source.files} to $destinationDir") + logger.lifecycle("Copying mods ${source.files} into $destinationDir") } } @@ -75,3 +75,9 @@ idea { excludeDirs.add(file("src/factorio-server")) } } + +val runFactorioServer by tasks.registering { + group = rootProject.name + + dependsOn(tasks.processRun) +} diff --git a/modules/infra-factorio-server/src/docker-compose.yml b/modules/infra-factorio-server/src/docker-compose.yml index 499e029..df48a83 100644 --- a/modules/infra-factorio-server/src/docker-compose.yml +++ b/modules/infra-factorio-server/src/docker-compose.yml @@ -20,7 +20,7 @@ services: driver: syslog options: syslog-address: "tcp://host.docker.internal:12034" # pc - #syslog-address: "tcp://localhost:12034" # nuc +# syslog-address: "tcp://localhost:12034" # nuc syslog-format: "rfc5424" tag: ${KAFKATORIO_TOKEN} diff --git a/modules/infra-kafka-cluster/build.gradle.kts b/modules/infra-kafka-cluster/build.gradle.kts index 9341a8e..e5f2040 100644 --- a/modules/infra-kafka-cluster/build.gradle.kts +++ b/modules/infra-kafka-cluster/build.gradle.kts @@ -27,3 +27,9 @@ tasks.dockerEnvUpdate { // "SCHEMA_REGISTRY_MEM_LIMIT" to "256m", ) } + +val runKafkaCluster by tasks.registering { + group = rootProject.name + + dependsOn(tasks.processRun) +} diff --git a/modules/infra-kafka-cluster/src/.env b/modules/infra-kafka-cluster/src/.env index d5c8942..2edf2d3 100644 --- a/modules/infra-kafka-cluster/src/.env +++ b/modules/infra-kafka-cluster/src/.env @@ -2,6 +2,6 @@ COMPOSE_PROJECT_NAME=kafkatorio GRAFANA_VERSION=8.5.0 KAFKATORIO_VERSION=0.9.0 -KAFKA_DL_URL=https\://dlcdn.apache.org/kafka/3.2.1/kafka_2.13-3.2.1.tgz -KAFKA_VERSION=3.2.1 +KAFKA_DL_URL=https\://dlcdn.apache.org/kafka/3.2.2/kafka_2.13-3.2.2.tgz +KAFKA_VERSION=3.2.2 PROMETHEUS_VERSION=v2.34.0 diff --git a/modules/modules.md b/modules/modules.md new file mode 100644 index 0000000..de6bd3d --- /dev/null +++ b/modules/modules.md @@ -0,0 +1,36 @@ +```mermaid +flowchart TD + +events-library-->events-processors & events-mod & web-map & events-server-web + +kafka[(Kafka)] + + +events-mod-->infra-factorio-client + + +subgraph Kafka Workers +events-processor-core --> events-processor-syslog & events-processors +end + +kafka<==>events-processors + +subgraph infra +subgraph factorio +infra-factorio-client<-->infra-factorio-server +end +infra-factorio-server-->infra-kafka-pipe +infra-kafka-pipe-->kafka +infra-kafka-cluster-->kafka +end + +kafka-->events-server-web + +events-server-web-->web-map + +versions-platform + +subgraph experiments +kt-rcon +end +``` diff --git a/modules/web-map/build.gradle.kts b/modules/web-map/build.gradle.kts index fc475cd..61c10d0 100644 --- a/modules/web-map/build.gradle.kts +++ b/modules/web-map/build.gradle.kts @@ -178,3 +178,10 @@ tasks.assemble { tasks.withType().configureEach { enabled = false } + + +val runWebMap by tasks.registering { + group = rootProject.name + + dependsOn(tasks.matching { it.name == "browserDevelopmentRun" }) +} diff --git a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/App.kt b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/App.kt index 532527b..f88e2ff 100644 --- a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/App.kt +++ b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/App.kt @@ -58,13 +58,13 @@ object App : Application() { require("./css/kafkatorio.css") - WebsocketService.packetsFlow - .filterIsInstance() - .onEach { tileSavedEvent -> -// println("[packetsFlow] triggering tile refresh ${tileSavedEvent.filename.value}") - siteStateStore.dispatch(SiteAction.EventServerUpdate(tileSavedEvent)) -// reduxStore.map.refreshUpdatedTilePng(tileSavedEvent.filename) - }.launchIn(coroutineScope) +// WebsocketService.packetsFlow +// .filterIsInstance() +// .onEach { tileSavedEvent -> +//// println("[packetsFlow] triggering tile refresh ${tileSavedEvent.filename.value}") +//// siteStateStore.dispatch(SiteAction.EventServerUpdate(tileSavedEvent)) +// siteStateStore.getState().factorioGameState?.map?.refreshUpdatedTilePng(tileSavedEvent.filename) +// }.launchIn(coroutineScope) } diff --git a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/FactorioMap.kt b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/FactorioMap.kt index e47c3fe..3407a67 100644 --- a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/FactorioMap.kt +++ b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/FactorioMap.kt @@ -3,8 +3,10 @@ package dev.adamko.kafkatorio.webmap import dev.adamko.kafkatorio.schema.common.FactorioServerId import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer import dev.adamko.kafkatorio.schema.common.ServerMapTilePngFilename +import dev.adamko.kafkatorio.schema.packets.EventServerPacket import dev.adamko.kafkatorio.webmap.externals.TileOnLoadFn import dev.adamko.kafkatorio.webmap.externals.tileOnLoad +import dev.adamko.kafkatorio.webmap.services.WebsocketService import io.kvision.maps.Maps import io.kvision.maps.Maps.Companion.L import io.kvision.maps.externals.leaflet.DoneCallback @@ -18,8 +20,15 @@ import io.kvision.maps.externals.leaflet.layer.tile.TileLayer import io.kvision.utils.px import kotlinx.browser.document import kotlinx.browser.window +import kotlinx.coroutines.CoroutineName +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.await import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.flow.filterIsInstance +import kotlinx.coroutines.flow.launchIn +import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch import kotlinx.js.jso import org.w3c.dom.HTMLElement @@ -36,6 +45,12 @@ class FactorioMap( private val serverId: FactorioServerId, ) { + private val coroutineScope: CoroutineScope = CoroutineScope( + Dispatchers.Default + + SupervisorJob(rootJob) + + CoroutineName("FactorioMap") + ) + val kvMap: Maps = Maps { minHeight = 800.px minWidth = 800.px @@ -90,6 +105,15 @@ class FactorioMap( playerIconsLayer.addTo(this) } + + + WebsocketService.packetsFlow + .filterIsInstance() + .onEach { tileSavedEvent -> +// println("[packetsFlow] triggering tile refresh ${tileSavedEvent.filename.value}") +// siteStateStore.dispatch(SiteAction.EventServerUpdate(tileSavedEvent)) + refreshUpdatedTilePng(tileSavedEvent.filename) + }.launchIn(coroutineScope) } diff --git a/settings.gradle.kts b/settings.gradle.kts index 13ece3e..e16d58a 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -1,18 +1,30 @@ rootProject.name = "kafkatorio" + +@Suppress("UnstableApiUsage") +pluginManagement { + includeBuild("./gradle-plugins/settings-plugins/") +} + + +includeBuild("./gradle-plugins/builder-plugins/") +includeBuild("./gradle-plugins/factorio-mod-manager/") + + include( ":modules:events-library", ":modules:events-mod", ":modules:events-processor-core", - ":modules:events-processor-entities", - ":modules:events-processor-packets", - ":modules:events-processor-tiles", +// ":modules:events-processor-entities", + ":modules:events-processors", +// ":modules:events-processor-tiles", - ":modules:events-server", - ":modules:events-server-api", - ":modules:events-server-core", + +// ":modules:events-server", +// ":modules:events-server-core", + ":modules:events-server-web", ":modules:events-server-syslog", ":modules:infra-factorio-client", @@ -29,10 +41,13 @@ include( ":modules:versions-platform", ) + enableFeaturePreview("TYPESAFE_PROJECT_ACCESSORS") + apply(from = "./buildSrc/repositories.settings.gradle.kts") + @Suppress("UnstableApiUsage") // Central declaration of repositories is an incubating feature dependencyResolutionManagement { repositoriesMode.set(RepositoriesMode.PREFER_SETTINGS)