lazzexus/Kafkatorio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

map chunk grouping: reduce the amount of tiles generated (thanks to pixelated css), and reduce debounce duration

Browse source
This commit is contained in:
Adam 2022-06-26 15:31:54 +02:00
parent f3f92afd22
commit acbc15b6d4
2 changed files with 30 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

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

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

31
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt
View file

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