From 7813756c6967164440ec61e737b0319d7586b7f0 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Tue, 28 Dec 2021 16:26:23 +0100 Subject: [PATCH] kafka -> websockets working!! --- build.gradle.kts | 1 - buildSrc/build.gradle.kts | 10 +- .../kafkatorio/lang/kotlin-jvm.gradle.kts | 16 +-- gradle/libs.versions.toml | 2 +- modules/events-processor/build.gradle.kts | 15 +-- .../processor/FactorioEventsTopology.kt | 51 ++++++++-- .../kafkatorio/processor/WebsocketServer.kt | 98 +++++++++++++++++++ .../kafkatorio/processor/eventProcessor.kt | 32 ++++++ .../dev/adamko/kafkatorio/processor/main.kt | 82 ---------------- .../dev/adamko/kafkatorio/processor/serdes.kt | 39 ++++++++ .../adamko/kafkatorio/events/schema/json.kt | 4 +- .../kafkaconnect/websocketClient.kt | 57 +++++------ 12 files changed, 265 insertions(+), 142 deletions(-) create mode 100644 modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt create mode 100644 modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt delete mode 100644 modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt create mode 100644 modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes.kt diff --git a/build.gradle.kts b/build.gradle.kts index 94fee6c..0ed4990 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -30,7 +30,6 @@ tasks.wrapper { gradleVersion = "7.3.3" distributionType = Wrapper.DistributionType.ALL } -tasks.assemble { dependsOn(tasks.wrapper) } val startInfra by tasks.registering { group = project.name diff --git a/buildSrc/build.gradle.kts b/buildSrc/build.gradle.kts index dc26678..2a694a2 100644 --- a/buildSrc/build.gradle.kts +++ b/buildSrc/build.gradle.kts @@ -37,11 +37,11 @@ tasks.withType().configureEach { } kotlinOptions.freeCompilerArgs += listOf( - "-Xopt-in=kotlin.RequiresOptIn", - "-Xopt-in=kotlin.ExperimentalStdlibApi", - "-Xopt-in=kotlin.time.ExperimentalTime", - "-Xopt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", - "-Xopt-in=kotlinx.serialization.ExperimentalSerializationApi", + "-opt-in=kotlin.RequiresOptIn", + "-opt-in=kotlin.ExperimentalStdlibApi", + "-opt-in=kotlin.time.ExperimentalTime", + "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", + "-opt-in=kotlinx.serialization.ExperimentalSerializationApi", ) } diff --git a/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/lang/kotlin-jvm.gradle.kts b/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/lang/kotlin-jvm.gradle.kts index a0fe076..122d74a 100644 --- a/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/lang/kotlin-jvm.gradle.kts +++ b/buildSrc/src/main/kotlin/dev/adamko/kafkatorio/lang/kotlin-jvm.gradle.kts @@ -26,14 +26,14 @@ dependencies { // val junitVersion = "5.8.2" - testImplementation(enforcedPlatform("org.junit:junit-bom:$junitVersion")) + testImplementation(platform("org.junit:junit-bom:$junitVersion")) testImplementation("org.junit.jupiter:junit-jupiter") testRuntimeOnly("org.junit.platform:junit-platform-launcher") { because("Only needed to run tests in a version of IntelliJ IDEA that bundles older versions") } val kotestVersion = "5.0.3" - testImplementation(enforcedPlatform("io.kotest:kotest-bom:$kotestVersion")) + testImplementation(platform("io.kotest:kotest-bom:$kotestVersion")) testImplementation("io.kotest:kotest-runner-junit5:$kotestVersion") testImplementation("io.kotest:kotest-assertions-core:$kotestVersion") testImplementation("io.kotest:kotest-property:$kotestVersion") @@ -54,16 +54,16 @@ tasks.withType().configureEach { } kotlinOptions.freeCompilerArgs += listOf( - "-Xopt-in=kotlin.RequiresOptIn", - "-Xopt-in=kotlin.ExperimentalStdlibApi", - "-Xopt-in=kotlin.time.ExperimentalTime", -// "-Xopt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", - "-Xopt-in=kotlinx.serialization.ExperimentalSerializationApi", + "-opt-in=kotlin.RequiresOptIn", + "-opt-in=kotlin.ExperimentalStdlibApi", + "-opt-in=kotlin.time.ExperimentalTime", +// "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", + "-opt-in=kotlinx.serialization.ExperimentalSerializationApi", ) } tasks.compileTestKotlin { - kotlinOptions.freeCompilerArgs += "-Xopt-in=io.kotest.common.ExperimentalKotest" + kotlinOptions.freeCompilerArgs += "-opt-in=io.kotest.common.ExperimentalKotest" } tasks.withType { diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index edf618f..acfc587 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -149,11 +149,11 @@ http4k = [ "http4k-client-okhttp", "http4k-client-ws", "http4k-server", - # "http4k-kotlinx", "http4k-cloudnative", "http4k-format-xml", "http4k-format-json", "http4k-format-yaml", + "http4k-format-kotlinx", ] http4k-test = [ diff --git a/modules/events-processor/build.gradle.kts b/modules/events-processor/build.gradle.kts index dd7b8f6..ed4964e 100644 --- a/modules/events-processor/build.gradle.kts +++ b/modules/events-processor/build.gradle.kts @@ -1,3 +1,5 @@ +import org.jetbrains.kotlin.gradle.tasks.KotlinCompile + plugins { id("dev.adamko.kafkatorio.lang.kotlin-jvm") kotlin("plugin.serialization") @@ -28,12 +30,7 @@ dependencies { implementation(libs.kotlinx.coroutines) implementation(platform(libs.http4k.bom)) -// implementation(libs.http4k.core) - implementation(libs.http4k.cloudnative) -// implementation(libs.http4k.format.json) - implementation(libs.http4k.format.yaml) { - because("parsing yaml properties files") - } + implementation(libs.bundles.http4k) implementation(libs.http4k.format.kotlinx) @@ -43,3 +40,9 @@ dependencies { application { mainClass.set("dev.adamko.factorioevents.processor.mainKt") } + +tasks.withType { + kotlinOptions.freeCompilerArgs += listOf( + "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", + ) +} diff --git a/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/FactorioEventsTopology.kt b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/FactorioEventsTopology.kt index 4bd53cb..d9e2359 100644 --- a/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/FactorioEventsTopology.kt +++ b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/FactorioEventsTopology.kt @@ -1,20 +1,59 @@ package dev.adamko.kafkatorio.processor -import java.util.Properties +import dev.adamko.kafkatorio.events.schema.FactorioEvent +import dev.adamko.kafkatorio.events.schema.FactorioObjectData +import java.time.Duration +import kotlinx.serialization.decodeFromString +import kotlinx.serialization.encodeToString +import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.streams.KafkaStreams import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.TopologyDescription +import org.apache.kafka.streams.kstream.Consumed +import org.apache.kafka.streams.kstream.Produced -class FactorioEventsTopology { +class FactorioEventsTopology( + val websocketServer: WebsocketServer, + val builder: StreamsBuilder = StreamsBuilder(), +) { + + private val sourceTopic = "factorio-server-log" fun build() { - val topology = StreamsBuilder().build() + builder.stream( + sourceTopic, + Consumed.with(Serdes.String(), Serdes.String()) + ) + .mapValues { readOnlyKey, value -> + println("Mapping $readOnlyKey:$value") + jsonMapper.decodeFromString>(value) + } + .peek { _, value -> + websocketServer.sendMessage(jsonMapper.encodeToString(value)) + } + .to( + { _, value, _ -> +// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}") + value.data.objectName + }, + Produced.with(Serdes.String(), JsonSerdes) + ) + val topology = builder.build() - val properties = Properties() + val streams = KafkaStreams(topology, appProps.kafkaConfig) - val kafkaStreams = KafkaStreams(topology, properties) + streams.setUncaughtExceptionHandler(StreamsExceptionHandler()) + Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(1)) }) + +// streams.cleanUp() + + val description: TopologyDescription = topology.describe() + println(description) + + streams.start() } -} \ No newline at end of file +} diff --git a/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt new file mode 100644 index 0000000..4789991 --- /dev/null +++ b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt @@ -0,0 +1,98 @@ +package dev.adamko.kafkatorio.processor + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.launchIn +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.flow.withIndex +import kotlinx.coroutines.job +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.supervisorScope +import org.http4k.core.Request +import org.http4k.core.Response +import org.http4k.core.Status.Companion.OK +import org.http4k.routing.bind +import org.http4k.routing.websockets +import org.http4k.server.PolyHandler +import org.http4k.websocket.Websocket +import org.http4k.websocket.WsMessage +import org.http4k.websocket.WsStatus + + +class WebsocketServer { + + // create a scope+job for the server + private val serverJob = SupervisorJob() + private val serverScope = CoroutineScope(serverJob) + + private val clients: MutableSharedFlow = MutableSharedFlow( + replay = 100, + extraBufferCapacity = 100, + BufferOverflow.DROP_OLDEST, + ) + + private val messages: MutableSharedFlow = MutableSharedFlow( + replay = 100, + extraBufferCapacity = 100, + BufferOverflow.DROP_OLDEST, + ) + + private fun bindClient(ws: Websocket) = runBlocking(serverScope.coroutineContext) { + clients.emit(ws) + } + + fun sendMessage(msg: String) = runBlocking(serverScope.coroutineContext) { + messages.emit(msg) + } + + suspend fun start(): PolyHandler = supervisorScope builder@{ + + coroutineContext.job.invokeOnCompletion { e -> + e?.let { serverJob.completeExceptionally(it) } ?: serverJob.complete() + } + + clients + .withIndex() + .onEach { (id, ws) -> + // create a scope+job for each client + val clientJob = SupervisorJob(serverJob) + val clientScope = CoroutineScope(clientJob) + + clientJob.invokeOnCompletion { e -> + println("[clientJob.invokeOnCompletion] closing client ws $e") + when (e) { + null -> ws.close(WsStatus.NORMAL) + else -> ws.close(WsStatus.ABNORMAL_CLOSE) + } + } + + ws.onError { e -> + sendMessage("[ws.onError] client $id error [${e}]") + clientJob.completeExceptionally(e) + } + ws.onClose { status -> + sendMessage("[ws.onClose] client $id disconnected [${status}]") + clientJob.complete() + } + ws.onMessage { wsMsg -> + sendMessage("$id says: ${wsMsg.bodyString()}") + } + + sendMessage("new client $id!") + + messages + .onEach { msg -> ws.send(WsMessage(msg)) } + .launchIn(clientScope) + + } + .launchIn(serverScope) + + val http = { _: Request -> Response(OK) } + val ws = websockets("/ws" bind ::bindClient) + + return@builder PolyHandler(http, ws) + + } +} diff --git a/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt new file mode 100644 index 0000000..75b47fb --- /dev/null +++ b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt @@ -0,0 +1,32 @@ +package dev.adamko.kafkatorio.processor + + +import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import kotlinx.coroutines.runBlocking +import org.http4k.server.Undertow +import org.http4k.server.asServer + +val appProps = ApplicationProperties() + +fun main() = runBlocking { + + val wsServer = WebsocketServer() + + val topology = FactorioEventsTopology(wsServer) + topology.build() + + wsServer.start().asServer(Undertow(9073)).start().block() +} + +// +// +//// a standard websocket app +//val websocketServer = websockets( +// "/lua-objects" bind { ws: Websocket -> +// ws.send(WsMessage("bob")) +// ws.onMessage { +// println("server received: $it") +// ws.send(it) +// } +// } +//).asServer(Undertow(9097)).start() diff --git a/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt deleted file mode 100644 index 2f899bf..0000000 --- a/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt +++ /dev/null @@ -1,82 +0,0 @@ -package dev.adamko.kafkatorio.processor - -import dev.adamko.kafkatorio.events.schema.FactorioEvent -import dev.adamko.kafkatorio.events.schema.FactorioObjectData -import dev.adamko.kafkatorio.events.schema.jsonMapper -import dev.adamko.kafkatorio.processor.config.ApplicationProperties -import java.time.Duration -import kotlinx.serialization.decodeFromString -import kotlinx.serialization.encodeToString -import kotlinx.serialization.modules.SerializersModule -import org.apache.kafka.common.serialization.Deserializer -import org.apache.kafka.common.serialization.Serde -import org.apache.kafka.common.serialization.Serdes -import org.apache.kafka.common.serialization.Serializer -import org.apache.kafka.streams.KafkaStreams -import org.apache.kafka.streams.StreamsBuilder -import org.apache.kafka.streams.TopologyDescription -import org.apache.kafka.streams.kstream.Consumed -import org.apache.kafka.streams.kstream.Produced -import org.http4k.format.ConfigurableKotlinxSerialization -import org.http4k.format.asConfigurable -import org.http4k.format.withStandardMappings - - -object KXS : ConfigurableKotlinxSerialization({ - ignoreUnknownKeys = true - prettyPrint = true - prettyPrintIndent = " " - asConfigurable().withStandardMappings().done() - serializersModule = SerializersModule { } -}) - -object JsonSerdes : Serde> { - override fun serializer() = Serializer> { _, message -> - jsonMapper.encodeToString>(message).encodeToByteArray() - } - - override fun deserializer() = Deserializer> { _, bytes -> - jsonMapper.decodeFromString(bytes.decodeToString()) - } -} - -fun main() { - - val appProps = ApplicationProperties() - - val builder = StreamsBuilder() - - builder.stream( - "factorio-server-log", - Consumed.with(Serdes.String(), Serdes.String()) - ) - .mapValues { readOnlyKey, value -> - println("Mapping $readOnlyKey:$value") - jsonMapper.decodeFromString>(value) -// KXS.asA>(value) - } - .to( - { key, value, recordContext -> -// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}") - value.data.objectName - }, - Produced.with(Serdes.String(), JsonSerdes) - ) - - val topology = builder.build() - val properties = appProps.kafkaConfig - - val streams = KafkaStreams(topology, properties) - - streams.setUncaughtExceptionHandler(StreamsExceptionHandler()) - - Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(1)) }) - -// streams.cleanUp() - - val description: TopologyDescription = topology.describe() - println(description) - - streams.start() - -} diff --git a/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes.kt b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes.kt new file mode 100644 index 0000000..14c40bf --- /dev/null +++ b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes.kt @@ -0,0 +1,39 @@ +package dev.adamko.kafkatorio.processor + +import dev.adamko.kafkatorio.events.schema.FactorioEvent +import dev.adamko.kafkatorio.events.schema.FactorioObjectData +import kotlinx.serialization.decodeFromString +import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json +import kotlinx.serialization.modules.SerializersModule +import org.apache.kafka.common.serialization.Deserializer +import org.apache.kafka.common.serialization.Serde +import org.apache.kafka.common.serialization.Serializer +import org.http4k.format.ConfigurableKotlinxSerialization +import org.http4k.format.asConfigurable +import org.http4k.format.withStandardMappings + + +object JsonSerdes : Serde> { + override fun serializer() = Serializer> { _, message -> + jsonMapper.encodeToString>(message).encodeToByteArray() + } + + override fun deserializer() = Deserializer> { _, bytes -> + jsonMapper.decodeFromString(bytes.decodeToString()) + } +} + +val jsonMapper = Json { + prettyPrint = true + prettyPrintIndent = " " + serializersModule = SerializersModule { } +} + +object KXS : ConfigurableKotlinxSerialization({ + ignoreUnknownKeys = true + prettyPrint = true + prettyPrintIndent = " " + asConfigurable().withStandardMappings().done() + serializersModule = SerializersModule { } +}) diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/json.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/json.kt index 3723e74..d99ec7e 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/json.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/json.kt @@ -13,7 +13,7 @@ import kotlinx.serialization.modules.SerializersModule /** Factorio outputs lists as Json objects - this serializer converts an object back to a list. */ -class ListAsObjectSerializer(dataSerializer: KSerializer) : +internal class ListAsObjectSerializer(dataSerializer: KSerializer) : JsonTransformingSerializer>(ListSerializer(dataSerializer)) { override fun transformDeserialize(element: JsonElement): JsonElement = @@ -27,7 +27,7 @@ class ListAsObjectSerializer(dataSerializer: KSerializer) : ) } -val jsonMapper = Json { +internal val jsonMapper = Json { prettyPrint = true prettyPrintIndent = " " serializersModule = SerializersModule { } diff --git a/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/websocketClient.kt b/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/websocketClient.kt index b7ee898..1763679 100644 --- a/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/websocketClient.kt +++ b/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/websocketClient.kt @@ -7,40 +7,35 @@ import org.http4k.websocket.WsMessage fun main() { -// val client = WebsocketClient.blocking( -// uri = Uri.of("ws://localhost:8095/ws"), -//// autoReconnection = true, -//// timeout = Duration.ofSeconds(30) -// ) -//// client.send(WsMessage("server sent on connection")) -// -// client.received() -// .take(100) -// .forEachIndexed { index, wsMessage -> -// println("[$index] ${wsMessage.bodyString()}") -// } - - - val nonBlockingClient = WebsocketClient.nonBlocking( - uri = Uri.of("ws://localhost:9292/lua-objects"), - onError = { it.printStackTrace() }, - onConnect = { ws -> - println("onConnect, ws.upgradeRequest ${ws.upgradeRequest}") -// ws.send(WsMessage("client sent on connection")) - } + val client = WebsocketClient.blocking( + uri = Uri.of("ws://localhost:9073/ws"), +// autoReconnection = true, +// timeout = Duration.ofSeconds(30) ) +// client.send(WsMessage("server sent on connection")) - nonBlockingClient.onMessage { msg -> print(msg.bodyString()) } + client.received() + .take(100) + .forEachIndexed { index, wsMessage -> + println("[$index] ${wsMessage.bodyString()}") + } - nonBlockingClient.onClose { - println("non-blocking client closing") - } - - - nonBlockingClient.onError { it.printStackTrace() } - - - Thread.sleep(Duration.ofMinutes(1).toMillis()) +// val nonBlockingClient = WebsocketClient.nonBlocking( +// uri = Uri.of("ws://localhost:9097/lua-objects"), +// onError = { it.printStackTrace() }, +// onConnect = { ws -> +// println("onConnect, ws.upgradeRequest ${ws.upgradeRequest}") +//// ws.send(WsMessage("client sent on connection")) +// } +// ) +// +// nonBlockingClient.onMessage { msg -> print(msg.bodyString()) } +// nonBlockingClient.onClose { +// println("non-blocking client closing") +// } +//// nonBlockingClient.onError { it.printStackTrace() } +// +// Thread.sleep(Duration.ofMinutes(1).toMillis()) }