refactor websocket/tile server

modules/events-server/.gitignore

@@ -1 +1,2 @@
 kafka-state/
+kafkatorio-tiles/

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

@@ -1,23 +1,26 @@
 package dev.adamko.kafkatorio.processor
 
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer
+import dev.adamko.kafkatorio.processor.topology.broadcastToWebsocket
 import dev.adamko.kafkatorio.processor.topology.colourMapChunks
 import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream
-import dev.adamko.kafkatorio.processor.topology.broadcastToWebsocket
 import dev.adamko.kafkatorio.processor.topology.saveMapTiles
 import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
-import java.time.Duration
 import java.util.Properties
-import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.resume
-import kotlin.coroutines.cancellation.CancellationException
+import kotlin.coroutines.resumeWithException
 import kotlin.io.path.Path
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancel
-import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.currentCoroutineContext
-import kotlinx.coroutines.cancelChildren
 import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.suspendCancellableCoroutine
+import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.streams.KafkaStreams
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.StreamsConfig
@@ -27,13 +30,12 @@ import org.apache.kafka.streams.TopologyDescription
 
 internal class KafkatorioTopology(
   private val websocketServer: WebsocketServer,
-  private val appProps: ApplicationProperties = ApplicationProperties(),
+  private val appProps: ApplicationProperties,
-) : CoroutineScope {
+) {
 
-  override val coroutineContext: CoroutineContext =
+  private val builderScope = CoroutineScope(CoroutineName("KafkaTopology"))
-    Dispatchers.Default + rootJob + CoroutineName("KafkatorioTopology")
 
-  fun start() {
+  suspend fun start() = builderScope.launch {
     splitPackets()
 
     groupTilesMapChunks()
@@ -42,7 +44,7 @@ internal class KafkatorioTopology(
     websocketBroadcaster()
   }
 
-  private fun splitPackets() {
+  private suspend fun splitPackets() {
     val builder = StreamsBuilder()
     val packets = factorioServerPacketStream(builder)
     splitFactorioServerPacketStream(packets)
@@ -52,45 +54,77 @@ internal class KafkatorioTopology(
     launchTopology("splitPackets", topology)
   }
 
-  private fun groupTilesMapChunks() {
+  private suspend fun groupTilesMapChunks() {
     val builder = StreamsBuilder()
     val topology = colourMapChunks(builder)
     launchTopology("groupTilesMapChunks", topology)
   }
 
-  private fun saveTiles() {
+  private suspend fun saveTiles() {
     val builder = StreamsBuilder()
-    val topology = saveMapTiles(builder, appProps.webmapTileDir)
+    val topology = saveMapTiles(builder, appProps.tileDir)
     launchTopology("saveTiles", topology)
   }
 
-  private fun websocketBroadcaster() {
+  private suspend fun websocketBroadcaster() {
     val builder = StreamsBuilder()
     broadcastToWebsocket(websocketServer, builder)
 
     val topology = builder.build()
 
-    launchTopology("websocketBroadcaster", topology)
+    launchTopology(
+      "websocketBroadcaster", topology, mapOf(
+        StreamsConfig.COMMIT_INTERVAL_MS_CONFIG to "${1.seconds.inWholeMilliseconds}",
+        StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG) to "${10.milliseconds.inWholeMilliseconds}"
+      )
+    )
   }
 
-  private fun launchTopology(
+  private suspend fun launchTopology(
     id: String,
     topology: Topology,
-    config: Map<String, String> = mapOf(),
+    additionalProperties: Map<String, String> = mapOf(),
   ) {
 
-    val props: Properties = appProps.kafkaConfig
+    val props: Properties = appProps.kafkaStreamsConfig.toProperties()
 
-    config.forEach { (k, v) -> props.setProperty(k, v) }
+    additionalProperties.forEach { (k, v) -> props.setProperty(k, v) }
 
-    val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String
+    val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as String
     props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)
 
-    val streams = KafkaStreams(topology, props)
+    builderScope.launch(CoroutineName(appId)) {
-    streams.setUncaughtExceptionHandler(StreamsExceptionHandler {
-      coroutineContext.job.cancel(CancellationException(cause = it))
-    })
 
+      val streamsState: KafkaStreams.State = suspendCancellableCoroutine { cont ->
+
+        val streams = KafkaStreams(topology, props)
+        streams.setUncaughtExceptionHandler(StreamsExceptionHandler { e ->
+          cont.resumeWithException(e)
+        })
+
+        streams.setStateListener { newState: KafkaStreams.State, _ ->
+          when (newState) {
+            KafkaStreams.State.NOT_RUNNING,
+            KafkaStreams.State.ERROR -> cont.resume(newState)
+            else                     -> Unit // do nothing
+          }
+        }
+
+        cont.invokeOnCancellation {
+          println("[$appId] coroutine cancelled -> closing. cause:$it")
+          streams.close()
+        }
+
+        printTopologyDescription(appId, topology)
+        streams.start()
+        println("launched Topology $appId")
+      }
+      println("exiting Topology $appId: $streamsState")
+      currentCoroutineContext().job.cancel("$appId: $streamsState")
+    }
+  }
+
+  private fun printTopologyDescription(appId: String, topology: Topology) {
     val description: TopologyDescription = topology.describe()
 
     val descFile = Path("./build/$appId.txt").toFile().apply {
@@ -107,19 +141,12 @@ internal class KafkatorioTopology(
         |----------------
       """.trimMargin()
     )
-
-    coroutineContext.job.invokeOnCompletion {
-      println("closing $id KafkaStreams")
-      rootJob.cancelChildren()
-      rootJob.cancel()
-      streams.close(Duration.ofSeconds(1))
-    }
-
-    launch { streams.start() }
   }
 
   companion object {
-    val rootJob = SupervisorJob()
+    private fun Map<String, String>.toProperties(): Properties =
+      entries.fold(Properties()) { props, (k, v) ->
+        props.apply { setProperty(k, v) }
+      }
   }
-
 }

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

@@ -1,93 +0,0 @@
-package dev.adamko.kafkatorio.processor
-
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.channels.BufferOverflow
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.launchIn
-import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.flow.withIndex
-import kotlinx.coroutines.job
-import kotlinx.coroutines.runBlocking
-import kotlinx.coroutines.supervisorScope
-import org.http4k.routing.bind
-import org.http4k.routing.websockets
-import org.http4k.websocket.Websocket
-import org.http4k.websocket.WsHandler
-import org.http4k.websocket.WsMessage
-import org.http4k.websocket.WsStatus
-
-
-class WebsocketServer {
-
-  // create a scope+job for the server
-  private val serverJob = KafkatorioTopology.rootJob
-  private val serverScope = CoroutineScope(serverJob)
-
-  private val clients: MutableSharedFlow<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 build(): WsHandler = supervisorScope builder@{
-
-    coroutineContext.job.invokeOnCompletion { e ->
-      e?.let { serverJob.completeExceptionally(it) } ?: serverJob.complete()
-    }
-
-    clients
-      .withIndex()
-      .onEach { (id, ws) ->
-        // create a scope+job for each client
-        val clientJob = SupervisorJob(serverJob)
-        val clientScope = CoroutineScope(clientJob)
-
-        clientJob.invokeOnCompletion { e ->
-          println("[clientJob.invokeOnCompletion] closing client ws $e")
-          when (e) {
-            null -> ws.close(WsStatus.NORMAL)
-            else -> ws.close(WsStatus.ABNORMAL_CLOSE)
-          }
-        }
-
-        ws.onError { e ->
-          sendMessage("[ws.onError] client $id error [${e}]")
-          clientJob.completeExceptionally(e)
-        }
-        ws.onClose { status ->
-          sendMessage("[ws.onClose] client $id disconnected [${status}]")
-          clientJob.complete()
-        }
-        ws.onMessage { wsMsg ->
-          sendMessage("$id says: ${wsMsg.bodyString()}")
-        }
-
-        sendMessage("new client $id!")
-
-        messages
-          .onEach { msg -> ws.send(WsMessage(msg)) }
-          .launchIn(clientScope)
-
-      }
-      .launchIn(serverScope)
-
-
-    return@builder websockets("/ws" bind ::bindClient)
-
-  }
-}

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

@@ -1,5 +1,6 @@
 package dev.adamko.kafkatorio.processor.admin
 
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
 import java.util.concurrent.TimeUnit
 import mu.KotlinLogging
 import org.apache.kafka.clients.admin.Admin
@@ -8,17 +9,17 @@ import org.apache.kafka.clients.admin.NewTopic
 import org.apache.kafka.clients.admin.TopicListing
 
 
-object KafkatorioKafkaAdmin {
+class KafkatorioKafkaAdmin(
+  appProps: ApplicationProperties
+) {
 
   private val logger = KotlinLogging.logger {}
 
-  private val kafkaAdmin: Admin =
+  init {
-    Admin.create(
+    println("Creating KafkatorioKafkaAdmin with config: ${appProps.kafkaAdminConfig}")
-      mapOf(
+  }
-        "bootstrap.servers" to "http://localhost:9092",
+
-        "client.id" to "kafkatorio.setup-admin",
+  private val kafkaAdmin: Admin = Admin.create(appProps.kafkaAdminConfig)
-      )
-    )
 
   fun createKafkatorioTopics() {
 

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

@@ -1,64 +1,94 @@
 package dev.adamko.kafkatorio.processor.config
 
-import dev.adamko.kafkatorio.processor.config.PortVal.Companion.portVal
+import com.sksamuel.hoplite.ConfigLoaderBuilder
+import com.sksamuel.hoplite.addEnvironmentSource
+import com.sksamuel.hoplite.addResourceOrFileSource
 import java.nio.file.Path
-import java.time.Clock
-import java.util.Properties
-import kotlin.io.path.Path
-import org.http4k.cloudnative.env.Environment
-import org.http4k.cloudnative.env.EnvironmentKey
-import org.http4k.lens.nonEmptyString
 
-/**
+
- * Load properties, in descending priority order, from
+// loaded by hoplite
- *
+data class ApplicationProperties(
- * 1. Environment variables
+  val webPort: PortVal,
- * 2. This app's env file (as defined by the environment variable [Lenses.APPLICATION_PROPERTIES_FILE_PATH])
+  val tileDir: Path,
- * # 3. A resources-dir yaml file
+  val kafkaAdminConfig: Map<String, String>,
- *
+  val kafkaStreamsConfig: Map<String, String>,
- */
-internal class ApplicationProperties(
-  vararg environments: Environment = arrayOf(
-    Environment.env(),
-    Environment.fromEnvVarPath(),
-    Environment.fromResourcesFile(".secret.config.yml")
-  )
 ) {
 
-
+  companion object {
-  private val appProps: Environment =
+    /** use Hoplite to load config from environment variables or properties files */
-    environments.reduce { acc, environment -> acc.overrides(environment) }
+    fun load(): ApplicationProperties =
-
+      ConfigLoaderBuilder.default()
-
+        .addDefaults()
-  fun clock(): Clock = Clock.systemUTC()
+        .allowUnresolvedSubstitutions()
-
+        .allowEmptyTree()
-
+        .addEnvironmentSource()
-  /**
+        .addResourceOrFileSource("/.secret.config.yml", optional = true)
-   * The port of the server
+        .addResourceOrFileSource("/config.yml", optional = false)
-   *
+        .build()
-   * Set by GCP ([docs](https://cloud.google.com/run/docs/reference/container-contract#env-vars)).
+        .loadConfigOrThrow()
-   */
+  }
-  val serverPort: PortVal
-    get() = EnvironmentKey.portVal().defaulted("PORT", PortVal.DEFAULT)[appProps]
-
-
-  val kafkaConfig: Properties
-    get() = EnvironmentKey
-      .nonEmptyString()
-      .map { srcString ->
-        srcString.reader().use { stringReader ->
-          Properties().also { properties ->
-            properties.load(stringReader)
-          }
-        }
-      }
-      .required("kafka.config")[appProps]
-
-
-  val webmapTileDir: Path
-    get() = EnvironmentKey
-      .nonEmptyString()
-      .map(::Path)
-      .required("tileServer.webmapTileDir")[appProps]
-
 }
+
+
+//import dev.adamko.kafkatorio.processor.config.PortVal.Companion.portVal
+//import java.nio.file.Path
+//import java.time.Clock
+//import java.util.Properties
+//import kotlin.io.path.Path
+//import org.http4k.cloudnative.env.Environment
+//import org.http4k.cloudnative.env.EnvironmentKey
+//import org.http4k.lens.nonEmptyString
+//
+///**
+// * Load properties, in descending priority order, from
+// *
+// * 1. Environment variables
+// * 2. This app's env file (as defined by the environment variable [Lenses.APPLICATION_PROPERTIES_FILE_PATH])
+// * # 3. A resources-dir yaml file
+// *
+// */
+//internal class ApplicationProperties(
+//  vararg environments: Environment = arrayOf(
+//    Environment.env(),
+//    Environment.fromEnvVarPath(),
+//    Environment.fromResourcesFile(".secret.config.yml")
+//  )
+//) {
+//
+//
+//  private val appProps: Environment =
+//    environments.reduce { acc, environment -> acc.overrides(environment) }
+//
+//
+//  fun clock(): Clock = Clock.systemUTC()
+//
+//
+//  /**
+//   * The port of the server
+//   *
+//   * Set by GCP ([docs](https://cloud.google.com/run/docs/reference/container-contract#env-vars)).
+//   */
+//  val serverPort: PortVal
+//    get() = EnvironmentKey.portVal().defaulted("PORT", PortVal.DEFAULT)[appProps]
+//
+//
+//  val kafkaConfig: Properties
+//    get() = EnvironmentKey
+//      .nonEmptyString()
+//      .map { srcString ->
+//        srcString.reader().use { stringReader ->
+//          Properties().also { properties ->
+//            properties.load(stringReader)
+//          }
+//        }
+//      }
+//      .required("kafka.config")[appProps]
+//
+//
+//  val webmapTileDir: Path
+//    get() = EnvironmentKey
+//      .nonEmptyString()
+//      .map(::Path)
+//      .required("tileServer.webmapTileDir")[appProps]
+//
+//}

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

@@ -1,38 +1,39 @@
 package dev.adamko.kafkatorio.processor.config
 
-import java.io.File
+//import java.io.File
-import org.http4k.cloudnative.env.Environment
+//import org.http4k.cloudnative.env.Environment
-import org.http4k.cloudnative.env.EnvironmentKey
+//import org.http4k.cloudnative.env.EnvironmentKey
-import org.http4k.core.Credentials
+//import org.http4k.core.Credentials
-import org.http4k.lens.BiDiLens
+//import org.http4k.lens.BiDiLens
-import org.http4k.lens.BiDiMapping
+//import org.http4k.lens.BiDiMapping
-
+//
-
+//
-internal object Lenses {
+//internal object Lenses {
-
+//
-  /** The location of this app's env file. */
+//  /** The location of this app's env file. */
-  val APPLICATION_PROPERTIES_FILE_PATH: BiDiLens<Environment, File?> =
+//  val APPLICATION_PROPERTIES_FILE_PATH: BiDiLens<Environment, File?> =
-    EnvironmentKey
+//    EnvironmentKey
-      .map(
+//      .map(
-        { path: String -> File(path) },
+//        { path: String -> File(path) },
-        { file: File -> file.canonicalPath },
+//        { file: File -> file.canonicalPath },
-      )
+//      )
-      .optional("APPLICATION_PROPERTIES_FILE_PATH")
+//      .optional("APPLICATION_PROPERTIES_FILE_PATH")
-
+//
-  /** Bidirectional [String] and [Credentials] mapper. */
+//
-  val credentialsLens: BiDiMapping<String, Credentials> = run {
+//  /** Bidirectional [String] and [Credentials] mapper. */
-    val separator = ":"
+//  val credentialsLens: BiDiMapping<String, Credentials> = run {
-    BiDiMapping(
+//    val separator = ":"
-      // from String to Credentials
+//    BiDiMapping(
-      { property ->
+//      // from String to Credentials
-        property.split(separator, limit = 2)
+//      { property ->
-          .takeIf { it.size == 2 }
+//        property.split(separator, limit = 2)
-          ?.let { (user, pass) -> Credentials(user, pass) }
+//          .takeIf { it.size == 2 }
-          ?: throw IllegalArgumentException("Credentials are invalid")
+//          ?.let { (user, pass) -> Credentials(user, pass) }
-      },
+//          ?: throw IllegalArgumentException("Credentials are invalid")
-      // from Credentials to String
+//      },
-      { it.user + separator + it.password }
+//      // from Credentials to String
-    )
+//      { it.user + separator + it.password }
-  }
+//    )
-
+//  }
-}
+//
+//}

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

@@ -1,74 +1,78 @@
 package dev.adamko.kafkatorio.processor.config
 
-import java.io.File
+//import java.io.File
-import java.io.FileNotFoundException
+//import java.io.FileNotFoundException
-import org.http4k.cloudnative.env.Environment
+//import org.http4k.cloudnative.env.Environment
-import org.http4k.cloudnative.env.MapEnvironment
+//import org.http4k.cloudnative.env.MapEnvironment
-import org.http4k.cloudnative.env.fromYaml
+//import org.http4k.cloudnative.env.fromYaml
-import org.http4k.lens.Lens
+//import org.http4k.lens.Lens
-
+//
-
+//
-/**
+///**
- * Load properties from either a `.yaml` or `.env` file.
+// * Load properties from either a `.yaml` or `.env` file.
- *
+// *
- * (An improved version of [Environment.fromResource])
+// * (An improved version of [Environment.fromResource])
- */
+// */
-fun Environment.Companion.fromResourcesFile(
+//fun Environment.Companion.fromResourcesFile(
-  resource: String,
+//  resource: String,
-  default: Environment = EMPTY
+//  default: Environment = EMPTY
-): Environment {
+//): Environment {
-  return try {
+//  return try {
-    val file =
+//    val file =
-      this::class.java
+//      this::class.java
-        .getResource("/${resource.removePrefix("/")}")
+//        .getResource("/${resource.removePrefix("/")}")
-        ?.file
+//        ?.file
-        ?.let { File(it) }
+//        ?.let { File(it) }
-    fromFile(file, default)
+//    fromFile(file, default)
-  } catch (e: FileNotFoundException) {
+//  } catch (e: FileNotFoundException) {
-    default
+//    default
-  }
+//  }
-}
+//}
-
+//
-/**
+//
- * Try fetching a properties file the location provided by the
+///**
- * [`JVM`][jvmProperties] or [`System`][env] environment variable
+// * Try fetching a properties file the location provided by the
- * [Lenses.APPLICATION_PROPERTIES_FILE_PATH], if it exists.
+// * [`JVM`][jvmProperties] or [`System`][env] environment variable
- *
+// * [Lenses.APPLICATION_PROPERTIES_FILE_PATH], if it exists.
- * If it does not exist, return [default].
+// *
- */
+// * If it does not exist, return [default].
-fun Environment.Companion.fromEnvVarPath(
+// */
-  location: Lens<Environment, File?> = Lenses.APPLICATION_PROPERTIES_FILE_PATH,
+//fun Environment.Companion.fromEnvVarPath(
-  default: Environment = EMPTY,
+//  location: Lens<Environment, File?> = Lenses.APPLICATION_PROPERTIES_FILE_PATH,
-): Environment {
+//  default: Environment = EMPTY,
-  val file = (jvmProperties() overrides env())[location]
+//): Environment {
-
+//  val file = (jvmProperties() overrides env())[location]
-  return fromFile(file, default)
+//
-}
+//  return fromFile(file, default)
-
+//}
-/**
+//
- * Load properties from either a `.yaml` or `.env` file.
+//
- *
+///**
- * (An improved version of [Environment.from])
+// * Load properties from either a `.yaml` or `.env` file.
- */
+// *
-internal fun Environment.Companion.fromFile(
+// * (An improved version of [Environment.from])
-  file: File?,
+// */
-  default: Environment = EMPTY,
+//internal fun Environment.Companion.fromFile(
-): Environment {
+//  file: File?,
-  return file
+//  default: Environment = EMPTY,
-    ?.takeIf { it.isFile }
+//): Environment {
-    .let {
+//  return file
-      when (it?.extension) {
+//    ?.takeIf { it.isFile }
-        "yml", "yaml" -> fromYaml(it)
+//    .let {
-        "env"         -> from(it)
+//      when (it?.extension) {
-        else          -> default
+//        "yml", "yaml" -> fromYaml(it)
-      }
+//        "env"         -> from(it)
-    }
+//        else          -> default
-}
+//      }
-
+//    }
-/** The same as [Environment.ENV], but refreshes on each invocation (which is useful for testing). */
+//}
-fun Environment.Companion.env(): Environment =
+//
-  MapEnvironment.from(System.getenv().toProperties())
+//
-
+///** The same as [Environment.ENV], but refreshes on each invocation (which is useful for testing). */
-/** The same as [Environment.JVM_PROPERTIES], but refreshes on each invocation (which is useful for testing). */
+//fun Environment.Companion.env(): Environment =
-internal fun Environment.Companion.jvmProperties(): Environment =
+//  MapEnvironment.from(System.getenv().toProperties())
-  MapEnvironment.from(System.getProperties())
+//
+//
+///** The same as [Environment.JVM_PROPERTIES], but refreshes on each invocation (which is useful for testing). */
+//internal fun Environment.Companion.jvmProperties(): Environment =
+//  MapEnvironment.from(System.getProperties())

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

@@ -2,33 +2,57 @@ package dev.adamko.kafkatorio.processor
 
 
 import dev.adamko.kafkatorio.processor.admin.KafkatorioKafkaAdmin
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
 import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer
-import kotlinx.coroutines.runBlocking
+import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer
-import org.http4k.server.PolyHandler
+import dev.adamko.kafkatorio.processor.tileserver.webServer
-import org.http4k.server.Undertow
+import kotlinx.coroutines.awaitCancellation
-import org.http4k.server.asServer
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.job
+import kotlinx.coroutines.launch
+import kotlinx.serialization.Serializable
+import org.http4k.server.Http4kServer
 
 
-fun main(): Unit = runBlocking {
+suspend fun main() {
 
-  KafkatorioKafkaAdmin.createKafkatorioTopics()
+  val appProps = ApplicationProperties.load()
+
+  val admin = KafkatorioKafkaAdmin(appProps)
+  admin.createKafkatorioTopics()
 
   val wsServer = WebsocketServer()
-  val tileServer = WebMapTileServer()
+  val tileServer = WebMapTileServer(appProps)
 
-  val topology = KafkatorioTopology(wsServer)
+  coroutineScope {
-  topology.start()
+    launch {
+      val topology = KafkatorioTopology(wsServer, appProps)
+      topology.start()
 
-  val webServer = PolyHandler(
+      println("launched KafkatorioTopology")
-    tileServer.build(),
-    wsServer.build(),
-  ).asServer(Undertow(9073))
 
-  Runtime.getRuntime().addShutdownHook(Thread {
+      awaitCancellation()
-    webServer.stop()
+    }
-  })
-//  launch {
-  webServer.start().block()
-//  }
 
+    launch {
+      val webServer: Http4kServer = webServer(tileServer, wsServer, appProps)
+
+      currentCoroutineContext().job.invokeOnCompletion {
+        println("webserver stopping. Cause: $it")
+        webServer.stop()
+      }
+
+      webServer.start()
+      println("launched webserver port:${webServer.port()}")
+
+      awaitCancellation()
+    }
+  }
 }
+//  Runtime.getRuntime().addShutdownHook(Thread {
+//    webServer.stop()
+//  })
+//  launch {
+//  }
+//}

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

@@ -1,6 +1,5 @@
 package dev.adamko.kafkatorio.processor.misc
 
-import dev.adamko.kafkatorio.processor.KafkatorioTopology
 import java.io.File
 import java.nio.file.FileSystems
 import java.nio.file.Files
@@ -56,7 +55,6 @@ class FileEventsFlow private constructor(
 
   override val coroutineContext: CoroutineContext =
     CoroutineName("FileEventsFlow") +
-        KafkatorioTopology.rootJob +
         Dispatchers.Unconfined +
         CoroutineExceptionHandler { _, e ->
           println("handling exception in FileEventsFlow")

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

@@ -18,18 +18,17 @@ import org.http4k.routing.routes
 import org.http4k.routing.static
 
 internal class WebMapTileServer(
-  appProps: ApplicationProperties = ApplicationProperties()
+  appProps: ApplicationProperties
 ) {
 
-  private val tilesLoader = ResourceLoader.Directory(appProps.webmapTileDir.absolutePathString())
+  private val tilesLoader = ResourceLoader.Directory(appProps.tileDir.toFile().canonicalPath)
 
   private val filters = listOf(
-    ETagChecker()(),
+    ETagChecker(),
   ).reduce(Filter::then)
 
   private val routes: RoutingHttpHandler = routes(
-    "/tiles" bind GET to
+    "/tiles" bind GET to static(tilesLoader).withFilter(filters)
-        static(tilesLoader).withFilter(filters)
   )
 
   fun build(): HttpHandler {
@@ -37,87 +36,40 @@ internal class WebMapTileServer(
   }
 
 
-  inner class ETagChecker
+  private class ETagChecker : Filter {
-//    : CoroutineScope
-  {
 
-//    private val tagState: MutableStateFlow<MutableMap<String, String>> =
+    override fun invoke(next: HttpHandler): HttpHandler = { request: Request ->
-//      MutableStateFlow(mutableMapOf())
+      val requestedETag = request.header("If-None-Match")
-//
-//    override val coroutineContext: CoroutineContext =
-//      CoroutineName("ETagChecker") +
-//          Dispatchers.Unconfined +
-//          KafkatorioTopology.rootJob
 
-//    init {
+      var response = next(request)
-//      launch {
+      val responseBytes = response.body.stream.use { it.readAllBytes() }
-//        appProps.webmapTileDir
+      val currentETag = responseBytes.md5()
-//          .fileEventsFlow()
-//          .filterIsInstance<KWatchEvent.PathEvent>()
-//          .filter { it.path.isRegularFile() }
-////          .filter { it.path.extension == "png" }
-//          .runningFold(mutableMapOf<String, String>()) { tags, event ->
-////            println("[WebMapTileServer] handling file update ${event.path}")
-//
-//            val relativePath = event.path
-//              .relativeTo(appProps.webmapTileDir)
-//              .invariantSeparatorsPathString
-//
-//            when (event) {
-//              is KWatchEvent.Deleted  -> tags.remove(relativePath)
-//              is KWatchEvent.Created,
-//              is KWatchEvent.Modified -> {
-//                val md5 = event.path.md5()
-//                if (md5 != null) {
-//                  tags[relativePath] = md5
-//                } else {
-//                  println("null md5 for ${event.path}")
-//                }
-//              }
-//            }
-//
-//            tags
-//          }.cancellable()
-//          .collect(tagState)
-//      }
-//    }
 
+      response = response.body(MemoryBody(responseBytes))
 
-    private val Request.etagKey: String get() = uri.path.removePrefix("/tiles/")
+      when {
-
+        currentETag == null          -> {
-
-    operator fun invoke() = Filter { next ->
-      { request ->
-        val requestedETag = request.header("If-None-Match")
-
-        var response = next(request)
-        val responseBytes = response.body.stream.use { it.readAllBytes() }
-        val currentETag = responseBytes.md5()
-
-        response = response.body(MemoryBody(responseBytes))
-
-        when {
-          currentETag == null          -> {
 //            println("[${request.etagKey}] currentETag is null (requested: $requestedETag), executing request")
-            response
+          response
-              .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
+            .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
-          }
+        }
-          requestedETag == currentETag -> {
+        requestedETag == currentETag -> {
 //            println("[${request.etagKey}] currentETag == requested, returning 304 $requestedETag")
-            Response(Status.NOT_MODIFIED)
+          Response(Status.NOT_MODIFIED)
-              .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
+            .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
-          }
+        }
-          else                         -> {
+        else                         -> {
 //            println("[${request.etagKey}] currentETag != requested, executing request and adding currentETag:$currentETag (requested:$requestedETag)")
-            response
+          response
-              .header("ETag", currentETag)
+            .header("ETag", currentETag)
 //              .header("Cache-Control", "no-store")
-              .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
+            .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
-          }
         }
       }
     }
+  }
 
+  companion object {
     private fun ByteArray.md5(): String? {
       return if (isEmpty()) {
         null

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

@@ -0,0 +1,16 @@
+package dev.adamko.kafkatorio.processor.tileserver
+
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import org.http4k.server.Http4kServer
+import org.http4k.server.PolyHandler
+import org.http4k.server.Undertow
+import org.http4k.server.asServer
+
+internal suspend fun webServer(
+  tileServer: WebMapTileServer,
+  wsServer: WebsocketServer,
+  appProps: ApplicationProperties,
+): Http4kServer = PolyHandler(
+  tileServer.build(),
+  wsServer.build(),
+).asServer(Undertow(appProps.webPort()))

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

@@ -0,0 +1,148 @@
+package dev.adamko.kafkatorio.processor.tileserver
+
+import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.TimeSource
+import kotlinx.coroutines.CoroutineExceptionHandler
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.withIndex
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.job
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.supervisorScope
+import kotlinx.coroutines.suspendCancellableCoroutine
+import org.http4k.routing.bind
+import org.http4k.routing.websockets
+import org.http4k.websocket.Websocket
+import org.http4k.websocket.WsHandler
+import org.http4k.websocket.WsMessage
+import org.http4k.websocket.WsStatus
+
+
+class WebsocketServer {
+
+  private val serverScope = CoroutineScope(
+    CoroutineName("WsServer") +
+        Dispatchers.IO +
+        SupervisorJob() +
+        CoroutineExceptionHandler { _, e ->
+          println("[ws.WsServer] handling exception ${e.message}")
+        }
+  )
+
+
+  private val clients: MutableSharedFlow<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) = serverScope.launch {
+    println("[ws.bindClient] ${ws.upgradeRequest.toMessage()}")
+    clients.emit(ws)
+  }
+
+
+  fun sendMessage(msg: String) = serverScope.launch {
+    println("[ws.sendMessage] $msg")
+    messages.emit(msg)
+  }
+
+
+  fun build(): WsHandler {
+    val startTime = TimeSource.Monotonic.markNow()
+
+    serverScope.launch {
+      supervisorScope {
+        clients
+          .withIndex()
+          .onEach { (id, ws) ->
+            createClient(id, ws)
+          }.launchIn(serverScope)
+      }
+    }
+
+    serverScope.launch {
+      while (isActive) {
+        delay(10.seconds)
+        sendMessage("elapsed ${startTime.elapsedNow()}")
+      }
+    }
+
+    return websockets(
+      "/ws/foo" bind { ws: Websocket ->
+        serverScope.launch { bindClient(ws) }
+      }
+    )
+  }
+
+
+  private fun createClient(id: Int, ws: Websocket) {
+    val clientJob = SupervisorJob(serverScope.coroutineContext.job)
+
+    serverScope.launch(CoroutineName("ws-client-$id") + clientJob) {
+//    return serverScope.launch(CoroutineName("ws-client-$id")) {
+//      supervisorScope {
+      launch {
+
+        suspendCancellableCoroutine<WsStatus> { cont ->
+          println("[ws-client] entering suspendCancellableCoroutine")
+
+          cont.invokeOnCancellation { e: Throwable? ->
+            launch { sendMessage("[ws-client.invokeOnCompletion] closing client ws ${e?.message}") }
+            when (e) {
+              null -> ws.close(WsStatus.NORMAL)
+              else -> ws.close(WsStatus.ABNORMAL_CLOSE)
+            }
+          }
+
+          ws.onError { e: Throwable ->
+            launch { sendMessage("[ws-client.onError] client $id error [${e.message}]") }
+            if (!cont.isCompleted) {
+              cont.resumeWithException(e)
+            }
+          }
+
+          ws.onClose { status: WsStatus ->
+            launch { sendMessage("[ws-client.onClose] client $id disconnected [${status}]") }
+            if (!cont.isCompleted) {
+              cont.resume(status)
+            }
+          }
+
+          ws.onMessage { wsMsg ->
+            launch { sendMessage("$id says: ${wsMsg.bodyString()}") }
+          }
+        }
+      }
+
+      println("[ws-client] $id listening to messages")
+      messages
+        .onEach { msg ->
+          println("[ws-client] sending message to $id")
+          ws.send(WsMessage(msg))
+        }
+        .launchIn(this)
+    }
+
+
+//      }
+  }
+}

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

@@ -6,6 +6,8 @@ import dev.adamko.kafkatorio.processor.admin.topicName
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kafkatorio.schema.common.FactorioServerId
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
+import dev.adamko.kotka.extensions.component1
+import dev.adamko.kotka.extensions.component2
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.extensions.producedAs
 import dev.adamko.kotka.extensions.streams.map
@@ -15,7 +17,6 @@ import kotlinx.serialization.decodeFromString
 import org.apache.kafka.common.serialization.Serdes
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.kstream.KStream
-import org.apache.kafka.streams.processor.RecordContext
 
 
 fun factorioServerPacketStream(
@@ -46,7 +47,7 @@ fun splitFactorioServerPacketStream(
         jsonMapper.serde(),
         jsonMapper.serde(),
       )
-    ) { _: FactorioServerId, value: KafkatorioPacket, _: RecordContext ->
+    ) { (_: FactorioServerId, value: KafkatorioPacket) ->
 //        println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
       value.data.topicName
     }

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

@@ -1,18 +1,22 @@
 package dev.adamko.kafkatorio.processor.topology
 
-import dev.adamko.kafkatorio.processor.WebsocketServer
 import dev.adamko.kafkatorio.processor.admin.TOPIC_BROADCAST_TO_WEBSOCKET
 import dev.adamko.kafkatorio.processor.admin.topicName
+import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer
 import dev.adamko.kafkatorio.schema.common.FactorioServerId
 import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio
 import dev.adamko.kafkatorio.schema.packets.EventServerPacket
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
 import dev.adamko.kafkatorio.schema.packets.PlayerUpdate
+import dev.adamko.kafkatorio.schema.packets.SurfaceUpdate
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.extensions.producedAs
+import dev.adamko.kotka.extensions.stream
 import dev.adamko.kotka.extensions.streams.forEach
 import dev.adamko.kotka.extensions.streams.map
+import dev.adamko.kotka.extensions.streams.to
 import dev.adamko.kotka.kxs.serde
+import kotlinx.coroutines.runBlocking
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
 import org.apache.kafka.streams.StreamsBuilder
@@ -32,7 +36,7 @@ fun broadcastToWebsocket(
   builder: StreamsBuilder = StreamsBuilder(),
 ) {
 
-  builder.preparePlayerUpdatesBroadcast()
+  builder.broadcastKafkatorioPackets()
 
   builder.stream<FactorioServerId, EventServerPacket>(
     TOPIC_BROADCAST_TO_WEBSOCKET,
@@ -40,25 +44,54 @@ fun broadcastToWebsocket(
   ).forEach(
     "$pid.websocket-server.send-message"
   ) { _, esPacket ->
-    websocketServer.sendMessage(jsonMapper.encodeToString(esPacket))
+    runBlocking {
+      websocketServer.sendMessage(jsonMapper.encodeToString(esPacket))
+    }
   }
 }
 
 
-private fun StreamsBuilder.preparePlayerUpdatesBroadcast() {
+private fun StreamsBuilder.broadcastKafkatorioPackets() {
-  stream<FactorioServerId, KafkatorioPacket>(
+  val playerUpdatesStream = stream(
+    consumedAs(
+      "$pid.player-updates",
+      jsonMapper.serde(FactorioServerId.serializer()),
+      jsonMapper.serde(KafkatorioPacket.serializer()),
+    ),
     PlayerUpdate::class.topicName,
-    consumedAs("$pid.player-updates", jsonMapper.serde(), jsonMapper.serde()),
+    SurfaceUpdate::class.topicName,
-  ).map<FactorioServerId, KafkatorioPacket, FactorioServerId, EventServerPacket>(
+  )
-    "$pid.player-updates.convert"
+
-  ) { serverId, playerPacket ->
+  playerUpdatesStream.map<FactorioServerId, KafkatorioPacket, FactorioServerId, EventServerPacket>(
-    serverId to EventServerPacket.Kafkatorio(playerPacket)
+    "$pid.packet-updates.convert"
+  ) { serverId, packet ->
+    serverId to EventServerPacket.Kafkatorio(packet)
+  }.to(
+    producedAs(
+      "$pid.event-server-packets.produce",
+      jsonMapper.serde(FactorioServerId.serializer()),
+      jsonMapper.serde(EventServerPacket.serializer()),
+    )
+  ) { TOPIC_BROADCAST_TO_WEBSOCKET }
+}
+
+
+private fun StreamsBuilder.prepareServerUpdateBroadcast() {
+
+  val surfaceUpdatesStream = stream<FactorioServerId, KafkatorioPacket>(
+    SurfaceUpdate::class.topicName,
+    consumedAs("$pid.surface-updates", jsonMapper.serde(), jsonMapper.serde()),
+  )
+
+  surfaceUpdatesStream.map<FactorioServerId, KafkatorioPacket, FactorioServerId, EventServerPacket>(
+    "$pid.surface-updates.convert"
+  ) { serverId, serverUpdatePacket ->
+    serverId to EventServerPacket.Kafkatorio(serverUpdatePacket)
   }.to(
-    TOPIC_BROADCAST_TO_WEBSOCKET,
     producedAs(
       "$pid.player-updates.stream",
       jsonMapper.serde(FactorioServerId.serializer()),
       jsonMapper.serde(EventServerPacket.serializer()),
     )
-  )
+  ) { TOPIC_BROADCAST_TO_WEBSOCKET }
 }

modules/events-server/src/main/resources/.gitignore

@@ -1 +0,0 @@
-kafkatorio-web-map/**