rename events-processor to events-server

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

@@ -0,0 +1,59 @@
+package dev.adamko.kafkatorio.processor
+
+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(
+  private val websocketServer: WebsocketServer,
+  private val builder: StreamsBuilder = StreamsBuilder(),
+) {
+
+  private val sourceTopic = "factorio-server-log"
+
+  fun 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 streams = KafkaStreams(topology, appProps.kafkaConfig)
+
+    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-server/src/main/kotlin/dev/adamko/kafkatorio/processor/StreamsExceptionHandler.kt

@@ -0,0 +1,11 @@
+package dev.adamko.kafkatorio.processor
+
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse
+
+class StreamsExceptionHandler : StreamsUncaughtExceptionHandler {
+  override fun handle(exception: Throwable): StreamThreadExceptionResponse {
+    println("${exception.message} - ${exception.stackTrace}")
+    return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION
+  }
+}

modules/events-server/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 build(): 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-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt

@@ -0,0 +1,59 @@
+package dev.adamko.kafkatorio.processor.config
+
+import dev.adamko.kafkatorio.processor.config.PortVal.Companion.portVal
+import java.time.Clock
+import java.util.Properties
+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
+ *
+ */
+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()
+
+  //<editor-fold desc="Server properties">
+
+  /**
+   * 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]
+
+  //</editor-fold>
+
+
+  //<editor-fold desc="Kafka properties">
+
+  val kafkaConfig: Properties
+    get() = EnvironmentKey
+      .nonEmptyString()
+      .map {
+        it.reader().use { sr ->
+          Properties().also { p ->
+            p.load(sr)
+          }
+        }
+      }
+      .required("kafka.config")[appProps]
+
+  //</editor-fold>
+
+}

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

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

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

@@ -0,0 +1,30 @@
+package dev.adamko.kafkatorio.processor.config
+
+import org.http4k.cloudnative.env.Port
+import org.http4k.lens.BiDiLensSpec
+import org.http4k.lens.StringBiDiMappings.int
+import org.http4k.lens.map
+
+/**
+ * Improved (slightly) version of http4k's [Port].
+ */
+@JvmInline
+value class PortVal(val value: Int) {
+
+  init {
+    require(value in VALID_RANGE) {
+      "Out of range. Port: '$value', range: $VALID_RANGE"
+    }
+  }
+
+  operator fun invoke(): Int = value
+
+  companion object {
+    val DEFAULT = PortVal(8080)
+    val VALID_RANGE: IntRange
+      get() = (1..65535)
+
+    fun <IN : Any> BiDiLensSpec<IN, String>.portVal(): BiDiLensSpec<IN, PortVal> =
+      map(int().map(::PortVal, PortVal::value))
+  }
+}

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

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

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

@@ -0,0 +1,19 @@
+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.build().asServer(Undertow(9073)).start().block()
+}

modules/events-server/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-server/src/main/resources/logback.xml

@@ -0,0 +1,21 @@
+<configuration>
+
+  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+    <withJansi>true</withJansi>
+    <!-- encoders are assigned the type
+         ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+    <encoder>
+      <pattern>%cyan(%d{HH:mm:ss.SSS}) %gray([%thread]) %highlight(%-5level) %magenta(%logger{36}) - %msg%n
+      </pattern>
+    </encoder>
+  </appender>
+
+  <logger name="org.apache.kafka" level="info" additivity="false">
+    <appender-ref ref="CONSOLE"/>
+  </logger>
+
+  <root level="info">
+    <appender-ref ref="STDOUT"/>
+  </root>
+
+</configuration>