Merge branch 'main' of github.com:adamko-dev/kafkatorio

modules/events-server/build.gradle.kts

@@ -1,12 +1,14 @@
 import dev.adamko.kafkatorio.task.KafkaConsumerGroupsTask
 import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
 
+
 plugins {
   dev.adamko.kafkatorio.lang.`kotlin-jvm`
   kotlin("plugin.serialization")
   application
 }
 
+
 description = """
     Receive raw Factorio Events from Kafka and process them into targeted topics or data formats. 
     
@@ -17,6 +19,7 @@ val projectId: String by project.extra
 
 val kafkaStateDir: Directory = layout.projectDirectory.dir("kafka-state")
 
+
 dependencies {
   implementation(platform(projects.modules.versionsPlatform))
 
@@ -42,6 +45,11 @@ dependencies {
 
   implementation(libs.okio.core)
 
+  implementation(libs.ktor.network)
+  implementation(libs.ktor.networkTls)
+
+  implementation(libs.simpleSyslogParser)
+
   implementation(libs.kafka.kotkaStreams) {
     isChanging = true
   }
@@ -49,9 +57,6 @@ dependencies {
   implementation(libs.bundles.hoplite)
 }
 
-//configurations.all {
-//  resolutionStrategy.cacheDynamicVersionsFor(1, TimeUnit.MINUTES)
-//}
 
 application {
   mainClass.set("dev.adamko.kafkatorio.processor.EventProcessorKt")
@@ -93,7 +98,3 @@ idea {
     excludeDirs = excludeDirs + kafkaStateDir.asFile
   }
 }
-
-//tasks.withType<KotlinNpmInstallTask>().all {
-//  notCompatibleWithConfigurationCache("error: Projects must be configuring")
-//}

modules/events-server/docker/.env

@@ -1,3 +1,3 @@
 # Do not edit manually. This file was created with task 'dockerEnv'
 COMPOSE_PROJECT_NAME=kafkatorio
-KAFKATORIO_VERSION=0.6.0
+KAFKATORIO_VERSION=0.6.1

modules/events-server/docker/docker-compose.yml

@@ -15,16 +15,16 @@ services:
     environment:
       KAFKA_BOOTSTRAP_SERVERS: "http://kafka:29092"
       # KAFKA_STREAMS_STATE_DIR: "http://kafka:29092"
-      SERVER_PORT: "12081"
+      SERVER_PORT: "8080"
     ports:
-      - "12080:12081" # http and ws
+      - "12080:8080" # http and ws
     networks:
       - "kafka-nw"
       - "kafkatorio-nw"
       - "traefik-nw"
     labels:
       - "traefik.enable=true"
-      - "traefik.http.services.events-server-kafkatorio.loadbalancer.server.port=12081"
+#      - "traefik.http.services.events-server-kafkatorio.loadbalancer.server.port=12081"
 
 networks:
   kafka-nw:

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

@@ -1,6 +1,9 @@
 package dev.adamko.kafkatorio.processor
 
+import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
+import dev.adamko.kafkatorio.processor.admin.TOPIC_SRC_SERVER_LOG
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.processor.tileserver.SyslogSocketServer
 import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer
 import dev.adamko.kafkatorio.processor.topology.broadcastToWebsocket
 import dev.adamko.kafkatorio.processor.topology.colourMapChunks
@@ -17,10 +20,17 @@ import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.suspendCancellableCoroutine
+import org.apache.kafka.clients.producer.KafkaProducer
 import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.header.internals.RecordHeader
+import org.apache.kafka.common.serialization.Serdes
 import org.apache.kafka.streams.KafkaStreams
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.StreamsConfig
@@ -29,12 +39,14 @@ import org.apache.kafka.streams.TopologyDescription
 
 
 internal class KafkatorioTopology(
-  private val websocketServer: WebsocketServer,
   private val appProps: ApplicationProperties,
+  private val websocketServer: WebsocketServer,
+  private val syslogServer: SyslogSocketServer,
 ) {
 
   private val builderScope = CoroutineScope(CoroutineName("KafkaTopology"))
 
+
   suspend fun start() = builderScope.launch {
     splitPackets()
 
@@ -42,8 +54,11 @@ internal class KafkatorioTopology(
     saveTiles()
 
     websocketBroadcaster()
+
+    syslogProducer()
   }
 
+
   private suspend fun splitPackets() {
     val builder = StreamsBuilder()
     val packets = factorioServerPacketStream(builder)
@@ -54,18 +69,21 @@ internal class KafkatorioTopology(
     launchTopology("splitPackets", topology)
   }
 
+
   private suspend fun groupTilesMapChunks() {
     val builder = StreamsBuilder()
     val topology = colourMapChunks(builder)
     launchTopology("groupTilesMapChunks", topology)
   }
 
+
   private suspend fun saveTiles() {
     val builder = StreamsBuilder()
     val topology = saveMapTiles(builder, appProps.tileDir)
     launchTopology("saveTiles", topology)
   }
 
+
   private suspend fun websocketBroadcaster() {
     val builder = StreamsBuilder()
     broadcastToWebsocket(websocketServer, builder)
@@ -75,11 +93,65 @@ internal class KafkatorioTopology(
     launchTopology(
       "websocketBroadcaster", topology, mapOf(
         StreamsConfig.COMMIT_INTERVAL_MS_CONFIG to "${1.seconds.inWholeMilliseconds}",
-        StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG) to "${10.milliseconds.inWholeMilliseconds}"
+        StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG) to "${1.milliseconds.inWholeMilliseconds}"
       )
     )
   }
 
+
+  private fun syslogProducer(
+    id: String = "syslog-producer"
+  ) {
+    val props: Properties = appProps.kafkaStreamsConfig.toProperties()
+
+    val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as String
+    props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)
+
+
+    val producer: KafkaProducer<String, String> = KafkaProducer(
+      props,
+      Serdes.String().serializer(),
+      Serdes.String().serializer(),
+    )
+
+    builderScope.coroutineContext.job.invokeOnCompletion {
+      producer.close()
+    }
+
+    syslogServer.messages
+      .filter { it.message?.startsWith("KafkatorioPacket:") == true }
+      .onEach { syslogMsg ->
+
+        val kafkatorioMessage = syslogMsg.message?.substringAfter("KafkatorioPacket:")
+
+        val record = ProducerRecord<String, String>(
+          TOPIC_SRC_SERVER_LOG,
+          "syslog-test",
+          kafkatorioMessage,
+        )
+
+        val headers = SyslogFieldKeys.values()
+          .filter { "HEADER_" in it.name }
+          .mapNotNull { headerKey ->
+            val headerValue = syslogMsg.src[headerKey.field]
+            when {
+              headerValue.isNullOrBlank() -> null
+              else                        -> RecordHeader(
+                headerKey.field,
+                headerValue.toByteArray()
+              )
+            }
+          }
+
+        headers.forEach { header ->
+          record.headers().add(header)
+        }
+
+        producer.send(record)
+      }.launchIn(builderScope)
+  }
+
+
   private suspend fun launchTopology(
     id: String,
     topology: Topology,
@@ -105,7 +177,7 @@ internal class KafkatorioTopology(
         streams.setStateListener { newState: KafkaStreams.State, _ ->
           when (newState) {
             KafkaStreams.State.NOT_RUNNING,
-            KafkaStreams.State.ERROR -> cont.resume(newState)
+            KafkaStreams.State.ERROR -> if (cont.isActive) cont.resume(newState)
             else                     -> Unit // do nothing
           }
         }
@@ -124,6 +196,7 @@ internal class KafkatorioTopology(
     }
   }
 
+
   private fun printTopologyDescription(appId: String, topology: Topology) {
     val description: TopologyDescription = topology.describe()
 
@@ -143,6 +216,7 @@ internal class KafkatorioTopology(
     )
   }
 
+
   companion object {
     private fun Map<String, String>.toProperties(): Properties =
       entries.fold(Properties()) { props, (k, v) ->

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

@@ -48,14 +48,15 @@ fun allTopics(): Set<String> = buildSet {
 
 val KafkatorioPacketData.topicName: String
   get() = when (this) {
-    is ConfigurationUpdate  -> packetTopicNames.getValue(ConfigurationUpdate::class)
-    is ConsoleChatUpdate    -> packetTopicNames.getValue(ConsoleChatUpdate::class)
-    is ConsoleCommandUpdate -> packetTopicNames.getValue(ConsoleCommandUpdate::class)
-    is PrototypesUpdate     -> packetTopicNames.getValue(PrototypesUpdate::class)
-    is SurfaceUpdate        -> packetTopicNames.getValue(SurfaceUpdate::class)
-    is EntityUpdate         -> packetTopicNames.getValue(EntityUpdate::class)
-    is MapChunkUpdate       -> packetTopicNames.getValue(MapChunkUpdate::class)
-    is PlayerUpdate         -> packetTopicNames.getValue(PlayerUpdate::class)
+    is ConfigurationUpdate        -> packetTopicNames.getValue(ConfigurationUpdate::class)
+    is ConsoleChatUpdate          -> packetTopicNames.getValue(ConsoleChatUpdate::class)
+    is ConsoleCommandUpdate       -> packetTopicNames.getValue(ConsoleCommandUpdate::class)
+    is PrototypesUpdate           -> packetTopicNames.getValue(PrototypesUpdate::class)
+    is SurfaceUpdate              -> packetTopicNames.getValue(SurfaceUpdate::class)
+    is EntityUpdate               -> packetTopicNames.getValue(EntityUpdate::class)
+    is MapChunkUpdate             -> packetTopicNames.getValue(MapChunkUpdate::class)
+    is PlayerUpdate               -> packetTopicNames.getValue(PlayerUpdate::class)
+    is KafkatorioPacketData.Error -> packetTopicNames.getValue(KafkatorioPacketData.Error::class)
   }
 
 
@@ -68,6 +69,7 @@ private val packetTopicNames: Map<KClass<out KafkatorioPacketData>, String> = ma
   EntityUpdate::class to "entity",
   MapChunkUpdate::class to "map-chunk",
   PlayerUpdate::class to "player",
+  KafkatorioPacketData.Error::class to "error",
 ).mapValues { (_, v) ->
   "$DOMAIN.packet.$v"
 }

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

@@ -12,6 +12,9 @@ data class ApplicationProperties(
   val tileDir: Path,
   val kafkaAdminConfig: Map<String, String>,
   val kafkaStreamsConfig: Map<String, String>,
+
+  val socketServerHost: String,
+  val socketServerPort: Int,
 ) {
 
   companion object {

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

@@ -3,6 +3,7 @@ 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.SyslogSocketServer
 import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer
 import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer
 import dev.adamko.kafkatorio.processor.tileserver.webServer
@@ -11,7 +12,6 @@ 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
 
 
@@ -22,12 +22,18 @@ suspend fun main() {
   val admin = KafkatorioKafkaAdmin(appProps)
   admin.createKafkatorioTopics()
 
+
+  val syslogServer = SyslogSocketServer(appProps)
   val wsServer = WebsocketServer()
   val tileServer = WebMapTileServer(appProps)
+  val topology = KafkatorioTopology(appProps, wsServer, syslogServer)
 
   coroutineScope {
     launch {
-      val topology = KafkatorioTopology(wsServer, appProps)
+      syslogServer.start()
+    }
+
+    launch {
       topology.start()
 
       println("launched KafkatorioTopology")

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

@@ -30,14 +30,17 @@ class DebounceProcessor<K, V>(
   private val storeName: String,
 ) : Processor<K, V, K, V> {
 
+
   private val checkInterval: DurationJvm = checkInterval.toJavaDuration()
 
   private lateinit var state: State<K, V>
 
+
   override fun init(context: ProcessorContext<K, V>) {
     state = Active(context)
   }
 
+
   override fun process(record: Record<K, V>) {
     (state as Active).apply {
       val (key, value, _) = record
@@ -49,21 +52,28 @@ class DebounceProcessor<K, V>(
     }
   }
 
+
   override fun close() {
     (state as? Active)?.schedule?.cancel()
     state = Closed()
   }
 
+
   private sealed interface State<K, V>
 
+
   private class Closed<K, V> : State<K, V>
 
+
   private inner class Active(
     val context: ProcessorContext<K, V>
   ) : State<K, V> {
     val store: TimestampedKeyValueStore<K, V> = context.getStateStore(storeName)
-    val schedule: Cancellable =
-      context.schedule(checkInterval, PunctuationType.WALL_CLOCK_TIME, ::checkRecords)
+    val schedule: Cancellable = context.schedule(
+      checkInterval,
+      PunctuationType.WALL_CLOCK_TIME,
+      ::checkRecords
+    )
 
     private fun checkRecords(now: Long) {
       store.useAll {

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

@@ -0,0 +1,168 @@
+package dev.adamko.kafkatorio.processor.tileserver
+
+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.processor.tileserver.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.utils.io.readUTF8Line
+import java.util.EnumSet
+import kotlinx.coroutines.CoroutineDispatcher
+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
+
+
+class SyslogSocketServer(
+  appProps: ApplicationProperties,
+  dispatcher: CoroutineDispatcher = Dispatchers.IO,
+) {
+
+  private val selectorManager: SelectorManager = SelectorManager(dispatcher)
+
+
+  private val serverSocketTcp = aSocket(selectorManager)
+    .tcp()
+    .bind(
+      appProps.socketServerHost,
+      appProps.socketServerPort,
+    )
+
+
+//  private val serverSocketUdp: BoundDatagramSocket = aSocket(selectorManager)
+//    .udp()
+//    .bind(
+//      localAddress = InetSocketAddress(
+//        appProps.socketServerHost,
+//        appProps.socketServerPortUdp,
+//      ),
+//    )
+
+
+  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() {
+//    udp()
+    tcp()
+  }
+
+//  private suspend fun udp() {
+//    println("[SyslogSocketServer] started listening: ${serverSocketUdp.localAddress}")
+//
+//    serverSocketUdp.incoming
+//      .receiveAsFlow()
+//      .map {udpPacket: Datagram ->
+//        val msg = udpPacket.packet.readText()
+//        println("[SyslogSocketServer] received msg: $msg")
+//        syslogParser.parseLine(msg).toSyslogMsg()
+//      }.collect(_messages)
+//  }
+
+  private suspend fun tcp() = 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 = channel.readUTF8Line()
+            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/processor/topology/colourMapChunks.kt

@@ -17,6 +17,7 @@ import dev.adamko.kotka.extensions.groupedAs
 import dev.adamko.kotka.extensions.materializedAs
 import dev.adamko.kotka.extensions.producedAs
 import dev.adamko.kotka.extensions.repartitionedAs
+import dev.adamko.kotka.extensions.stream
 import dev.adamko.kotka.extensions.streams.filter
 import dev.adamko.kotka.extensions.streams.map
 import dev.adamko.kotka.extensions.streams.mapValues
@@ -55,26 +56,30 @@ fun colourMapChunks(builder: StreamsBuilder): Topology {
 
   chunkTilesColouredTable032.streamMapChunkColouredTo(
     ChunkSize.CHUNK_032,
-    TOPIC_MAP_CHUNK_COLOURED_032_STATE
+    TOPIC_MAP_CHUNK_COLOURED_032_STATE,
   )
 
-  val chunkTilesColoured032Debounced: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
+  val chunkTilesColoured: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
     builder.stream(
-      TOPIC_MAP_CHUNK_COLOURED_STATE,
       consumedAs(
-        "$pid.consume.map-chunks-coloured",
-        kxsBinary.serde(),
-        kxsBinary.serde(),
+        "$pid.consume.map-chunks-debounced",
+        kxsBinary.serde(ServerMapChunkId.serializer()),
+        kxsBinary.serde(ServerMapChunkTiles.serializer(ColourHex.serializer())),
       ),
+      TOPIC_MAP_CHUNK_COLOURED_STATE,
     )
 
   mapOf(
-    ChunkSize.CHUNK_032 to ChunkSize.CHUNK_064,
-    ChunkSize.CHUNK_064 to ChunkSize.CHUNK_128,
-    ChunkSize.CHUNK_128 to ChunkSize.CHUNK_256,
+//    ChunkSize.CHUNK_032 to ChunkSize.CHUNK_064,
+//    ChunkSize.CHUNK_064 to ChunkSize.CHUNK_128,
+//    ChunkSize.CHUNK_128 to ChunkSize.CHUNK_256,
+//    ChunkSize.CHUNK_256 to ChunkSize.CHUNK_512,
+
+    ChunkSize.CHUNK_032 to ChunkSize.CHUNK_256,
     ChunkSize.CHUNK_256 to ChunkSize.CHUNK_512,
+
   ).forEach { (from, to) ->
-    chunkTilesColoured032Debounced
+    chunkTilesColoured
       .filter("$pid.combine-chunks.filter-$from-to-$to") { key, _ ->
         key.chunkSize == from
       }.groupByChunkPosition(to)
@@ -86,9 +91,9 @@ fun colourMapChunks(builder: StreamsBuilder): Topology {
       namePrefix = "groupTilesMapChunks",
       sourceTopic = TOPIC_MAP_CHUNK_COLOURED_032_STATE,
       sinkTopic = TOPIC_MAP_CHUNK_COLOURED_STATE,
-      inactivityDuration = 15.seconds,
-      keySerde = kxsBinary.serde<ServerMapChunkId>(),
-      valueSerde = kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
+      inactivityDuration = 5.seconds,
+      keySerde = kxsBinary.serde(ServerMapChunkId.serializer()),
+      valueSerde = kxsBinary.serde(ServerMapChunkTiles.serializer(ColourHex.serializer())),
     )
 }
 
@@ -182,7 +187,7 @@ private fun KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>.enr
     val missingProtos = mutableSetOf<TileProtoHashCode>()
 
     val tileColours = tiles.map.mapValues { (_, code) ->
-      colourDict.map.getOrElse(code) {
+      colourDict.getOrElse(code) {
         missingProtos.add(code)
         ColourHex.TRANSPARENT
       }

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

@@ -5,7 +5,9 @@ import dev.adamko.kafkatorio.processor.admin.TOPIC_SRC_SERVER_LOG
 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.common.Tick
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
+import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData
 import dev.adamko.kotka.extensions.component1
 import dev.adamko.kotka.extensions.component2
 import dev.adamko.kotka.extensions.consumedAs
@@ -31,8 +33,26 @@ fun factorioServerPacketStream(
       Serdes.String(),
     )
   ).map("decode-packets") { serverId: String, value: String ->
-    val packet = jsonMapper.decodeFromString<KafkatorioPacket>(value)
+    val packet = runCatching {
+      jsonMapper.decodeFromString<KafkatorioPacket>(value)
+    }.getOrElse { e ->
+     KafkatorioPacket(
+        modVersion = "unknown",
+        tick = Tick(0u),
+        data = KafkatorioPacketData.Error(
+          message = e.message,
+          rawValue = value,
+        )
+      )
+    }
+
+    if (packet.data is KafkatorioPacketData.Error) {
+      println("error parsing $TOPIC_SRC_SERVER_LOG message: ${packet.data}")
+    }
+
     FactorioServerId(serverId) to packet
+  }.filter { _, value ->
+    value.data !is KafkatorioPacketData.Error
   }
 }
 

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

@@ -1 +0,0 @@
-package dev.adamko.kafkatorio.processor.topology

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

@@ -29,7 +29,9 @@ value class TileProtoHashCode private constructor(val code: Int) {
 
 @JvmInline
 @Serializable
-value class TileColourDict(val map: Map<TileProtoHashCode, ColourHex>)
+value class TileColourDict(
+  private val map: Map<TileProtoHashCode, ColourHex>
+) : Map<TileProtoHashCode, ColourHex> by map
 
 
 /** Get the latest map-tile colours per server. */
@@ -48,9 +50,9 @@ fun KStream<FactorioServerId, PrototypesUpdate>.createTilePrototypeTable()
 
     TileColourDict(map)
   }.filter("$pid.filterMapTileProtos") { _, dict: TileColourDict ->
-    dict.map.isNotEmpty()
+    dict.isNotEmpty()
   }.peek { serverId, dict: TileColourDict ->
-    println("$pid: server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}")
+    println("$pid: server $serverId has TileColourDict[${dict.size}]: ${dict.entries.joinToString()}")
   }.repartition(
     repartitionedAs(
       "$pid.pre-table-repartition",

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

@@ -1,12 +1,18 @@
 # Factorio Events Processor config
 
-web-port: ${SERVER_PORT:-12080}
+web-port: ${SERVER_PORT:-8080}
 tile-dir: ${TILE_DIR:-./tile-dir}
 
+
+socket-server-host: ${SOCKET_SERVER_HOST:-127.0.0.1}
+socket-server-port: ${SOCKET_SERVER_PORT:-9002}
+
+
 kafka-admin-config:
   client.id: kafkatorio.events-server-admin
   bootstrap.servers: ${KAFKA_BOOTSTRAP_SERVERS:-http://localhost:9092}
 
+
 kafka-streams-config:
   application.id: kafkatorio-events-processor
   bootstrap.servers: ${KAFKA_BOOTSTRAP_SERVERS:-http://localhost:9092}