kafka -> websockets working!!

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

buildSrc/build.gradle.kts

@@ -37,11 +37,11 @@ tasks.withType<KotlinCompile>().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",
   )
 }
 

buildSrc/src/main/kotlin/dev/adamko/kafkatorio/lang/kotlin-jvm.gradle.kts

@@ -26,14 +26,14 @@ dependencies {
   //<editor-fold desc="Test">
 
   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<KotlinCompile>().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<Test> {

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 = [

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<KotlinCompile> {
+  kotlinOptions.freeCompilerArgs += listOf(
+    "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi",
+  )
+}

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<FactorioEvent<FactorioObjectData>>(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()
   }
 
-}
+}

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<Websocket> = MutableSharedFlow(
+    replay = 100,
+    extraBufferCapacity = 100,
+    BufferOverflow.DROP_OLDEST,
+  )
+
+  private val messages: MutableSharedFlow<String> = 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)
+
+  }
+}

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()

modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt

@@ -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<FactorioEvent<FactorioObjectData>> {
-  override fun serializer() = Serializer<FactorioEvent<FactorioObjectData>> { _, message ->
-    jsonMapper.encodeToString<FactorioEvent<FactorioObjectData>>(message).encodeToByteArray()
-  }
-
-  override fun deserializer() = Deserializer<FactorioEvent<FactorioObjectData>> { _, 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<FactorioEvent<FactorioObjectData>>(value)
-//      KXS.asA<FactorioEvent<FactorioObjectData>>(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()
-
-}

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<FactorioEvent<FactorioObjectData>> {
+  override fun serializer() = Serializer<FactorioEvent<FactorioObjectData>> { _, message ->
+    jsonMapper.encodeToString<FactorioEvent<FactorioObjectData>>(message).encodeToByteArray()
+  }
+
+  override fun deserializer() = Deserializer<FactorioEvent<FactorioObjectData>> { _, 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 { }
+})

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<T>(dataSerializer: KSerializer<T>) :
+internal class ListAsObjectSerializer<T>(dataSerializer: KSerializer<T>) :
   JsonTransformingSerializer<List<T>>(ListSerializer(dataSerializer)) {
 
   override fun transformDeserialize(element: JsonElement): JsonElement =
@@ -27,7 +27,7 @@ class ListAsObjectSerializer<T>(dataSerializer: KSerializer<T>) :
     )
 }
 
-val jsonMapper = Json {
+internal val jsonMapper = Json {
   prettyPrint = true
   prettyPrintIndent = "  "
   serializersModule = SerializersModule { }

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())
 
 }