diff --git a/build.gradle.kts b/build.gradle.kts index fd9e82d..ae6a2ff 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -8,7 +8,7 @@ plugins { `build-dashboard` // kotlin("multiplatform") apply false // try to stop error 'Projects must be configuring' - kotlin("js") apply false // try to stop error 'Projects must be configuring' +// kotlin("js") apply false // try to stop error 'Projects must be configuring' } group = "dev.adamko.kafkatorio" diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 7cda9a6..9d174aa 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -35,6 +35,8 @@ ktor = "2.0.2" hoplite = "2.1.5" +simpleSyslogParser = "0.0.6" + kotkaStreams = "dev-SNAPSHOT" ## test ## @@ -156,10 +158,12 @@ ktor-serverCallLogging = { group = "io.ktor", name = "ktor-server-call-l ktor-serverCallId = { group = "io.ktor", name = "ktor-server-call-id" } ktor-network = { group = "io.ktor", name = "ktor-network" } +ktor-networkTls = { group = "io.ktor", name = "ktor-network-tls" } ktor-test-clientMock = { group = "io.ktor", name = "ktor-client-mock" } ktor-test-serverHost = { group = "io.ktor", name = "ktor-server-test-host" } + ### jackson ## #jackson-bom = { group = " com.fasterxml.jackson.module", name = ":jackson-module-kotlin", version = "2.13.+" } #jackson-kotlin = { group = " com.fasterxml.jackson.module", name = ":jackson-module-kotlin", version = "2.13.+" } @@ -205,6 +209,10 @@ hoplite-core = { group = "com.sksamuel.hoplite", name = "hoplite-core", hoplite-datetime = { group = "com.sksamuel.hoplite", name = "hoplite-datetime", version.ref = "hoplite" } hoplite-yaml = { group = "com.sksamuel.hoplite", name = "hoplite-yaml", version.ref = "hoplite" } + +simpleSyslogParser = { group = "com.github.palindromicity", name = "simple-syslog", version.ref = "simpleSyslogParser" } + + ############# ### test ### ############# diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/packets/kafkatorioPacket.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/packets/kafkatorioPacket.kt index e1d52fb..b3f2a53 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/packets/kafkatorioPacket.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/packets/kafkatorioPacket.kt @@ -16,4 +16,13 @@ data class KafkatorioPacket( @Serializable @SerialName("kafkatorio.packet.KafkatorioPacketData") -sealed class KafkatorioPacketData +sealed class KafkatorioPacketData { + + @Serializable + @SerialName("kafkatorio.packet.KafkatorioPacketData.Error") + data class Error( + val message: String? = null, + val rawValue: String? = null, + ) : KafkatorioPacketData() + +} diff --git a/modules/events-server/build.gradle.kts b/modules/events-server/build.gradle.kts index 7978403..58084d5 100644 --- a/modules/events-server/build.gradle.kts +++ b/modules/events-server/build.gradle.kts @@ -45,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 } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt index 1aa3e53..d86e259 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt +++ b/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 = 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( + 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,11 +177,8 @@ internal class KafkatorioTopology( streams.setStateListener { newState: KafkaStreams.State, _ -> when (newState) { KafkaStreams.State.NOT_RUNNING, -// KafkaStreams.State.ERROR, - -> if (cont.isActive) { - cont.resume(newState) - } - else -> Unit // do nothing + KafkaStreams.State.ERROR -> if (cont.isActive) cont.resume(newState) + else -> Unit // do nothing } } @@ -127,6 +196,7 @@ internal class KafkatorioTopology( } } + private fun printTopologyDescription(appId: String, topology: Topology) { val description: TopologyDescription = topology.describe() @@ -146,6 +216,7 @@ internal class KafkatorioTopology( ) } + companion object { private fun Map.toProperties(): Properties = entries.fold(Properties()) { props, (k, v) -> diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt index 1bbfe9e..41c0206 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt @@ -48,14 +48,15 @@ fun allTopics(): Set = 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, 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" } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt index 6881ecb..85ac987 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/config/ApplicationProperties.kt +++ b/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, val kafkaStreamsConfig: Map, + + val socketServerHost: String, + val socketServerPort: Int, ) { companion object { diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt index 2515009..ae1e860 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt +++ b/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") diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/SyslogSocketServer.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/SyslogSocketServer.kt new file mode 100644 index 0000000..3bf3731 --- /dev/null +++ b/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.socketServerPortTcp, + ) + + +// 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 = MutableSharedFlow() + val messages: SharedFlow + 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 +) { + + 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.toSyslogMsg(): SyslogMsg { + val map = entries.mapNotNull { (k, v) -> + when (v) { + is String -> k to v + else -> null + } + }.toMap() + return SyslogMsg(map) + } + } +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt index f19cfc5..283f062 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt +++ b/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(value) + val packet = runCatching { + jsonMapper.decodeFromString(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 } } diff --git a/modules/events-server/src/main/resources/config.yml b/modules/events-server/src/main/resources/config.yml index fa31386..9b8eb5c 100644 --- a/modules/events-server/src/main/resources/config.yml +++ b/modules/events-server/src/main/resources/config.yml @@ -3,10 +3,16 @@ web-port: ${SERVER_PORT:-12080} 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} diff --git a/modules/infra-factorio-server/src/docker-compose.yml b/modules/infra-factorio-server/src/docker-compose.yml index 262ca06..3b12252 100644 --- a/modules/infra-factorio-server/src/docker-compose.yml +++ b/modules/infra-factorio-server/src/docker-compose.yml @@ -4,8 +4,6 @@ services: factorio-server: image: factoriotools/factorio:${FACTORIO_VERSION} - # logging: - # driver: "json-file" container_name: "factorio-server" restart: "no" networks: @@ -18,12 +16,11 @@ services: - "27015:27015/tcp" # rcon volumes: - "./factorio-server:/factorio" - # logging: - # driver: "mickyg/kafka-logdriver:latest" - # options: - # KAFKA_BROKER_ADDR: "kafka-kraft" - # LOG_TOPIC: "dockerlogs-factorio-server" - # LOG_TAG: "factorio-server" + logging: + driver: syslog + options: + syslog-address: "tcp://host.docker.internal:9002" + syslog-format: "rfc5424" # environment: # RCON_PORT: 27715 diff --git a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt index 7c0ca27..077472f 100644 --- a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt +++ b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt @@ -5,6 +5,7 @@ import dev.adamko.kafkatorio.schema.packets.ConsoleChatUpdate import dev.adamko.kafkatorio.schema.packets.ConsoleCommandUpdate import dev.adamko.kafkatorio.schema.packets.EntityUpdate import dev.adamko.kafkatorio.schema.packets.EventServerPacket +import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData import dev.adamko.kafkatorio.schema.packets.MapChunkUpdate import dev.adamko.kafkatorio.schema.packets.PlayerUpdate import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate @@ -74,7 +75,7 @@ class WebsocketService( is EventServerPacket.Kafkatorio -> when (val packetData = packet.packet.data) { - is PlayerUpdate -> reduxStore.dispatch( + is PlayerUpdate -> reduxStore.dispatch( FactorioUpdate.Player(packet.packet.tick, packetData) ) is ConfigurationUpdate, @@ -83,7 +84,8 @@ class WebsocketService( is PrototypesUpdate, is SurfaceUpdate, is EntityUpdate, - is MapChunkUpdate -> { + is MapChunkUpdate, + is KafkatorioPacketData.Error -> { // to be continued... } }