- start migrating tile-server to use Ktor

- start splitting up Kafka processor and tile-server to two subprojects

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

@@ -79,7 +79,7 @@ internal class KafkatorioTopology(
 
   private suspend fun saveTiles() {
     val builder = StreamsBuilder()
-    val topology = saveMapTiles(builder, appProps.tileDir)
+    val topology = saveMapTiles(builder, appProps.serverDataDir)
     launchTopology("saveTiles", topology)
   }
 

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

@@ -9,7 +9,7 @@ import java.nio.file.Path
 // loaded by hoplite
 data class ApplicationProperties(
   val webPort: PortVal,
-  val tileDir: Path,
+  val serverDataDir: Path,
   val kafkaAdminConfig: Map<String, String>,
   val kafkaStreamsConfig: Map<String, String>,
 

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

@@ -17,6 +17,7 @@ value class PortVal(val value: Int) {
 
   companion object {
     val DEFAULT = PortVal(8080)
-    val VALID_RANGE: IntRange = 1..65535
+    val VALID_RANGE: IntRange
+      get() = (1..65535) // get() is required to prevent init ordering causing NPE
   }
 }

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

@@ -2,20 +2,9 @@ package dev.adamko.kafkatorio.processor.serdes
 
 import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio
 import kotlinx.serialization.json.Json
-import org.http4k.format.ConfigurableKotlinxSerialization
-import org.http4k.format.asConfigurable
-import org.http4k.format.withStandardMappings
 
 val jsonMapper: Json = Json {
   prettyPrint = true
   prettyPrintIndent = "  "
   serializersModule = jsonMapperKafkatorio.serializersModule
 }
-
-object KXS : ConfigurableKotlinxSerialization({
-  ignoreUnknownKeys = true
-  prettyPrint = true
-  prettyPrintIndent = "  "
-  asConfigurable().withStandardMappings().done()
-  serializersModule = jsonMapperKafkatorio.serializersModule
-})

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

@@ -2,7 +2,6 @@ package dev.adamko.kafkatorio.processor.tileserver
 
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
 import java.security.MessageDigest
-import kotlin.io.path.absolutePathString
 import org.http4k.core.Filter
 import org.http4k.core.HttpHandler
 import org.http4k.core.MemoryBody
@@ -21,16 +20,19 @@ internal class WebMapTileServer(
   appProps: ApplicationProperties
 ) {
 
-  private val tilesLoader = ResourceLoader.Directory(appProps.tileDir.toFile().canonicalPath)
+  private val tilesLoader = ResourceLoader.Directory(appProps.serverDataDir.toFile().canonicalPath)
+
 
   private val filters = listOf(
     ETagChecker(),
   ).reduce(Filter::then)
 
+
   private val routes: RoutingHttpHandler = routes(
     "/tiles" bind GET to static(tilesLoader).withFilter(filters)
   )
 
+
   fun build(): HttpHandler {
     return routes
   }
@@ -69,6 +71,7 @@ internal class WebMapTileServer(
     }
   }
 
+
   companion object {
     private fun ByteArray.md5(): String? {
       return if (isEmpty()) {

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

@@ -32,11 +32,11 @@ private const val WEB_MAP_TILE_SIZE_PX = 256
 
 
 /**
- * @param[tileDirectory] [dev.adamko.kafkatorio.processor.config.ApplicationProperties.webmapTileDir]
+ * @param[serverDataDir] [dev.adamko.kafkatorio.processor.config.ApplicationProperties.serverDataDir]
  */
 fun saveMapTiles(
   builder: StreamsBuilder,
-  tileDirectory: Path,
+  serverDataDir: Path,
 ): Topology {
 
   val subdividedMapChunkTilesDebounced: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
@@ -50,7 +50,7 @@ fun saveMapTiles(
     )
 
   val savedChunkFilenames: KStream<ServerMapChunkId, TilePngFilename> =
-    subdividedMapChunkTilesDebounced.saveChunkAsImage(tileDirectory)
+    subdividedMapChunkTilesDebounced.saveChunkAsImage(serverDataDir)
 
   savedChunkFilenames.broadcastSavedTileUpdates()
 
@@ -95,10 +95,10 @@ private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.saveChunkA
 @Synchronized
 private fun saveTile(
   filename: TilePngFilename,
-  tileDirectory: Path,
+  serverDataDir: Path,
   img: ImmutableImage,
 ) {
-  val chunkImageFile = tileDirectory.resolve(filename.value).toFile()
+  val chunkImageFile = serverDataDir.resolve(filename.value).toFile()
   if (chunkImageFile.parentFile.mkdirs()) {
     println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
   }

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

@@ -0,0 +1,38 @@
+package dev.adamko.kafkatorio.server
+
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import io.ktor.http.HttpStatusCode
+import io.ktor.server.application.Application
+import io.ktor.server.application.install
+import io.ktor.server.cio.CIO
+import io.ktor.server.engine.embeddedServer
+import io.ktor.server.plugins.callloging.CallLogging
+import io.ktor.server.plugins.statuspages.StatusPages
+import io.ktor.server.response.respondText
+
+
+fun main() {
+  val appProps = ApplicationProperties.load()
+
+  embeddedServer(CIO, port = appProps.webPort.value) {
+
+    common()
+
+    webmapWebsocketServer()
+
+    webmapTileServer(appProps)
+
+    syslogSocketServer(appProps)
+
+  }.start(wait = true)
+}
+
+
+fun Application.common() {
+  install(StatusPages) {
+    exception<Throwable> { call, cause ->
+      call.respondText(text = "500: $cause", status = HttpStatusCode.InternalServerError)
+    }
+  }
+  install(CallLogging)
+}

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

@@ -0,0 +1,152 @@
+package dev.adamko.kafkatorio.server
+
+import com.github.palindromicity.syslog.AllowableDeviations
+import com.github.palindromicity.syslog.SyslogParser
+import com.github.palindromicity.syslog.SyslogParserBuilder
+import com.github.palindromicity.syslog.SyslogSpecification
+import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.server.SyslogMsg.Companion.toSyslogMsg
+import io.ktor.network.selector.SelectorManager
+import io.ktor.network.sockets.aSocket
+import io.ktor.network.sockets.awaitClosed
+import io.ktor.network.sockets.openReadChannel
+import io.ktor.server.application.Application
+import io.ktor.utils.io.readUTF8Line
+import java.util.EnumSet
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.asSharedFlow
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.yield
+
+
+fun Application.syslogSocketServer(appProps: ApplicationProperties) {
+  val server = SyslogSocketServer(appProps)
+
+  launch {
+    server.start()
+  }
+}
+
+
+class SyslogSocketServer(
+  appProps: ApplicationProperties,
+) {
+
+  private val selectorManager: SelectorManager = SelectorManager(Dispatchers.IO)
+
+
+  private val serverSocketTcp = aSocket(selectorManager)
+    .tcp()
+    .bind(
+      appProps.socketServerHost,
+      appProps.socketServerPort,
+    )
+
+
+  private val syslogParser: SyslogParser = SyslogParserBuilder()
+    .withDeviations(EnumSet.of(AllowableDeviations.PRIORITY, AllowableDeviations.VERSION))
+    .forSpecification(SyslogSpecification.RFC_5424)
+    .build()
+
+
+  private val _messages: MutableSharedFlow<SyslogMsg> = MutableSharedFlow()
+  val messages: SharedFlow<SyslogMsg>
+    get() = _messages.asSharedFlow()
+
+
+  suspend fun start(): Unit = coroutineScope {
+    println("[SyslogSocketServer] started listening: ${serverSocketTcp.localAddress}")
+
+    while (isActive) {
+      val socket = serverSocketTcp.accept()
+
+      launch {
+        val channel = socket.openReadChannel()
+        try {
+          while (socket.socketContext.isActive) {
+            val line = runCatching { channel.readUTF8Line() }.getOrNull()
+            if (line != null) {
+
+              println("[SyslogSocketServer] received: $line")
+
+              val syslog = syslogParser.parseLine(line).toSyslogMsg()
+
+              _messages.emit(syslog)
+            }
+
+            yield()
+          }
+        } catch (e: Exception) {
+          socket.awaitClosed()
+        }
+      }
+
+      yield()
+    }
+  }
+}
+
+
+data class SyslogMsg(
+  val src: Map<String, String?>
+) {
+
+  val message: String?
+    get() = src[SyslogFieldKeys.MESSAGE.field]
+
+  val headerAppName: String?
+    get() = src[SyslogFieldKeys.HEADER_APPNAME.field]
+
+  val headerHostName: String?
+    get() = src[SyslogFieldKeys.HEADER_HOSTNAME.field]
+
+  val headerPri: String?
+    get() = src[SyslogFieldKeys.HEADER_PRI.field]
+
+  val headerPriSeverity: String?
+    get() = src[SyslogFieldKeys.HEADER_PRI_SEVERITY.field]
+
+  val headerPriFacility: String?
+    get() = src[SyslogFieldKeys.HEADER_PRI_FACILITY.field]
+
+  val headerProcId: String?
+    get() = src[SyslogFieldKeys.HEADER_PROCID.field]
+
+  val headerTimestamp: String?
+    get() = src[SyslogFieldKeys.HEADER_TIMESTAMP.field]
+
+  val headerMsgId: String?
+    get() = src[SyslogFieldKeys.HEADER_MSGID.field]
+
+  val headerVersion: String?
+    get() = src[SyslogFieldKeys.HEADER_VERSION.field]
+
+  val structuredBase: String?
+    get() = src[SyslogFieldKeys.STRUCTURED_BASE.field]
+
+  val structuredElementIdFmt: String?
+    get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_FMT.field]
+
+  val structuredElementIdPnameFmt: String?
+    get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_FMT.field]
+
+  val structuredElementIdPnamePattern: String?
+    get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_PATTERN.field]
+
+  companion object {
+    fun Map<String, Any?>.toSyslogMsg(): SyslogMsg {
+      val map = entries.mapNotNull { (k, v) ->
+        when (v) {
+          is String -> k to v
+          else      -> null
+        }
+      }.toMap()
+      return SyslogMsg(map)
+    }
+  }
+}

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

@@ -0,0 +1,148 @@
+package dev.adamko.kafkatorio.server
+
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
+import dev.adamko.kafkatorio.schema.common.PlayerIndex
+import dev.adamko.kafkatorio.schema.common.SurfaceIndex
+import dev.adamko.kafkatorio.schema.common.TilePngFilename
+import io.ktor.http.ContentType
+import io.ktor.http.HttpStatusCode
+import io.ktor.http.content.EntityTagVersion
+import io.ktor.http.content.versions
+import io.ktor.resources.Resource
+import io.ktor.serialization.kotlinx.json.json
+import io.ktor.server.application.Application
+import io.ktor.server.application.ApplicationCall
+import io.ktor.server.application.call
+import io.ktor.server.application.install
+import io.ktor.server.http.content.LastModifiedVersion
+import io.ktor.server.plugins.autohead.AutoHeadResponse
+import io.ktor.server.plugins.compression.Compression
+import io.ktor.server.plugins.compression.excludeContentType
+import io.ktor.server.plugins.compression.minimumSize
+import io.ktor.server.plugins.conditionalheaders.ConditionalHeaders
+import io.ktor.server.plugins.contentnegotiation.ContentNegotiation
+import io.ktor.server.plugins.partialcontent.PartialContent
+import io.ktor.server.resources.Resources
+import io.ktor.server.resources.get
+import io.ktor.server.response.respond
+import io.ktor.server.response.respondFile
+import io.ktor.server.routing.routing
+import io.ktor.util.combineSafe
+import io.ktor.util.pipeline.PipelineContext
+import kotlin.io.path.Path
+import kotlinx.serialization.Serializable
+
+
+fun Application.webmapTileServer(appProps: ApplicationProperties) {
+
+  install(AutoHeadResponse)
+  install(Compression) {
+    default()
+    minimumSize(1024)
+    excludeContentType(ContentType.Video.Any)
+  }
+  install(ConditionalHeaders)
+  install(ContentNegotiation) {
+    json()
+  }
+  install(PartialContent)
+  install(Resources)
+
+  val server = WebmapTileServer(appProps)
+
+  routing {
+    with(server) {
+
+      get<FactorioServers.FactorioServer.MapData.Tile> { tile ->
+        handleTileRequest(tile)
+      }
+
+      get<FactorioServers> {
+
+      }
+    }
+  }
+}
+
+
+private class WebmapTileServer(
+  private val appProps: ApplicationProperties
+) {
+  private val serverDataDir by appProps::serverDataDir
+
+  suspend fun PipelineContext<Unit, ApplicationCall>.handleTileRequest(
+    tile: FactorioServers.FactorioServer.MapData.Tile
+  ) {
+
+    val tileFilename = TilePngFilename(
+      tile.serverId,
+      tile.surfaceIndex,
+      tile.zoomLevel,
+      tile.chunkX,
+      tile.chunkY,
+    )
+
+    val tileFile = serverDataDir.combineSafe(Path(tileFilename.value))
+
+    if (tileFile.exists()) {
+      call.respondFile(tileFile) {
+        versions = versions + listOf(
+          EntityTagVersion(tileFile.lastModified().hashCode().toString()),
+          LastModifiedVersion(tileFile.lastModified()),
+        )
+      }
+    } else {
+      call.respond(HttpStatusCode.NotFound)
+    }
+  }
+}
+
+
+@Serializable
+@Resource("servers")
+private class FactorioServers {
+
+  @Serializable
+  @Resource("{factorioServerId}")
+  data class FactorioServer(
+    val factorioServerId: FactorioServerId,
+  ) {
+
+    @Serializable
+    @Resource("map")
+    class MapData {
+
+      @Serializable
+      @Resource("tiles/s{surfaceIndex}/z{zoomLevel}/x{chunkX}/y{chunkY}.png")
+      data class Tile(
+        val parent: FactorioServer,
+        val surfaceIndex: SurfaceIndex,
+        val zoomLevel: Int,
+        val chunkX: Int,
+        val chunkY: Int,
+      ) {
+        val serverId: FactorioServerId by parent::factorioServerId
+      }
+
+      @Serializable
+      @Resource("buildings/s{surfaceIndex}")
+      data class Buildings(
+        val surfaceIndex: SurfaceIndex,
+        // a list of buildings... (Sealed interface, different types?)
+      )
+
+    }
+
+    @Serializable
+    @Resource("players")
+    class Players {
+
+      @Serializable
+      @Resource("{playerId}")
+      data class Id(
+        val playerIndex: PlayerIndex,
+      )
+    }
+  }
+}

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

@@ -0,0 +1,130 @@
+package dev.adamko.kafkatorio.server
+
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import io.ktor.server.application.Application
+import io.ktor.server.application.install
+import io.ktor.server.routing.routing
+import io.ktor.server.websocket.WebSockets
+import io.ktor.server.websocket.webSocket
+import io.ktor.websocket.DefaultWebSocketSession
+import io.ktor.websocket.Frame
+import io.ktor.websocket.WebSocketDeflateExtension
+import io.ktor.websocket.send
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.zip.Deflater
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.TimeSource
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.flow.update
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.job
+import kotlinx.coroutines.launch
+
+
+fun Application.webmapWebsocketServer() {
+  install(WebSockets) {
+    extensions {
+      install(WebSocketDeflateExtension) {
+        // Compression level to use for java.util.zip.Deflater
+        compressionLevel = Deflater.DEFAULT_COMPRESSION
+
+        // Prevent compressing small outgoing frames
+        compressIfBiggerThan(bytes = 4 * 1024)
+      }
+    }
+  }
+
+  val wsServer = WebmapWebsocketServer()
+
+  routing {
+
+    webSocket("/ws") {
+      wsServer.bindClient(this)
+    }
+  }
+}
+
+
+private class WebmapWebsocketServer {
+
+  private val startTime = TimeSource.Monotonic.markNow()
+
+  private val clients: MutableStateFlow<Set<WSConnection>> = MutableStateFlow(emptySet())
+
+  private val messagesToClients: MutableSharedFlow<String> = MutableSharedFlow(
+    extraBufferCapacity = Int.MAX_VALUE,
+    onBufferOverflow = BufferOverflow.DROP_OLDEST
+  )
+
+  private val messagesFromClients: MutableSharedFlow<Frame> = MutableSharedFlow(
+    extraBufferCapacity = Int.MAX_VALUE,
+    onBufferOverflow = BufferOverflow.DROP_OLDEST
+  )
+
+  suspend fun bindClient(session: DefaultWebSocketSession) {
+    val client = WSConnection(session)
+
+    clients.update { clients -> clients + client }
+
+    messagesToClients.onEach { msg ->
+      runCatching {
+        client.send(msg)
+      }
+    }.launchIn(client)
+
+    client.launch {
+      client.incoming.receiveAsFlow()
+        .collect(messagesFromClients)
+    }
+
+    client.launch {
+      while (isActive) {
+        runCatching {
+          client.send("elapsed ${startTime.elapsedNow()}")
+        }
+        delay(10.seconds)
+      }
+    }
+
+    client.coroutineContext.job.invokeOnCompletion {
+      log("client ${client.name} completed")
+      clients.update { clients -> clients - client }
+    }
+  }
+
+
+  suspend fun sendMessageToClients(msg: String) {
+    log("sending message to clients: $msg")
+    messagesToClients.emit(msg)
+  }
+}
+
+
+private data class WSConnection(
+  private val session: DefaultWebSocketSession,
+) : DefaultWebSocketSession by session {
+  val name: String = "user${lastId.getAndIncrement()}"
+
+  override suspend fun send(frame: Frame): Unit = runCatching {
+    super.send(frame)
+  }.fold(
+    onSuccess = { log("sent message to client $name: $frame") },
+    onFailure = { log("failed to send message to client $name: $frame") }
+  )
+
+  override fun equals(other: Any?): Boolean = name == other
+  override fun hashCode(): Int = name.hashCode()
+
+  companion object {
+    private val lastId = AtomicInteger(0)
+  }
+}
+
+
+private fun log(msg: String) = println("[WebmapWebsocketServer] $msg")

modules/events-server/src/main/resources/config.yml

@@ -1,7 +1,7 @@
 # Factorio Events Processor config
 
 web-port: ${SERVER_PORT:-8080}
-tile-dir: ${TILE_DIR:-./tile-dir}
+servers-data-dir: ${SERVER_DATA_DIR:-./server-data}
 
 
 socket-server-host: ${SOCKET_SERVER_HOST:-127.0.0.1}