From acbc15b6d4f3aa7d5868e8831ab67c677fbd0ad6 Mon Sep 17 00:00:00 2001
From: Adam <897017+aSemy@users.noreply.github.com>
Date: Sun, 26 Jun 2022 15:31:54 +0200
Subject: [PATCH] map chunk grouping: reduce the amount of tiles generated
 (thanks to pixelated css), and reduce debounce duration

---
 .../processor/misc/DebounceProcessor.kt       | 14 +++++++--
 .../processor/topology/colourMapChunks.kt     | 31 +++++++++++--------
 2 files changed, 30 insertions(+), 15 deletions(-)

diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt
index 2658751..9502e63 100644
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt
+++ b/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 {
diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt
index 2a171f0..c471d18 100644
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt
+++ b/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())),
     )
 }