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

chunk pngs almost working...

Browse source 
- rm yarn lock, it should be in ./gradle/kotlin-js-store
 - move kafka extensions to another project, kotka-streams
 - move prototypes to their own package
 - create a Factorio command for emitting prototypes
 - add bounding box to chunk updates
 - bump typed-factorio version
 - fix MapEntityPosition -> MapChunkPosition x/y rounding
 - lots of work on the Kafka Streams topology
This commit is contained in:
Adam 2022-01-16 23:16:09 +01:00
parent 49ad3654a3
commit 8533be0af8
47 changed files with 1097 additions and 4179 deletions
Download patch file Download diff file Expand all files Collapse all files

3486
.kotlin-js-store/yarn.lock
View file

File diff suppressed because it is too large Load diff

2
build.gradle.kts
View file

@ -6,7 +6,7 @@ plugins {
}
group = "dev.adamko.kafkatorio"
version = "0.2.3"
version = "0.2.4"
val licenseFile: RegularFile by extra(layout.projectDirectory.file("LICENSE"))

2
buildSrc/build.gradle.kts
View file

@ -64,5 +64,3 @@ idea {
isDownloadJavadoc = true
}
}
//tasks.assemble { dependsOn(tasks.htmlDependencyReport) }

19
buildSrc/repositories.settings.gradle.kts
View file

@ -1,4 +1,4 @@
@Suppress("UnstableApiUsage")
@Suppress("UnstableApiUsage") // Central declaration of repositories is an incubating feature
dependencyResolutionManagement {
// https://github.com/gradle/gradle/issues/15754#issuecomment-763614651
@ -11,10 +11,10 @@ dependencyResolutionManagement {
repositories {
mavenCentral()
gradlePluginPortal()
jitpack()
maven("https://maven.pkg.jetbrains.space/kotlin/p/kotlin/dev/")
maven("https://maven.pkg.jetbrains.space/public/p/kotlinx-html/maven")
gradlePluginPortal()
// maven("https://maven.pkg.jetbrains.space/kotlin/p/kotlin/dev/")
// maven("https://maven.pkg.jetbrains.space/public/p/kotlinx-html/maven")
// Declare the Node.js download repository
ivy("https://nodejs.org/dist/") {
@ -37,7 +37,7 @@ dependencyResolutionManagement {
gradlePluginPortal()
mavenCentral()
jitpack()
maven("https://maven.pkg.jetbrains.space/kotlin/p/kotlin/dev/")
// maven("https://maven.pkg.jetbrains.space/kotlin/p/kotlin/dev/")
}
}
@ -45,5 +45,12 @@ dependencyResolutionManagement {
}
fun RepositoryHandler.jitpack() {
maven("https://jitpack.io")
maven("https://jitpack.io") {
// content {
// includeGroup("com.github.aSemy")
// includeGroup("com.github.adamko-dev")
// includeGroup("com.github.adamko-dev.kotka-streams")
// includeGroup("dev.adamko.kotka")
// }
}
}

9
buildSrc/src/main/kotlin/dev/adamko/kafkatorio/lang/kotlin-multiplatform.gradle.kts
View file

@ -3,6 +3,7 @@ package dev.adamko.kafkatorio.lang
import org.gradle.kotlin.dsl.configure
import org.gradle.kotlin.dsl.kotlin
import org.jetbrains.kotlin.gradle.targets.js.nodejs.NodeJsRootExtension
import org.jetbrains.kotlin.gradle.targets.js.yarn.YarnRootExtension
plugins {
@ -21,4 +22,12 @@ afterEvaluate {
}
}
rootProject.extensions.configure<YarnRootExtension> {
// kotlin-js adds a directory in the root-dir for the Yarn lock.
// That's a bit annoying. It's a little neater if it's in the
// gradle dir, next to the version-catalog.
lockFileDirectory = rootProject.rootDir.resolve("gradle/kotlin-js-store")
}
}

1
modules/events-mod/src/main/typescript/control.ts
View file

@ -1,5 +1,6 @@
import {handleConfigurationUpdate} from "./main/config-update/handleConfigurationUpdate";
require("./main/events/registerHandlers")
require("./main/commands/kafkatorioCommands")
script.on_configuration_changed(handleConfigurationUpdate)

1
modules/events-mod/src/main/typescript/generated/kafkatorio-schema/config-update.d.ts vendored
View file

@ -16,5 +16,4 @@ interface FactorioConfigurationUpdate extends KafkatorioPacket {
factorioData: FactorioGameDataUpdate;
modVersion: string;
packetType: PacketType;
prototypes: FactorioPrototype[];
}

1
modules/events-mod/src/main/typescript/generated/kafkatorio-schema/object-data.d.ts vendored
View file

@ -50,6 +50,7 @@ interface MapTiles extends FactorioObjectData {
}
interface MapChunk extends FactorioObjectData {
area: MapBoundingBox;
objectName: ObjectName;
position: MapChunkPosition;
tiles: MapTiles;

6
modules/events-mod/src/main/typescript/generated/kafkatorio-schema/prototype.d.ts vendored
View file

@ -6,6 +6,12 @@ interface FactorioPrototype {
prototypeObjectName: PrototypeObjectName;
}
interface FactorioPrototypes extends KafkatorioPacket {
modVersion: string;
packetType: PacketType;
prototypes: FactorioPrototype[];
}
interface MapTilePrototype extends FactorioPrototype {
canBeMined: boolean;
collisionMasks: string[];

17
modules/events-mod/src/main/typescript/generated/kafkatorio-schema/schema.d.ts vendored
View file

@ -1,6 +1,6 @@
// Generated by TypeScriptGenerator - do not edit this file manually
type PacketType = "EVENT" | "CONFIG";
type PacketType = "EVENT" | "CONFIG" | "PROTOTYPES";
interface KafkatorioPacket {
modVersion: string;
@ -27,12 +27,17 @@ interface MapEntityPosition {
y: double;
}
interface MapChunkPosition {
x: int;
y: int;
}
interface MapTilePosition {
x: int;
y: int;
}
interface MapBoundingBox {
bottomRight: MapTilePosition;
topLeft: MapTilePosition;
}
interface MapChunkPosition {
x: int;
y: int;
}

12
modules/events-mod/src/main/typescript/main/commands/kafkatorioCommands.ts Normal file
View file

@ -0,0 +1,12 @@
import {emitPrototypes} from "../config-update/prototypes";
commands.add_command(
"kafkatorio",
"kafkatorio innit bruv",
(e: CustomCommandData) => {
if (e.parameter == undefined) {
// do nothing
} else if ("PROTOTYPES" == e.parameter.toUpperCase()) {
emitPrototypes()
}
})

6
modules/events-mod/src/main/typescript/main/config-update/handleConfigurationUpdate.ts
View file

@ -1,6 +1,6 @@
import {emitPacket} from "../emitKafkatorioPacket";
import {convertPrototypes} from "./convertPrototypes";
import {updateGlobal} from "./updateGlobal";
import {emitPrototypes} from "./prototypes";
export function handleConfigurationUpdate(changeData: ConfigurationChangedData) {
updateGlobal()
@ -9,10 +9,10 @@ export function handleConfigurationUpdate(changeData: ConfigurationChangedData)
modVersion: global.MOD_VERSION,
packetType: "CONFIG",
allMods: allMods(changeData),
factorioData: factorioData(changeData),
prototypes: convertPrototypes()
factorioData: factorioData(changeData)
})
emitPrototypes()
}
function factorioData(changeData: ConfigurationChangedData): FactorioGameDataUpdate {

11
modules/events-mod/src/main/typescript/main/config-update/convertPrototypes.ts → modules/events-mod/src/main/typescript/main/config-update/prototypes.ts
View file

@ -1,6 +1,15 @@
import {Converters} from "../events/converters";
import {emitPacket} from "../emitKafkatorioPacket";
export function convertPrototypes(): FactorioPrototype[] {
export function emitPrototypes() {
emitPacket<FactorioPrototypes>({
modVersion: global.MOD_VERSION,
packetType: "PROTOTYPES",
prototypes: prototypes()
})
}
function prototypes(): FactorioPrototype[] {
let prototypes: FactorioPrototype[] = []
prototypes.push(...getMapTilePrototypes())

7
modules/events-mod/src/main/typescript/main/events/converters.ts
View file

@ -114,4 +114,11 @@ export namespace Converters {
// }
// }
export function convertBoundingBox(bb: BoundingBoxTable): MapBoundingBox {
return {
topLeft: mapTilePosition(bb.left_top as PositionTable),
bottomRight: mapTilePosition(bb.right_bottom as PositionTable),
}
}
}

5
modules/events-mod/src/main/typescript/main/events/handlers.ts
View file

@ -1,6 +1,6 @@
import {Converters} from "./converters"
import {emitEvent} from "./emitEvent";
import mapChunkPosition = Converters.mapChunkPosition;
export function handlePlayerUpdate(tick: uint, eventType: string, playerIndex: uint) {
let player: LuaPlayer = game.players[playerIndex]
@ -80,7 +80,8 @@ export function handleChunkUpdate(
let mapChunk: MapChunk = {
objectName: "MapChunk",
tiles: createMapTiles(surfaceIndex, convertedTiles),
position: mapChunkPosition(position)
position: Converters.mapChunkPosition(position),
area: Converters.convertBoundingBox(area)
}
emitEvent(mapChunk, tick, eventType)

8
modules/events-mod/src/main/typescript/package-lock.json generated
View file

@ -1,6 +1,6 @@
{
"name": "kafkatorio-events-mod",
"version": "0.2.2",
"version": "0.2.4",
"lockfileVersion": 1,
"requires": true,
"dependencies": {
@ -74,9 +74,9 @@
"integrity": "sha512-GNzQvQTOIP6RyTfE2Qxb8ZVlNmw0n88vp1szwWRimP02mnTsx3Wtn5qRdqY9w2XduFNUgvOwhNnQsjwCp+kqaQ=="
},
"typed-factorio": {
"version": "0.15.0",
"resolved": "https://registry.npmjs.org/typed-factorio/-/typed-factorio-0.15.0.tgz",
"integrity": "sha512-hEpW+nRL3q5vg3aOxbNbYpwARxYzMTncihurAucwnpQ/IDc4w4PrO2/OWHI5R+PkZSEQUY5LSRxG7qrGqDeWHQ=="
"version": "0.16.0",
"resolved": "https://registry.npmjs.org/typed-factorio/-/typed-factorio-0.16.0.tgz",
"integrity": "sha512-I6W1lhRkeE+mHzLBKBVIDZdquwBjBYaS7jkV0CKiWWkA9ptGqT+MblOkNUdYZ0ykR2Rvn1eF/6pzh8WwQ+mYVQ=="
},
"typescript": {
"version": "4.5.4",

4
modules/events-mod/src/main/typescript/package.json
View file

@ -1,6 +1,6 @@
{
"name": "kafkatorio-events-mod",
"version": "0.2.2",
"version": "0.2.4",
"private": true,
"scripts": {
"build": "tstl",
@ -14,7 +14,7 @@
"lua-types": "^2.11.0",
"@types/node": "^12.0.8",
"typescript-to-lua": "1.3.0",
"typed-factorio": "0.15.0",
"typed-factorio": "0.16.0",
"typescript": "4.5.4"
},
"license": "UNLICENSED",

2
modules/events-schema/build.gradle.kts
View file

@ -20,7 +20,7 @@ kotlin {
}
}
@Suppress("UnstableApiUsage") // enforcedPlatform is incubating
@Suppress("UnstableApiUsage") // platform + version catalogs is incubating
sourceSets {
all {

25
modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioCommonData.kt
View file

@ -19,8 +19,8 @@ data class MapEntityPosition(
) {
fun toMapChunkPosition() = MapChunkPosition(
(x / MAP_CHUNK_SIZE).roundToInt(),
(y / MAP_CHUNK_SIZE).roundToInt(),
(x / MAP_CHUNK_SIZE).toInt(),
(y / MAP_CHUNK_SIZE).toInt(),
)
}
@ -36,10 +36,17 @@ data class MapChunkPosition(
val x: Int,
val y: Int,
) {
fun toMapEntityPosition() = MapEntityPosition(
x * MAP_CHUNK_SIZE.toDouble(),
y * MAP_CHUNK_SIZE.toDouble(),
)
fun toMapTilePosition() = MapChunkPosition(
x * MAP_CHUNK_SIZE,
y * MAP_CHUNK_SIZE,
)
}
/**
@ -62,6 +69,14 @@ data class MapTilePosition(
}
@Serializable
data class MapBoundingBox(
val topLeft: MapTilePosition,
val bottomRight: MapTilePosition,
)
/**
* Red, green, blue and alpha values, all in range `[0, 1]` or all in range `[0, 255]` if any
* value is > 1.
@ -81,7 +96,7 @@ data class Colour(
) {
/** True if any value is greater than 1, so the values are hexadecimal. */
fun isDecimal(): Boolean {
fun isHexadecimal(): Boolean {
return red > 1f
|| green > 1f
|| blue > 1f
@ -89,9 +104,9 @@ data class Colour(
}
/** True if all values are between `[0..1]`. */
fun isPercentage(): Boolean = !isDecimal()
fun isPercentage(): Boolean = !isHexadecimal()
fun asDecimal() = if (isPercentage()) {
fun asHexadecimal() = if (isPercentage()) {
Colour(
(red * 255).coerceIn(0f..255f),
(green * 255).coerceIn(0f..255f),

6
modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioConfiguration.kt
View file

@ -1,10 +1,7 @@
package dev.adamko.kafkatorio.events.schema
import kotlinx.serialization.EncodeDefault
import kotlinx.serialization.Required
import kotlinx.serialization.SerialName
import kotlinx.serialization.Serializable
import kotlinx.serialization.Transient
@Serializable
@ -12,10 +9,9 @@ data class FactorioConfigurationUpdate(
override val modVersion: String,
val factorioData: FactorioGameDataUpdate,
val allMods: List<FactorioModInfo>,
val prototypes: List<FactorioPrototype>,
) : KafkatorioPacket() {
@EncodeDefault
override val packetType = PacketType.CONFIG
override val packetType = PacketType.CONFIG
}
@Serializable

1
modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioLuaObjects.kt
View file

@ -119,6 +119,7 @@ data class SurfaceData(
data class MapChunk(
val tiles: MapTiles,
val position: MapChunkPosition,
val area: MapBoundingBox,
) : FactorioObjectData() {
@EncodeDefault
override val objectName = ObjectName.MapChunk

11
modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioPrototypes.kt
View file

@ -10,6 +10,16 @@ import kotlinx.serialization.json.jsonObject
import kotlinx.serialization.json.jsonPrimitive
@Serializable
data class FactorioPrototypes(
override val modVersion: String,
val prototypes: List<FactorioPrototype>,
) : KafkatorioPacket() {
@EncodeDefault
override val packetType = PacketType.PROTOTYPES
}
@Serializable(with = FactorioPrototypeSerializer::class)
sealed class FactorioPrototype {
@EncodeDefault
@ -49,6 +59,7 @@ data class MapTilePrototype(
val name: String,
val layer: UInt,
val mapColour: Colour,
@Serializable(with = FactorioJsonListSerializer::class)
val collisionMasks: List<String>,
val order: String,
/** Can the tile be mined for resources? */

29
modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/json.kt
View file

@ -10,8 +10,17 @@ import kotlinx.serialization.json.JsonTransformingSerializer
import kotlinx.serialization.json.jsonArray
import kotlinx.serialization.json.jsonObject
val jsonMapperKafkatorio = Json {
prettyPrint = true
prettyPrintIndent = " "
// serializersModule = FactorioEvent.kxsModule
// serializersModule = KafkatorioPacket.kxsModule + FactorioEvent.kxsModule
// serializersModule = SerializersModule { }
}
/** Factorio outputs lists as Json objects - this serializer converts an object back to a list. */
// actually I'm not sure that it does - I think *empty* lists get encoded as *empty* objects
internal class ListAsObjectSerializer<T>(dataSerializer: KSerializer<T>) :
JsonTransformingSerializer<List<T>>(ListSerializer(dataSerializer)) {
@ -26,10 +35,18 @@ internal class ListAsObjectSerializer<T>(dataSerializer: KSerializer<T>) :
)
}
val jsonMapperKafkatorio = Json {
prettyPrint = true
prettyPrintIndent = " "
// serializersModule = FactorioEvent.kxsModule
// serializersModule = KafkatorioPacket.kxsModule + FactorioEvent.kxsModule
// serializersModule = SerializersModule { }
/** Factorio outputs lists as Json objects - this serializer converts an object back to a list. */
// actually I'm not sure that it does - I think *empty* lists get encoded as *empty* objects
internal class FactorioJsonListSerializer<T>(dataSerializer: KSerializer<T>) :
JsonTransformingSerializer<List<T>>(ListSerializer(dataSerializer)) {
override fun transformDeserialize(element: JsonElement): JsonElement {
return if (element is JsonObject) {
JsonArray(emptyList())
} else {
element
}
}
}

2
modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/kafkatorioPacket.kt
View file

@ -20,6 +20,7 @@ sealed class KafkatorioPacket {
enum class PacketType {
EVENT,
CONFIG,
PROTOTYPES,
}
}
@ -41,6 +42,7 @@ object KafkatorioPacketSerializer : JsonContentPolymorphicSerializer<KafkatorioP
return when (type) {
KafkatorioPacket.PacketType.EVENT -> FactorioEvent.serializer()
KafkatorioPacket.PacketType.CONFIG -> FactorioConfigurationUpdate.serializer()
KafkatorioPacket.PacketType.PROTOTYPES -> FactorioPrototypes.serializer()
null ->
throw Exception("Unknown KafkatorioPacket ${key}: '$type' ")
}

45
modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/PrototypesTest.kt Normal file
View file

@ -0,0 +1,45 @@
package dev.adamko.kafkatorio.events.schema
import io.kotest.core.spec.style.BehaviorSpec
import kotlinx.serialization.decodeFromString
class PrototypesTest : BehaviorSpec(
{
Given("Prototypes Json") {
//Language=JSON
val raw = """
{
"modVersion":"0.2.4",
"packetType":"PROTOTYPES",
"prototypes":[
{
"prototypeObjectName":"LuaTilePrototype",
"name":"tile-unknown",
"order":"z-a",
"layer":0,
"collisionMasks":{
},
"mapColour":{
"red":0,
"green":0,
"blue":0,
"alpha":255
},
"canBeMined":false
}
]
}
""".trimIndent()
Then("expect can be parsed") {
val data = jsonMapperKafkatorio.decodeFromString<KafkatorioPacket>(raw)
println(data)
}
}
}
) {
companion object {
}
}

26
modules/events-schema/src/jvmMain/kotlin/dev/adamko/kafkatorio/events/schema/kt2ts.kt
View file

@ -16,6 +16,8 @@ fun main(args: Array<String>) {
addAll(KafkatorioPacket::class.sealedSubclasses)
addAll(FactorioObjectData::class.sealedSubclasses)
addAll(FactorioPrototype::class.sealedSubclasses)
add(FactorioPrototypes::class)
add(FactorioConfigurationUpdate::class)
},
mappings = mapOf(
// builtin Factorio numeric types > `typed-factorio/generated/builtin-types.d.ts`
@ -40,7 +42,6 @@ fun main(args: Array<String>) {
println(definitions)
}
else -> {
val outPath = Path(filename)
outPath.createDirectories()
@ -64,17 +65,18 @@ private fun splitDefinitions(definitions: String): Map<String, String> {
.split("\n\n")
.groupBy { def ->
when {
"extends FactorioObjectData" in def ||
"interface FactorioObjectData" in def ||
"type ObjectName" in def -> "object-data.d.ts"
"extends FactorioPrototype" in def ||
"interface FactorioPrototype" in def ||
"type PrototypeObjectName" in def -> "prototype.d.ts"
"extends FactorioConfigurationUpdate" in def ||
"interface FactorioConfigurationUpdate" in def ||
"FactorioGameDataUpdate" in def ||
"FactorioModInfo" in def -> "config-update.d.ts"
else -> "schema.d.ts"
"extends FactorioObjectData " in def ||
"interface FactorioObjectData " in def ||
"type ObjectName " in def -> "object-data.d.ts"
"extends FactorioPrototype " in def ||
"interface FactorioPrototypes " in def ||
"interface FactorioPrototype " in def ||
"type PrototypeObjectName " in def -> "prototype.d.ts"
"extends FactorioConfigurationUpdate " in def ||
"interface FactorioConfigurationUpdate " in def ||
"FactorioGameDataUpdate " in def ||
"FactorioModInfo " in def -> "config-update.d.ts"
else -> "schema.d.ts"
}
}
.mapValues { (_, def) ->

12
modules/events-server/build.gradle.kts
View file

@ -1,7 +1,7 @@
import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
plugins {
id("dev.adamko.kafkatorio.lang.kotlin-jvm")
dev.adamko.kafkatorio.lang.`kotlin-jvm`
kotlin("plugin.serialization")
application
}
@ -36,6 +36,16 @@ dependencies {
implementation(libs.bundles.logging)
implementation("com.sksamuel.scrimage:scrimage-core:4.0.24")
val kotkaVersion = "0.0.5"
implementation("com.github.adamko-dev.kotka-streams:kotka-streams:$kotkaVersion") {
isChanging = true
}
// val kotkaVersion = "main-SNAPSHOT"
// implementation("dev.adamko.kotka:kotka-streams:$kotkaVersion") {
// isChanging = true
// }
}
application {

78
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt Normal file
View file

@ -0,0 +1,78 @@
package dev.adamko.kafkatorio.admin
import dev.adamko.kafkatorio.events.schema.FactorioObjectData
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
import dev.adamko.kafkatorio.processor.KafkatorioTopology
import java.util.concurrent.TimeUnit
import mu.KotlinLogging
import org.apache.kafka.clients.admin.Admin
import org.apache.kafka.clients.admin.CreateTopicsResult
import org.apache.kafka.clients.admin.NewTopic
private val logger = KotlinLogging.logger {}
fun main() {
CreateTopics.create()
}
object CreateTopics {
private val kafkaAdmin: Admin =
Admin.create(
mapOf(
"bootstrap.servers" to "http://localhost:9092",
"client.id" to "warehouse.setup-admin",
)
)
fun create() {
KafkatorioPacket.PacketType.values().forEach { packetType ->
val kafkatorioTopics = buildList {
add(KafkatorioTopology.sourceTopic)
when (packetType) {
KafkatorioPacket.PacketType.EVENT ->
FactorioObjectData.ObjectName.values().forEach { objectName ->
add("kafkatorio.${packetType.name}.${objectName.name}")
}
KafkatorioPacket.PacketType.CONFIG ->
add("kafkatorio.${packetType.name}.FactorioConfigurationUpdate")
KafkatorioPacket.PacketType.PROTOTYPES ->
add("kafkatorio.${packetType.name}.all")
}
}
val result = kafkatorioTopics
.run {
logger.info("Creating $size topics")
kafkaAdmin.createTopics { this }
}
// wait for all brokers to become aware of the created topics
Thread.sleep(TimeUnit.SECONDS.toMillis(5))
result.values().forEach {
logger.info("Created topic: ${it.key} ${it.value.get()}")
}
Thread.sleep(TimeUnit.SECONDS.toMillis(5))
logger.info("Listing topics")
kafkaAdmin.listTopics()
.listings()
.get()
.forEach {
logger.info("Topic: $it")
}
}
}
private fun Admin.createTopics(
numPartitions: Int = 1,
replicationFactor: Short = 1,
topicNames: () -> Collection<String>,
): CreateTopicsResult =
createTopics(topicNames().map {
NewTopic(it, numPartitions, replicationFactor)
})
}

54
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/GlobalKTableDefinition.kt
View file

@ -2,30 +2,30 @@ package dev.adamko.kafkatorio.extensions.kafkastreams
import org.apache.kafka.streams.state.QueryableStoreTypes.timestampedKeyValueStore
/**
* * update topic [topicName]
* * plus table-source topic [tableName]
* * plus state store [globalStoreName]
*
* equals a GlobalTableDefinition
*/
abstract class GlobalKTableDefinition<K, V>(
val tableName: String,
override val serdes: KeyValueSerdes<K, V>,
val globalStoreName: String = "$tableName-store",
/**
* The backing topic, from which this table will be filled.
*
* Updates to items can be sent or received with this topic.
*/
final override val topicName: String = "$tableName-topic",
val storeType: TimestampedQueryStoreType<K, V> = timestampedKeyValueStore(),
) : TopicDefinition<K, V> {
val topicNames: Set<String>
init {
require(tableName != topicName)
require(tableName != globalStoreName)
topicNames = setOf(tableName, topicName)
}
}
///**
// * * update topic [topicName]
// * * plus table-source topic [tableName]
// * * plus state store [globalStoreName]
// *
// * equals a GlobalTableDefinition
// */
//abstract class GlobalKTableDefinition<K, V>(
// val tableName: String,
// override val serdes: KeyValueSerdes<K, V>,
// val globalStoreName: String = "$tableName-store",
// /**
// * The backing topic, from which this table will be filled.
// *
// * Updates to items can be sent or received with this topic.
// */
// final override val topicName: String = "$tableName-topic",
// val storeType: TimestampedQueryStoreType<K, V> = timestampedKeyValueStore(),
//) : TopicDefinition<K, V> {
// val topicNames: Set<String>
//
// init {
// require(tableName != topicName)
// require(tableName != globalStoreName)
// topicNames = setOf(tableName, topicName)
// }
//}

32
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KGroupedStream.kt Normal file
View file

@ -0,0 +1,32 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
//
//import org.apache.kafka.common.utils.Bytes
//import org.apache.kafka.streams.kstream.KGroupedStream
//import org.apache.kafka.streams.kstream.KTable
//import org.apache.kafka.streams.kstream.Materialized
//import org.apache.kafka.streams.state.KeyValueStore
//
//
//fun <K> KGroupedStream<K, *>.count(
// name: String? = null,
// materialized: Materialized<K, Long, KeyValueStore<Bytes, ByteArray>>? = null,
//): KTable<K, Long> {
// return when {
// name != null && materialized != null -> count(withName(name), materialized)
// name == null && materialized != null -> count(materialized)
// name != null && materialized == null -> count(withName(name))
// else -> count()
// }
//}
//
//
//
////<VOut> CogroupedKStream<K, VOut> cogroup(final Aggregator<? super K, ? super V, VOut> aggregator);
////<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,final Aggregator<? super K, ? super V, VR> aggregator, final Named named, final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<W extends Window> TimeWindowedKStream<K, V> windowedBy(final Windows<W> windows);
////KTable<K, Long> count(final Materialized<K, Long, KeyValueStore<Bytes, byte[]>> materialized);
////KTable<K, V> reduce(final Reducer<V> reducer);
////KTable<K, V> reduce(final Reducer<V> reducer, final Named named, final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
////KTable<K, V> reduce(final Reducer<V> reducer,final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
////SessionWindowedKStream<K, V> windowedBy(final SessionWindows windows);
////TimeWindowedKStream<K, V> windowedBy(final SlidingWindows windows);

251
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamExtensions.kt
View file

@ -1,111 +1,152 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
import dev.adamko.kafkatorio.extensions.kafkastreams.KStreamExtensions.flatMap
import dev.adamko.kafkatorio.extensions.kafkastreams.KStreamExtensions.groupBy
import dev.adamko.kafkatorio.extensions.kafkastreams.KStreamExtensions.map
import dev.adamko.kafkatorio.extensions.kafkastreams.KStreamExtensions.mapValues
import dev.adamko.kafkatorio.extensions.kafkastreams.KafkaStreamsExtensions.withName
import dev.adamko.kafkatorio.extensions.kafkastreams.KeyPairExtensions.kvPair
import dev.adamko.kafkatorio.extensions.kafkastreams.KeyPairExtensions.toKeyValue
import org.apache.kafka.streams.KeyValue
import org.apache.kafka.streams.kstream.Grouped
import org.apache.kafka.streams.kstream.KGroupedStream
import org.apache.kafka.streams.kstream.KStream
import org.apache.kafka.streams.kstream.Transformer
import org.apache.kafka.streams.processor.ProcessorContext
object KStreamExtensions {
// /** @see KStream.map */
// inline fun <inK, inV, reified outK, reified outV : TopicData<outK>> KStream<inK, inV>.map(
// name: String,
// crossinline mapper: (key: inK, value: inV) -> outV
// ): KStream<outK, outV> =
// map({ k, v -> mapper(k, v).asKeyValue() }, withName(name))
/** @see KStream.map */
inline fun <inK, inV, reified outK, reified outV> KStream<inK, inV>.map(
name: String,
crossinline mapper: (key: inK, value: inV) -> Pair<outK, outV>
): KStream<outK, outV> =
map({ k, v -> mapper(k, v).toKeyValue() }, withName(name))
/** Wrapper for [KStream.mapValues] */
fun <K, inV, outV> KStream<K, inV>.mapValues(
name: String,
mapper: (key: K, value: inV) -> outV
): KStream<K, outV> =
mapValues({ k, v -> mapper(k, v) }, withName(name))
// /** @see KStream.flatMap */
// inline fun <inK, reified outK, inV, reified outV : TopicData<outK>> KStream<inK, inV>.flatMap(
// name: String,
// crossinline mapper: (key: inK, value: inV) -> Iterable<outV>
// ): KStream<outK, outV> = flatMap(
// { k, v -> mapper(k, v).map { a: outV -> a.asKeyValue() } },
// withName(name)
// )
/** @see KStream.flatMap */
inline fun <reified inK, inV, reified outK, reified outV> KStream<inK, inV>.flatMap(
name: String,
crossinline mapper: (key: inK, value: inV) -> Iterable<Pair<outK, outV>>
): KStream<outK, outV> = flatMap(
{ k, v -> mapper(k, v).map { it.toKeyValue() } },
withName(name)
)
/** @see KStream.groupBy */
inline fun <reified inK, outK, reified V> KStream<inK, V>.groupBy(
grouped: Grouped<outK, V>,
crossinline keySelector: KeyValue<inK, V>.() -> outK
): KGroupedStream<outK, V> =
groupBy(
{ k, v -> (k kvPair v).keySelector() },
grouped
)
// @JvmName("flatMapKeyValuePairs")
// inline fun <inK, reified outK, inV, reified outV> KStream<inK, inV>.flatMap(
// name: String,
// crossinline mapper: (key: inK, value: inV) -> Iterable<Pair<outK, outV>>
// ): KStream<outK, outV> =
// flatMap(
// { k, v -> mapper(k, v).map { it.asKeyPair() } },
// withName(name)
// )
// inline fun <reified K, inV, reified outV> KStream<K, inV>.mapValues(
// name: String,
// crossinline mapper: (readOnlyKey: K, value: inV) -> outV
// ): KStream<K, outV> =
// map({ k, v -> (k to mapper(v)).asKeyPair() }, KafkaStreamsExtensions.withName(name))
}
//open class TransformerKt<inK, inV, outK, outV>(
//) : Transformer<inK, inV, KeyValue<outK, outV>> {
//
// val onClose: TransformerKt<inK, inV, outK, outV>.() -> Unit = {}
// val transformer: TransformerKt<inK, inV, outK, outV>.(key: inK, value: inV) -> KeyValue<outK, outV>? =
// {}
//import org.apache.kafka.streams.kstream.Grouped
//import org.apache.kafka.streams.kstream.KGroupedStream
//import org.apache.kafka.streams.kstream.KStream
//import org.apache.kafka.streams.kstream.Produced
//import org.apache.kafka.streams.processor.TopicNameExtractor
//
// private lateinit var _context: ProcessorContext
// val context: ProcessorContext?
// get() = if (this::_context.isInitialized) _context else null
//// /** @see KStream.map */
//// inline fun <inK, inV, reified outK, reified outV : TopicData<outK>> KStream<inK, inV>.map(
//// name: String,
//// crossinline mapper: (key: inK, value: inV) -> outV
//// ): KStream<outK, outV> =
//// map({ k, v -> mapper(k, v).asKeyValue() }, withName(name))
//
// override fun init(context: ProcessorContext) {
// this._context = context
// }
///** @see KStream.map */
//inline fun <inK, inV, reified outK, reified outV> KStream<inK, inV>.map(
// name: String,
// crossinline mapper: (key: inK, value: inV) -> Pair<outK, outV>
//): KStream<outK, outV> =
// map({ k, v -> mapper(k, v).toKeyValue() }, withName(name))
//
// override fun close() {
// onClose()
// }
///** @see [KStream.mapValues] */
//inline fun <K, inV, outV> KStream<K, inV>.mapValues(
// name: String,
// crossinline mapper: (key: K, value: inV) -> outV
//): KStream<K, outV> = mapValues({ k, v -> mapper(k, v) }, withName(name))
//
// override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
// return transformer(key, value)
// }
//}
//// /** @see KStream.flatMap */
//// inline fun <inK, reified outK, inV, reified outV : TopicData<outK>> KStream<inK, inV>.flatMap(
//// name: String,
//// crossinline mapper: (key: inK, value: inV) -> Iterable<outV>
//// ): KStream<outK, outV> = flatMap(
//// { k, v -> mapper(k, v).map { a: outV -> a.toKeyValue() } },
//// withName(name)
//// )
//
///** @see KStream.flatMap */
//inline fun <reified inK, inV, reified outK, reified outV> KStream<inK, inV>.flatMap(
// name: String,
// crossinline mapper: (key: inK, value: inV) -> Iterable<Pair<outK, outV>>
//): KStream<outK, outV> = flatMap(
// { k, v -> mapper(k, v).map { it.toKeyValue() } },
// withName(name)
//)
//
///** @see KStream.groupBy */
//fun <K, V, outK> KStream<K, V>.groupBy(
// grouped: Grouped<outK, V>,
// keySelector: (K, V) -> outK
//): KGroupedStream<outK, V> = groupBy(keySelector, grouped)
//
//fun <K, V> KStream<K, V>.to(
// produced: Produced<K, V>,
// topicNameExtractor: TopicNameExtractor<K, V>,
//) = to(topicNameExtractor, produced)
//
//fun <K, V> KStream<K, V>.filter(
// name: String,
// predicate: (K, V) -> Boolean,
//): KStream<K, V> = filter(predicate, withName(name))
//
//
//fun <K, V> KStream<K, V>.filterNot(
// name: String,
// predicate: (K, V) -> Boolean,
//): KStream<K, V> = filterNot(predicate, withName(name))
//
//
//fun <K, V> KStream<K, V>.merge(
// name: String,
// other: KStream<K, V>,
//): KStream<K, V> = merge(other, withName(name))
//
//
//// KStream<K, V> filter(final Predicate<? super K, ? super V> predicate, final Named named);
//// KStream<K, V> filterNot(final Predicate<? super K, ? super V> predicate, final Named named);
//
//// @JvmName("flatMapKeyValuePairs")
//// inline fun <inK, reified outK, inV, reified outV> KStream<inK, inV>.flatMap(
//// name: String,
//// crossinline mapper: (key: inK, value: inV) -> Iterable<Pair<outK, outV>>
//// ): KStream<outK, outV> =
//// flatMap(
//// { k, v -> mapper(k, v).map { it.asKeyPair() } },
//// withName(name)
//// )
//
//// inline fun <reified K, inV, reified outV> KStream<K, inV>.mapValues(
//// name: String,
//// crossinline mapper: (readOnlyKey: K, value: inV) -> outV
//// ): KStream<K, outV> =
//// map({ k, v -> (k to mapper(v)).asKeyPair() }, KafkaStreamsExtensions.withName(name))
//
//
////open class TransformerKt<inK, inV, outK, outV>(
////) : Transformer<inK, inV, KeyValue<outK, outV>> {
////
//// val onClose: TransformerKt<inK, inV, outK, outV>.() -> Unit = {}
//// val transformer: TransformerKt<inK, inV, outK, outV>.(key: inK, value: inV) -> KeyValue<outK, outV>? =
//// {}
////
//// private lateinit var _context: ProcessorContext
//// val context: ProcessorContext?
//// get() = if (this::_context.isInitialized) _context else null
////
//// override fun init(context: ProcessorContext) {
//// this._context = context
//// }
////
//// override fun close() {
//// onClose()
//// }
////
//// override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
//// return transformer(key, value)
//// }
////}
////
////
//// /** one or the other or both */
//// KTable<K, V> toTable(final Named named, Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
//// <GK, GV, RV> KStream<K, RV> join(final GlobalKTable<GK, GV> globalTable, KeyValueMapper<? super K, ? super V, ? extends GK> keySelector, ValueJoinerWithKey<? super K, ? super V, ? super GV, ? extends RV> joiner, Named named);
//// <GK, GV, RV> KStream<K, RV> leftJoin(final GlobalKTable<GK, GV> globalTable, KeyValueMapper<? super K, ? super V, ? extends GK> keySelector, ValueJoinerWithKey<? super K, ? super V, ? super GV, ? extends RV> valueJoiner, Named named);
//// <K1, V1> KStream<K1, V1> flatTransform(final TransformerSupplier<? super K, ? super V, Iterable<KeyValue<K1, V1>>> transformerSupplier, Named named, String... stateStoreNames);
//// <K1, V1> KStream<K1, V1> flatTransform(final TransformerSupplier<? super K, ? super V, Iterable<KeyValue<K1, V1>>> transformerSupplier, String... stateStoreNames);
//// <K1, V1> KStream<K1, V1> transform(final TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, Named named, String... stateStoreNames);
//// <K1, V1> KStream<K1, V1> transform(final TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier, String... stateStoreNames);
//// <KR, VR> KStream<KR, VR> flatMap(final KeyValueMapper<? super K, ? super V, ? extends Iterable<? extends KeyValue<? extends KR, ? extends VR>>> mapper, Named named);
//// <KR, VR> KStream<KR, VR> map(final KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> mapper, Named named);
//// <KR> KGroupedStream<KR, V> groupBy(final KeyValueMapper<? super K, ? super V, KR> keySelector, Grouped<KR, V> grouped);
//// <KR> KStream<KR, V> selectKey(final KeyValueMapper<? super K, ? super V, ? extends KR> mapper, Named named);
//// <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream, ValueJoinerWithKey<? super K, ? super V, ? super VO, ? extends VR> joiner, JoinWindows windows, StreamJoined<K, V, VO> streamJoined);
//// <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream, ValueJoinerWithKey<? super K, ? super V, ? super VO, ? extends VR> joiner, JoinWindows windows, StreamJoined<K, V, VO> streamJoined);
//// <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream, ValueJoinerWithKey<? super K, ? super V, ? super VO, ? extends VR> joiner, JoinWindows windows, StreamJoined<K, V, VO> streamJoined);
//// <VR> KStream<K, VR> flatMapValues(final ValueMapperWithKey<? super K, ? super V, ? extends Iterable<? extends VR>> mapper, Named named);
//// <VR> KStream<K, VR> flatTransformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, Iterable<VR>> valueTransformerSupplier, Named named, String... stateStoreNames);
//// <VR> KStream<K, VR> mapValues(final ValueMapperWithKey<? super K, ? super V, ? extends VR> mapper, Named named);
//// <VR> KStream<K, VR> transformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> valueTransformerSupplier, Named named, String... stateStoreNames);
//// <VT, VR> KStream<K, VR> join(final KTable<K, VT> table, ValueJoinerWithKey<? super K, ? super V, ? super VT, ? extends VR> joiner, Joined<K, V, VT> joined);
//// <VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table, ValueJoinerWithKey<? super K, ? super V, ? super VT, ? extends VR> joiner, Joined<K, V, VT> joined);
//// BranchedKStream<K, V> split(final Named named);
//// KGroupedStream<K, V> groupByKey(final Grouped<K, V> grouped);
//// KStream<K, V> merge(final KStream<K, V> stream, final Named named);
//// KStream<K, V> peek(final ForeachAction<? super K, ? super V> action, final Named named);
//// KStream<K, V> repartition(final Repartitioned<K, V> repartitioned);
//// void foreach(final ForeachAction<? super K, ? super V> action, final Named named);
//// void print(final Printed<K, V> printed);
//// void process(final ProcessorSupplier<? super K, ? super V, Void, Void> processorSupplier, Named named, String... stateStoreNames);
//// void to(final String topic, Produced<K, V> produced);
//// void to(final TopicNameExtractor<K, V> topicExtractor, Produced<K, V> produced);
////

178
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamTransform.kt
View file

@ -1,90 +1,90 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
import dev.adamko.kafkatorio.extensions.kafkastreams.KeyPairExtensions.toKeyValue
import kotlin.experimental.ExperimentalTypeInference
import org.apache.kafka.streams.KeyValue
import org.apache.kafka.streams.kstream.KStream
import org.apache.kafka.streams.kstream.Transformer
import org.apache.kafka.streams.kstream.TransformerSupplier
import org.apache.kafka.streams.processor.ProcessorContext
/** @see KStream.transform */
inline fun <inK, inV, reified outK, reified outV> KStream<inK, inV>.transformKt(
crossinline block: BuildTransformer<inK, inV, outK, outV>
): KStream<outK, outV> {
val transformer = object : Transformer<inK, inV, KeyValue<outK, outV>> {
private lateinit var scope: TransformerScope<inK, inV, outK, outV>
override fun init(procContext: ProcessorContext) {
scope = TransformerScope(procContext)
scope.block()
}
override fun close() {
scope.close()
}
override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
return scope.mapper(key, value)?.toKeyValue()
}
}
return transform({ transformer })
}
/** @see KStream.transform */
@OptIn(ExperimentalTypeInference::class)
inline fun <inK, inV, reified outK, reified outV, outKV : KeyValue<outK, outV>> KStream<inK, inV>.transformKt2(
@BuilderInference block: Transformer<inK, inV, outKV>.() -> Unit
): KStream<outK, outV> {
val transformer = object : Transformer<inK, inV, KeyValue<outK, outV>> {
private lateinit var scope: TransformerScope<inK, inV, outK, outV>
override fun init(procContext: ProcessorContext) {
scope = TransformerScope(procContext)
}
override fun close() {
scope.close()
sequence<Long> {
}
}
override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
return scope.mapper(key, value)?.toKeyValue()
}
}
return transform(TransformerSupplier { transformer })
}
typealias BuildTransformer<inK, inV, outK, outV> = TransformerScope<inK, inV, outK, outV>.() -> Unit
typealias TransformerMapper<inK, inV, outK, outV> = (key: inK, value: inV) -> Pair<outK, outV>?
typealias TransformerCloseHandler = (ProcessorContext) -> Unit
open class TransformerScope<inK, inV, outK, outV>(
val processorContext: ProcessorContext,
) : AutoCloseable {
private var closeHandler: TransformerCloseHandler = {}
lateinit var mapper: TransformerMapper<inK, inV, outK, outV>
fun handleClose(closeHandler: TransformerCloseHandler) {
this.closeHandler = closeHandler
}
fun mapEach(mapper: TransformerMapper<inK, inV, outK, outV>) {
this.mapper = mapper
}
override fun close() {
closeHandler(processorContext)
}
}
//
//import dev.adamko.kafkatorio.extensions.kafkastreams.*
//import kotlin.experimental.ExperimentalTypeInference
//import org.apache.kafka.streams.KeyValue
//import org.apache.kafka.streams.kstream.KStream
//import org.apache.kafka.streams.kstream.Transformer
//import org.apache.kafka.streams.kstream.TransformerSupplier
//import org.apache.kafka.streams.processor.ProcessorContext
//
///** @see KStream.transform */
//inline fun <inK, inV, reified outK, reified outV> KStream<inK, inV>.transformKt(
// crossinline block: BuildTransformer<inK, inV, outK, outV>
//): KStream<outK, outV> {
//
// val transformer = object : Transformer<inK, inV, KeyValue<outK, outV>> {
//
// private lateinit var scope: TransformerScope<inK, inV, outK, outV>
//
// override fun init(procContext: ProcessorContext) {
// scope = TransformerScope(procContext)
// scope.block()
// }
//
// override fun close() {
// scope.close()
// }
//
// override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
// return scope.mapper(key, value)?.toKeyValue()
// }
// }
//
// return transform({ transformer })
//}
//
///** @see KStream.transform */
//@OptIn(ExperimentalTypeInference::class)
//inline fun <inK, inV, reified outK, reified outV, outKV : KeyValue<outK, outV>> KStream<inK, inV>.transformKt2(
// @BuilderInference block: Transformer<inK, inV, outKV>.() -> Unit
//): KStream<outK, outV> {
//
// val transformer = object : Transformer<inK, inV, KeyValue<outK, outV>> {
//
// private lateinit var scope: TransformerScope<inK, inV, outK, outV>
//
// override fun init(procContext: ProcessorContext) {
// scope = TransformerScope(procContext)
// }
//
// override fun close() {
// scope.close()
//
// sequence<Long> {
//
// }
// }
//
// override fun transform(key: inK, value: inV): KeyValue<outK, outV>? {
// return scope.mapper(key, value)?.toKeyValue()
// }
// }
//
// return transform(TransformerSupplier { transformer })
//}
//
//typealias BuildTransformer<inK, inV, outK, outV> = TransformerScope<inK, inV, outK, outV>.() -> Unit
//
//typealias TransformerMapper<inK, inV, outK, outV> = (key: inK, value: inV) -> Pair<outK, outV>?
//typealias TransformerCloseHandler = (ProcessorContext) -> Unit
//
//open class TransformerScope<inK, inV, outK, outV>(
// val processorContext: ProcessorContext,
//) : AutoCloseable {
//
// private var closeHandler: TransformerCloseHandler = {}
// lateinit var mapper: TransformerMapper<inK, inV, outK, outV>
//
// fun handleClose(closeHandler: TransformerCloseHandler) {
// this.closeHandler = closeHandler
// }
//
// fun mapEach(mapper: TransformerMapper<inK, inV, outK, outV>) {
// this.mapper = mapper
// }
//
// override fun close() {
// closeHandler(processorContext)
// }
//}

65
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KTableExtensions.kt
View file

@ -1,19 +1,48 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
import dev.adamko.kafkatorio.extensions.kafkastreams.KTableExtensions.mapValues
import dev.adamko.kafkatorio.extensions.kafkastreams.KafkaStreamsExtensions.withName
import org.apache.kafka.streams.kstream.KTable
object KTableExtensions {
/** @see [KTable.mapValues] */
fun <K, inV, outV> KTable<K, inV>.mapValues(
name: String,
mapper: (readOnlyKey: K, value: inV) -> outV
): KTable<K, outV> =
mapValues(
{ k, v -> mapper(k, v) },
withName(name)
)
}
//
//
//import org.apache.kafka.streams.KeyValue
//import org.apache.kafka.streams.kstream.Grouped
//import org.apache.kafka.streams.kstream.KGroupedTable
//import org.apache.kafka.streams.kstream.KTable
//import org.apache.kafka.streams.kstream.KeyValueMapper
//
///** @see [KTable.mapValues] */
//fun <K, inV, outV> KTable<K, inV>.mapValues(
// name: String,
// mapper: (readOnlyKey: K, value: inV) -> outV
//): KTable<K, outV> =
// mapValues(
// { k, v -> mapper(k, v) },
// withName(name)
// )
//
///** @See [KTable.groupBy] */
//fun <inK, inV, outK, outV> KTable<inK, inV>.groupBy(
// grouped: Grouped<outK, outV>? = null,
// selector: KeyValueMapper<inK, inV, KeyValue<outK, outV>>
//): KGroupedTable<outK, outV> = when (grouped) {
// null -> groupBy(selector)
// else -> groupBy(selector, grouped)
//}
//
//
////
////<KR> KStream<KR, V> toStream(final KeyValueMapper<? super K, ? super V, ? extends KR> mapper,final Named named);
////<VO, VR> KTable<K, VR> join(final KTable<K, VO> other,final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<VR, KO, VO> KTable<K, VR> join(final KTable<KO, VO> other,final Function<V, KO> foreignKeyExtractor,final ValueJoiner<V, VO, VR> joiner,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////
////<VO, VR> KTable<K, VR> leftJoin(final KTable<K, VO> other,final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////
////<VO, VR> KTable<K, VR> outerJoin(final KTable<K, VO> other,final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<VR, KO, VO> KTable<K, VR> leftJoin(final KTable<KO, VO> other,final Function<V, KO> foreignKeyExtractor,final ValueJoiner<V, VO, VR> joiner,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<VR> KTable<K, VR> mapValues(final ValueMapper<? super V, ? extends VR> mapper,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<VR> KTable<K, VR> mapValues(final ValueMapperWithKey<? super K, ? super V, ? extends VR> mapper,final Named named,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
////<VR> KTable<K, VR> transformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> transformerSupplier,final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized,final Named named,final String... stateStoreNames);
////KStream<K, V> toStream(final Named named);
/////** materialized, named - optional */
////KTable<K, V> filter(final Predicate<? super K, ? super V> predicate,final Named named,final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
/////** materialized, named - optional */
////KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate,final Named named,final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
////KTable<K, V> suppress(final Suppressed<? super K> suppressed);
////String queryableStoreName();

182
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KafkaStreamsExtensions.kt
View file

@ -1,93 +1,91 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
import org.apache.kafka.common.serialization.Serde
import org.apache.kafka.streams.kstream.Grouped
import org.apache.kafka.streams.kstream.Joined
import org.apache.kafka.streams.kstream.Materialized
import org.apache.kafka.streams.kstream.Named
import org.apache.kafka.streams.kstream.Produced
import org.apache.kafka.streams.kstream.Repartitioned
import org.apache.kafka.streams.processor.StateStore
/** Helpful extensions & shortcuts for Kafka Streams */
object KafkaStreamsExtensions {
/** @see [Named] */
fun withName(name: String): Named = Named.`as`(name)
// @Suppress("FunctionName")
// inline fun <reified K, reified V, S : StateStore> MaterializedAs(
// tableDefinition: GlobalTableDefinition<K, V>
// ) = MaterializedAs<K, V, S>(
// tableDefinition.globalStoreName,
// tableDefinition.serdes.keySerde,
// tableDefinition.serdes.valueSerde
// )
/** @see Materialized */
inline fun <reified K, reified V, S : StateStore> materializedAs(
name: String,
keySerde: Serde<K>,//= kxsJsonSerde(),
valueSerde: Serde<V>,//= kxsJsonSerde(),
): Materialized<K, V, S> =
Materialized.`as`<K, V, S>(name)
.withKeySerde(keySerde)
.withValueSerde(valueSerde)
/** @see Repartitioned */
inline fun <reified Key, reified Val> repartitionedAs(
name: String,
keySerde: Serde<Key>,
valueSerde: Serde<Val>,
): Repartitioned<Key, Val> =
Repartitioned.`as`<Key, Val>(name)
.withKeySerde(keySerde)
.withValueSerde(valueSerde)
/** @see Produced */
fun <K, V> producedAs(name: String): Produced<K, V> =
Produced.`as`(name)
/** @see Produced */
inline fun <reified Key, reified Val> producedAs(
name: String,
keySerde: Serde<Key>,
valueSerde: Serde<Val>,
): Produced<Key, Val> =
producedAs<Key, Val>(name)
.withKeySerde(keySerde)
.withValueSerde(valueSerde)
// @Suppress("FunctionName")
// fun <Key, Val, V_other> JoinedAs(name: String): Joined<Key, Val, V_other> =
// Joined.`as`(name)
/** @see Joined */
inline fun <reified Key, reified Val, reified V_other> joinedAs(
name: String,
keySerde: Serde<Key>,// = kxsJsonSerde(),
valueSerde: Serde<Val>,//= kxsJsonSerde(),
otherValueSerde: Serde<V_other>?,//= kxsJsonSerde(),
): Joined<Key, Val, V_other> =
Joined.`as`<Key, Val, V_other>(name)
.withKeySerde(keySerde)
.withValueSerde(valueSerde)
.withOtherValueSerde(otherValueSerde)
/** @see [Grouped.`as`] */
inline fun <reified Key, reified Val> groupedAs(
name: String,
keySerde: Serde<Key>?, //= kxsJsonSerde(),
valueSerde: Serde<Val>?,// = kxsJsonSerde(),
): Grouped<Key, Val> =
Grouped.`as`<Key, Val>(name)
.withKeySerde(keySerde)
.withValueSerde(valueSerde)
// fun <K, V> InteractiveQueryService.getQueryableStore(
// tableDef: GlobalTableDefinition<K, V>
// ): Optional<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>> {
// return getQueryableStore(tableDef.globalStoreName, tableDef.storeType)
// }
}
//
//import org.apache.kafka.common.serialization.Serde
//import org.apache.kafka.streams.kstream.Grouped
//import org.apache.kafka.streams.kstream.Joined
//import org.apache.kafka.streams.kstream.Materialized
//import org.apache.kafka.streams.kstream.Named
//import org.apache.kafka.streams.kstream.Produced
//import org.apache.kafka.streams.kstream.Repartitioned
//import org.apache.kafka.streams.processor.StateStore
//
///** Helpful extensions & shortcuts for Kafka Streams */
//
//
///** @see [Named] */
//fun withName(name: String): Named = Named.`as`(name)
//
//// @Suppress("FunctionName")
//// inline fun <reified K, reified V, S : StateStore> MaterializedAs(
//// tableDefinition: GlobalTableDefinition<K, V>
//// ) = MaterializedAs<K, V, S>(
//// tableDefinition.globalStoreName,
//// tableDefinition.serdes.keySerde,
//// tableDefinition.serdes.valueSerde
//// )
//
///** @see Materialized */
//inline fun <reified K, reified V, S : StateStore> materializedAs(
// name: String,
// keySerde: Serde<K>,//= kxsJsonSerde(),
// valueSerde: Serde<V>,//= kxsJsonSerde(),
//): Materialized<K, V, S> =
// Materialized.`as`<K, V, S>(name)
// .withKeySerde(keySerde)
// .withValueSerde(valueSerde)
//
///** @see Repartitioned */
//inline fun <reified Key, reified Val> repartitionedAs(
// name: String,
// keySerde: Serde<Key>,
// valueSerde: Serde<Val>,
//): Repartitioned<Key, Val> =
// Repartitioned.`as`<Key, Val>(name)
// .withKeySerde(keySerde)
// .withValueSerde(valueSerde)
//
///** @see Produced */
//fun <K, V> producedAs(name: String): Produced<K, V> =
// Produced.`as`(name)
//
///** @see Produced */
//inline fun <reified Key, reified Val> producedAs(
// name: String,
// keySerde: Serde<Key>,
// valueSerde: Serde<Val>,
//): Produced<Key, Val> =
// producedAs<Key, Val>(name)
// .withKeySerde(keySerde)
// .withValueSerde(valueSerde)
//
//// @Suppress("FunctionName")
//// fun <Key, Val, V_other> JoinedAs(name: String): Joined<Key, Val, V_other> =
//// Joined.`as`(name)
//
///** @see Joined */
//inline fun <reified Key, reified Val, reified V_other> joinedAs(
// name: String,
// keySerde: Serde<Key>,// = kxsJsonSerde(),
// valueSerde: Serde<Val>,//= kxsJsonSerde(),
// otherValueSerde: Serde<V_other>?,//= kxsJsonSerde(),
//): Joined<Key, Val, V_other> =
// Joined.`as`<Key, Val, V_other>(name)
// .withKeySerde(keySerde)
// .withValueSerde(valueSerde)
// .withOtherValueSerde(otherValueSerde)
//
///** @see [Grouped.`as`] */
//inline fun <reified Key, reified Val> groupedAs(
// name: String,
// keySerde: Serde<Key>?, //= jsonMapper.serde(),
// valueSerde: Serde<Val>?, //= jsonMapper.serde(),
//): Grouped<Key, Val> =
// Grouped.`as`<Key, Val>(name)
// .withKeySerde(keySerde)
// .withValueSerde(valueSerde)
//
//// fun <K, V> InteractiveQueryService.getQueryableStore(
//// tableDef: GlobalTableDefinition<K, V>
//// ): Optional<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>> {
//// return getQueryableStore(tableDef.globalStoreName, tableDef.storeType)
//// }

66
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyPairExtensions.kt
View file

@ -1,35 +1,33 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
import org.apache.kafka.streams.KeyValue
import org.apache.kafka.streams.state.ValueAndTimestamp
object KeyPairExtensions {
inline fun <reified K, reified V> Pair<K, V>.toKeyValue(): KeyValue<K, V> =
KeyValue.pair(first, second)
infix fun <K, V> K.kvPair(value: V): KeyValue<K, V> = KeyValue(this, value)
infix fun <V> V.timestamped(timestamp: Long): ValueAndTimestamp<V> =
ValueAndTimestamp.make(this, timestamp)
@JvmName("keyValueGetKey")
operator fun <K, V> KeyValue<K, V>.component1(): K = key
@JvmName("keyValueGetValue")
operator fun <K, V> KeyValue<K, V>.component2(): V = value
@JvmName("keyValueTimestampGetKey")
operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component1(): K = key
@JvmName("keyValueTimestampGetValue")
operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component2(): V = value.value()
@JvmName("keyValueTimestampGetTimestamp")
operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component3(): Long = value.timestamp()
@JvmName("valueTimestampGetValue")
operator fun <V> ValueAndTimestamp<V>.component1(): V = value()
@JvmName("valueTimestampGetTimestamp")
operator fun <V> ValueAndTimestamp<V>.component2(): Long = timestamp()
}
//
//import org.apache.kafka.streams.KeyValue
//import org.apache.kafka.streams.state.ValueAndTimestamp
//
//
//inline fun <reified K, reified V> Pair<K, V>.toKeyValue(): KeyValue<K, V> =
// KeyValue.pair(first, second)
//
//infix fun <K, V> K.kvPair(value: V): KeyValue<K, V> = KeyValue(this, value)
//
//infix fun <V> V.timestamped(timestamp: Long): ValueAndTimestamp<V> =
// ValueAndTimestamp.make(this, timestamp)
//
//
//@JvmName("keyValueGetKey")
//operator fun <K, V> KeyValue<K, V>.component1(): K = key
//@JvmName("keyValueGetValue")
//operator fun <K, V> KeyValue<K, V>.component2(): V = value
//
//
//@JvmName("keyValueTimestampGetKey")
//operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component1(): K = key
//@JvmName("keyValueTimestampGetValue")
//operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component2(): V = value.value()
//@JvmName("keyValueTimestampGetTimestamp")
//operator fun <K, V> KeyValue<K, ValueAndTimestamp<V>>.component3(): Long = value.timestamp()
//
//
//@JvmName("valueTimestampGetValue")
//operator fun <V> ValueAndTimestamp<V>.component1(): V = value()
//@JvmName("valueTimestampGetTimestamp")
//operator fun <V> ValueAndTimestamp<V>.component2(): Long = timestamp()

77
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyValueSerdes.kt
View file

@ -1,40 +1,41 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
import dev.adamko.kafkatorio.extensions.kafkastreams.KotlinxSerializationExtensions.kxsJsonSerde
import org.apache.kafka.common.serialization.Serde
import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.streams.kstream.Consumed
import org.apache.kafka.streams.kstream.Produced
import org.apache.kafka.streams.kstream.Repartitioned
/**
* A wrapper for two [Serde]s - [one for the Key][keySerde], [another for the Value][valueSerde].
*/
open class KeyValueSerdes<K, V>(
val keySerde: Serde<K>,
val valueSerde: Serde<V>,
) {
fun consumer(name: String): Consumed<K, V> =
Consumed.with(keySerde, valueSerde).withName(name)
fun producer(name: String): Produced<K, V> =
Produced.with(keySerde, valueSerde).withName(name)
fun repartitioned(name: String): Repartitioned<K, V> =
Repartitioned.with(keySerde, valueSerde).withName(name)
companion object {
inline fun <reified K, reified V> kxsJson(): KeyValueSerdes<K, V> =
KeyValueSerdes(
kxsJsonSerde(),
kxsJsonSerde(),
)
fun string(): KeyValueSerdes<String, String> =
KeyValueSerdes<String, String>(
Serdes.String(),
Serdes.String(),
)
}
}
//import dev.adamko.kafkatorio.extensions.kxs.KotlinxSerializationExtensions.serde
//import dev.adamko.kafkatorio.processor.jsonMapper
//import org.apache.kafka.common.serialization.Serde
//import org.apache.kafka.common.serialization.Serdes
//import org.apache.kafka.streams.kstream.Consumed
//import org.apache.kafka.streams.kstream.Produced
//import org.apache.kafka.streams.kstream.Repartitioned
//
///**
// * A wrapper for two [Serde]s - [one for the Key][keySerde], [another for the Value][valueSerde].
// */
//open class KeyValueSerdes<K, V>(
// val keySerde: Serde<K>,
// val valueSerde: Serde<V>,
//) {
//
// fun consumer(name: String): Consumed<K, V> =
// Consumed.with(keySerde, valueSerde).withName(name)
//
// fun producer(name: String): Produced<K, V> =
// Produced.with(keySerde, valueSerde).withName(name)
//
// fun repartitioned(name: String): Repartitioned<K, V> =
// Repartitioned.with(keySerde, valueSerde).withName(name)
//
// companion object {
// inline fun <reified K, reified V> kxsJson(): KeyValueSerdes<K, V> =
// KeyValueSerdes(
// jsonMapper.serde(),
// jsonMapper.serde(),
// )
//
// fun string(): KeyValueSerdes<String, String> =
// KeyValueSerdes<String, String>(
// Serdes.String(),
// Serdes.String(),
// )
// }
//}

55
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KotlinxSerializationExtensions.kt
View file

@ -1,55 +0,0 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
import kotlinx.serialization.decodeFromString
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
import org.apache.kafka.common.serialization.Deserializer
import org.apache.kafka.common.serialization.Serde
import org.apache.kafka.common.serialization.Serializer
object KotlinxSerializationExtensions {
inline fun <reified T> Json.kafkaSerializer() = Serializer { topic: String, data: T ->
runCatching {
encodeToString(data).encodeToByteArray()
// jsonMapper.encodeToString(data as T).encodeToByteArray()
}.getOrElse { e ->
println(
"""
Exception on encodeToString
Topic: $topic
topicData: $data
topicData as T: ${data as T}
""".trimIndent()
)
e.printStackTrace()
throw e
}
}
inline fun <reified T> Json.kafkaDeserializer() = Deserializer { topic: String, data: ByteArray ->
runCatching {
decodeFromString<T>(data.decodeToString())
}.getOrElse { e ->
println(
"""
Exception on decodeFromString,
Topic: $topic
topicData: $data
topicData as T: ${data as T}
""".trimIndent()
)
e.printStackTrace()
throw e
}
}
inline fun <reified T> Json.serde() = object : Serde<T> {
override fun serializer(): Serializer<T> = kafkaSerializer()
override fun deserializer(): Deserializer<T> = kafkaDeserializer()
}
}

30
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicData.kt
View file

@ -1,17 +1,17 @@
package dev.adamko.kafkatorio.extensions.kafkastreams
import org.apache.kafka.streams.KeyValue
interface TopicData<K> {
val topicKey: K
companion object {
fun <K, T : TopicData<K>> T.toPair(): Pair<K, T> = topicKey to this
fun <K, T : TopicData<K>> T.toKeyValue(): KeyValue<K, T> = KeyValue.pair(topicKey, this)
operator fun <K, T : TopicData<K>> T.component1(): K = topicKey
operator fun <K, T : TopicData<K>> T.component2(): T = this
}
}
//import org.apache.kafka.streams.KeyValue
//
//interface TopicData<K> {
//
// val topicKey: K
//
// companion object {
// fun <K, T : TopicData<K>> T.toPair(): Pair<K, T> = topicKey to this
// fun <K, T : TopicData<K>> T.toKeyValue(): KeyValue<K, T> = KeyValue.pair(topicKey, this)
//
// operator fun <K, T : TopicData<K>> T.component1(): K = topicKey
// operator fun <K, T : TopicData<K>> T.component2(): T = this
//
// }
//}

80
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicDefinition.kt
View file

@ -1,42 +1,42 @@
package dev.adamko.kafkatorio.extensions.kafkastreams;
/**
* A definition for all topics that are created in Kafka.
*
* These can be used to create KTopics, KTables, or GlobalKTables.
*/
interface TopicDefinition<K, V> {
/** unique kafka processor id */
val pid: String
get() = this::class.simpleName!!
/**
* Unique name for each [TopicDefinition]
*
* https://devshawn.com/blog/apache-kafka-topic-naming-conventions/
*
* https://www.xeotek.com/topic-naming-conventions-how-do-i-name-my-topics-5-recommendations-with-examples/
*
* Domain: warehouse
* Operation: stock-management
* Data type:
* - cdc=all instances of a specific object/thing/item,
* - cmd=command (mutation),
* - fct=fact (information (complete or partial)
* about an event at specific point in time)
* - sys=internal system topic - not meant for external use
* - tbl=a global table store (must be paired with another topic)
* Description: the data type
*
* Version: stored in message header
*
* ```
* warehouse.stock-management.sts/cdc/cmd.warehouse-item
* ```
*/
val topicName: String
val serdes: KeyValueSerdes<K, V>
}
///**
// * A definition for all topics that are created in Kafka.
// *
// * These can be used to create KTopics, KTables, or GlobalKTables.
// */
//interface TopicDefinition<K, V> {
//
// /** unique kafka processor id */
// val pid: String
// get() = this::class.simpleName!!
//
// /**
// * Unique name for each [TopicDefinition]
// *
// * https://devshawn.com/blog/apache-kafka-topic-naming-conventions/
// *
// * https://www.xeotek.com/topic-naming-conventions-how-do-i-name-my-topics-5-recommendations-with-examples/
// *
// * Domain: warehouse
// * Operation: stock-management
// * Data type:
// * - cdc=all instances of a specific object/thing/item,
// * - cmd=command (mutation),
// * - fct=fact (information (complete or partial)
// * about an event at specific point in time)
// * - sys=internal system topic - not meant for external use
// * - tbl=a global table store (must be paired with another topic)
// * Description: the data type
// *
// * Version: stored in message header
// *
// * ```
// * warehouse.stock-management.sts/cdc/cmd.warehouse-item
// * ```
// */
// val topicName: String
//
// val serdes: KeyValueSerdes<K, V>
//
//}

2
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/aliases.kt
View file

@ -4,4 +4,4 @@ import org.apache.kafka.streams.state.QueryableStoreType
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore
import org.apache.kafka.streams.state.ValueAndTimestamp
typealias TimestampedQueryStoreType<K, V> = QueryableStoreType<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>>
//typealias TimestampedQueryStoreType<K, V> = QueryableStoreType<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>>

55
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kxs/KotlinxSerializationExtensions.kt Normal file
View file

@ -0,0 +1,55 @@
package dev.adamko.kafkatorio.extensions.kxs
//import kotlinx.serialization.decodeFromString
//import kotlinx.serialization.encodeToString
//import kotlinx.serialization.json.Json
//import org.apache.kafka.common.serialization.Deserializer
//import org.apache.kafka.common.serialization.Serde
//import org.apache.kafka.common.serialization.Serializer
//
//object KotlinxSerializationExtensions {
//
// inline fun <reified T> Json.kafkaSerializer() = Serializer { topic: String, data: T ->
// runCatching {
// encodeToString(data).encodeToByteArray()
//// jsonMapper.encodeToString(data as T).encodeToByteArray()
// }.getOrElse { e ->
// println(
// """
// Exception on encodeToString
// Topic: $topic
// topicData: $data
// topicData as T: ${data as T}
// """.trimIndent()
// )
// e.printStackTrace()
// throw e
// }
// }
//
// inline fun <reified T> Json.kafkaDeserializer() = Deserializer { topic: String, data: ByteArray ->
// runCatching {
// decodeFromString<T>(data.decodeToString())
// }.getOrElse { e ->
// println(
// """
// Exception on decodeFromString,
// Topic: $topic
// topicData: $data
// topicData as T: ${data as T}
// """.trimIndent()
// )
// e.printStackTrace()
// throw e
// }
// }
//
//
// inline fun <reified T> Json.serde() = object : Serde<T> {
//
// override fun serializer(): Serializer<T> = kafkaSerializer()
//
// override fun deserializer(): Deserializer<T> = kafkaDeserializer()
// }
//
//}

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

@ -1,28 +1,39 @@
package dev.adamko.kafkatorio.processor
import com.sksamuel.scrimage.ImmutableImage
import com.sksamuel.scrimage.color.RGBColor
import com.sksamuel.scrimage.nio.PngWriter
import com.sksamuel.scrimage.pixels.Pixel
import com.sksamuel.scrimage.pixels.PixelTools
import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate
import dev.adamko.kafkatorio.events.schema.FactorioEvent
import dev.adamko.kafkatorio.events.schema.FactorioObjectData
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
import dev.adamko.kafkatorio.events.schema.MAP_CHUNK_SIZE
import dev.adamko.kafkatorio.events.schema.MapChunk
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
import dev.adamko.kafkatorio.events.schema.MapTile
import dev.adamko.kafkatorio.events.schema.MapTilePosition
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
import dev.adamko.kafkatorio.events.schema.MapTiles
import dev.adamko.kafkatorio.extensions.kafkastreams.KafkaStreamsExtensions.materializedAs
import dev.adamko.kafkatorio.extensions.kafkastreams.KeyPairExtensions.toKeyValue
import dev.adamko.kafkatorio.extensions.kafkastreams.KotlinxSerializationExtensions.serde
import dev.adamko.kafkatorio.extensions.kafkastreams.TopicData
import dev.adamko.kafkatorio.extensions.kafkastreams.TopicData.Companion.toKeyValue
import dev.adamko.kotka.extensions.consumedAs
import dev.adamko.kotka.extensions.groupedAs
import dev.adamko.kotka.extensions.materializedAs
import dev.adamko.kotka.extensions.producedAs
import dev.adamko.kotka.extensions.streams.filter
import dev.adamko.kotka.extensions.streams.flatMap
import dev.adamko.kotka.extensions.streams.mapValues
import dev.adamko.kotka.extensions.streams.merge
import dev.adamko.kotka.extensions.streams.to
import dev.adamko.kotka.kxs.serde
import dev.adamko.kotka.topicdata.TopicRecord
import dev.adamko.kotka.topicdata.flatMapTopicRecords
import dev.adamko.kotka.topicdata.mapTopicRecords
import java.awt.Color
import java.awt.image.BufferedImage
import java.io.File
import java.time.Duration
import kotlin.math.roundToInt
import kotlin.math.sign
import kotlinx.serialization.Serializable
import kotlinx.serialization.decodeFromString
import kotlinx.serialization.encodeToString
@ -34,7 +45,7 @@ import org.apache.kafka.streams.TopologyDescription
import org.apache.kafka.streams.kstream.Consumed
import org.apache.kafka.streams.kstream.KStream
import org.apache.kafka.streams.kstream.KTable
import org.apache.kafka.streams.kstream.Produced
import org.apache.kafka.streams.processor.RecordContext
import org.apache.kafka.streams.state.QueryableStoreTypes
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore
import org.apache.kafka.streams.state.ValueAndTimestamp
@ -45,27 +56,24 @@ class KafkatorioTopology(
private val builder: StreamsBuilder = StreamsBuilder(),
) {
private val sourceTopic = "factorio-server-log"
companion object {
const val sourceTopic = "factorio-server-log"
}
private lateinit var tilePrototypesStore: ReadOnlyKeyValueStore<String, ValueAndTimestamp<MapTilePrototype>>
fun build() {
parseAndSplitServerLog()
sendPlayerUpdatesToWebSocket()
// createMapTileImages()
val tilePrototypesTable: KTable<String, MapTilePrototype> = tilePrototypesTable()
mapChunksStream()
val topology = builder.build()
val streams = KafkaStreams(topology, appProps.kafkaConfig)
tilePrototypesStore = streams.store(
StoreQueryParameters.fromNameAndType(
tilePrototypesTable.queryableStoreName(),
QueryableStoreTypes.timestampedKeyValueStore<String, MapTilePrototype>()
)
)
streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(1)) })
@ -76,30 +84,47 @@ class KafkatorioTopology(
println(description)
streams.start()
tilePrototypesStore = streams.store(
StoreQueryParameters.fromNameAndType(
tilePrototypesTable.queryableStoreName(),
QueryableStoreTypes.timestampedKeyValueStore<String, MapTilePrototype>()
)
)
}
private fun parseAndSplitServerLog() {
builder.stream(
sourceTopic,
Consumed.with(Serdes.String(), Serdes.String())
consumedAs(
"read-raw-packets-from-server",
Serdes.String(),
Serdes.String(),
)
)
.mapValues { readOnlyKey, value ->
println("Mapping $readOnlyKey:$value")
.mapValues("decode-packets") { _: String, value: String ->
// println("Mapping $readOnlyKey:$value")
jsonMapper.decodeFromString<KafkatorioPacket>(value)
}
.to(
{ _, value, _ ->
producedAs(
"split-server-log",
Serdes.String(),
KafkatorioPacketSerde
)
) { _: String, value: KafkatorioPacket, _: RecordContext ->
// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
when (value) {
is FactorioEvent ->
"kafkatorio.${value.packetType.name}.${value.data.objectName.name}"
is FactorioConfigurationUpdate ->
"kafkatorio.${value.packetType.name}.FactorioConfigurationUpdate"
}
},
Produced.with(Serdes.String(), KafkatorioPacketSerde)
)
when (value) {
is FactorioEvent ->
"kafkatorio.${value.packetType.name}.${value.data.objectName.name}"
is FactorioConfigurationUpdate ->
"kafkatorio.${value.packetType.name}.FactorioConfigurationUpdate"
is FactorioPrototypes ->
"kafkatorio.${value.packetType.name}.all"
}
}
}
private fun sendPlayerUpdatesToWebSocket() {
@ -108,21 +133,23 @@ class KafkatorioTopology(
Consumed.with(Serdes.String(), KafkatorioPacketSerde)
)
.foreach { _, value ->
// println("sending ${value.packetType} packet to websocket")
websocketServer.sendMessage(jsonMapper.encodeToString(value))
}
}
private fun tilePrototypesTable(): KTable<String, MapTilePrototype> {
return builder.stream(
"kafkatorio.${KafkatorioPacket.PacketType.CONFIG}.FactorioConfigurationUpdate",
Consumed.with(Serdes.String(), jsonMapper.serde<FactorioConfigurationUpdate>())
"kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
Consumed.with(Serdes.String(), jsonMapper.serde<FactorioPrototypes>())
)
.flatMap { _, update: FactorioConfigurationUpdate ->
update
.prototypes
.flatMap("flatten-map-tile-prototypes") { _, prototypes: FactorioPrototypes ->
prototypes.prototypes
.filterIsInstance<MapTilePrototype>()
.map { proto -> proto.name to proto }
.map { it.toKeyValue() }
}
.peek { _, value ->
println("tile prototype: ${value.name}")
}
.toTable(
materializedAs(
@ -133,21 +160,51 @@ class KafkatorioTopology(
)
}
@Serializable
data class MapChunkDataPosition(
val position: MapChunkPosition,
val surfaceIndex: Int,
)
@Serializable
data class MapChunkData(
val position: MapChunkPosition,
val surfaceIndex: Int,
val tiles: Set<MapTile>,
) : TopicRecord<MapChunkDataPosition> {
override val topicKey = MapChunkDataPosition(position, surfaceIndex)
constructor(key: MapChunkDataPosition, tiles: Set<MapTile>) :
this(key.position, key.surfaceIndex, tiles)
}
private fun mapChunksStream() {
val luaTilesUpdatesStream: KStream<String, MapTiles> = builder.stream(
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}",
Consumed.with(Serdes.String(), jsonMapper.serde<FactorioEvent>())
)
.filter { _, event: FactorioEvent -> event.data is MapTiles }
.mapValues { _, event: FactorioEvent -> (event.data as? MapTiles)!! }
val chunkTilesUpdateStream: KStream<String, MapTiles> = builder.stream(
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
Consumed.with(Serdes.String(), jsonMapper.serde<FactorioEvent>())
)
.filter { _, event: FactorioEvent -> event.data is MapChunk }
.mapValues { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles }
.peek { k, v ->
println("trying to handle MapChunk update... $k: ${v.eventType}")
}
.filter("events.filter.map-tiles") { _: String, event: FactorioEvent -> event.data is MapTiles }
.mapValues("events.extract-map-tiles") { _, event: FactorioEvent -> (event.data as? MapTiles)!! }
.peek { key, value ->
println("MapChunk update $key, tiles count: ${value.tiles.size}")
}
val chunkTilesUpdateStream: KStream<String, MapTiles> = builder.stream(
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}",
Consumed.with(Serdes.String(), jsonMapper.serde<FactorioEvent>())
)
.peek { k, v ->
println("trying to handle LuaTiles update... $k: ${v.eventType} / ${v.data is MapChunk}")
}
.filter("events.filter.map-chunks") { _: String, event: FactorioEvent -> event.data is MapChunk }
.mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles }
.peek { key, value ->
println("LuaTiles update $key, tiles count: ${value.tiles.size}")
}
@Serializable
data class TileUpdateRecordKey(
@ -160,96 +217,109 @@ class KafkatorioTopology(
val surfaceIndex: Int,
val tilePosition: MapTilePosition,
val tile: MapTile
) : TopicData<TileUpdateRecordKey> {
) : TopicRecord<TileUpdateRecordKey> {
override val topicKey: TileUpdateRecordKey =
TileUpdateRecordKey(surfaceIndex, tilePosition)
}
val allTileUpdatesStream: KStream<TileUpdateRecordKey, TileUpdateRecord> =
luaTilesUpdatesStream
.merge(chunkTilesUpdateStream)
.flatMap { _, mapTiles: MapTiles ->
.merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream)
.flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles ->
println("all map tiles update ${mapTiles.tiles.size}")
mapTiles.tiles.map { tile ->
TileUpdateRecord(
mapTiles.surfaceIndex,
tile.position,
tile
).toKeyValue()
)
}
}
@Serializable
data class MapChunkDataPosition(
val position: MapChunkPosition,
val surfaceIndex: Int,
)
@Serializable
data class MapChunkData(
val position: MapChunkPosition,
val surfaceIndex: Int,
val tiles: Set<MapTile>,
) : TopicData<MapChunkDataPosition> {
override val topicKey = MapChunkDataPosition(position, surfaceIndex)
constructor(key: MapChunkDataPosition, tiles: Set<MapTile>) :
this(key.position, key.surfaceIndex, tiles)
}
val chunksTable =
val chunksTable: KTable<MapChunkDataPosition, MapChunkData> =
allTileUpdatesStream
.groupBy { _: TileUpdateRecordKey, value: TileUpdateRecord ->
MapChunkDataPosition(
.mapTopicRecords("map-tile-updates-to-MapChunkData") { _: TileUpdateRecordKey, value: TileUpdateRecord ->
val key = MapChunkDataPosition(
value.tilePosition.toMapChunkPosition(),
value.surfaceIndex,
)
MapChunkData(key, setOf(value.tile))
}
.aggregate(
{ setOf() },
) { _: MapChunkDataPosition, v: TileUpdateRecord, tiles: Set<MapTile> ->
tiles + v.tile
}
.mapValues { chunkPosition: MapChunkDataPosition, tiles: Set<MapTile> ->
MapChunkData(chunkPosition, tiles)
}
chunksTable.toStream()
.foreach { key: MapChunkDataPosition, value: MapChunkData ->
value.tiles
.map { tile ->
val colour = tilePrototypesStore.get(tile.prototypeName)
.value()
.mapColour
.asDecimal()
val pixelColour = PixelTools.argb(
colour.alpha.roundToInt(),
colour.red.roundToInt(),
colour.green.roundToInt(),
colour.blue.roundToInt(),
)
Pixel(
tile.position.x,
tile.position.y,
pixelColour
)
}
val chunkImg = ImmutableImage.create(32, 32, BufferedImage.TYPE_INT_ARGB)
val zoom = 1u
chunkImg.output(
PngWriter.MaxCompression,
File("${key.surfaceIndex}/$zoom/${key.position.x}/${key.position.y}.png")
// group all tiles by chunk position
.groupByKey(
groupedAs("group-tile-updates-per-chunk-position", jsonMapper.serde(), jsonMapper.serde())
)
.reduce { tileA, tileB ->
tileA.copy(tiles = tileA.tiles + tileB.tiles)
}
chunksTable
.toStream()
.foreach { key: MapChunkDataPosition, value: MapChunkData ->
runCatching {
saveMapTilesPng(key, value)
}.onFailure { e ->
println("error saving map tile png ${value.position}")
e.printStackTrace()
throw e
}
}
}
private fun saveMapTilesPng(key: MapChunkDataPosition, chunk: MapChunkData) {
val (xChunk, yChunk) = chunk.position.toMapTilePosition()
val xOffset = when (xChunk.sign >= 0) {
true -> {}
false -> {}
}
val chunkImage =
ImmutableImage.filled(
MAP_CHUNK_SIZE,
MAP_CHUNK_SIZE,
Color.BLACK,
BufferedImage.TYPE_INT_ARGB
)
chunk.tiles.forEach { tile ->
val prototypeColour = tilePrototypesStore.get(tile.prototypeName)
.value()
.mapColour
.asHexadecimal()
val rgbColour = RGBColor(
prototypeColour.red.roundToInt(),
prototypeColour.green.roundToInt(),
prototypeColour.blue.roundToInt(),
prototypeColour.alpha.roundToInt(),
)
val x = (tile.position.x - xChunk) * xChunk.sign
val y = (tile.position.y - yChunk) * yChunk.sign
chunkImage.setColor(
tile.position.x,
tile.position.y,
rgbColour
)
}
val zoom = 1u
val file =
File("kafkatorio-web-map/${key.surfaceIndex}/$zoom/${key.position.x}/${key.position.y}.png")
require(file.mkdirs()) {
"error creating map tile directory ${file.absolutePath}"
}
println("saving map tile $file")
chunkImage.output(PngWriter.MaxCompression, file)
}
//
// private fun createMapTileImages() {
//

2
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/StreamsExceptionHandler.kt
View file

@ -5,7 +5,7 @@ import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThr
class StreamsExceptionHandler : StreamsUncaughtExceptionHandler {
override fun handle(exception: Throwable): StreamThreadExceptionResponse {
println("${exception.message} - ${exception.stackTrace}")
exception.printStackTrace()
return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION
}
}

2
modules/infra-factorio-server/src/.env
View file

@ -1,4 +1,4 @@
# Do not edit manually. This file was created with task 'dockerEnv'
COMPOSE_PROJECT_NAME=kafkatorio
FACTORIO_VERSION=1.1.50
KAFKATORIO_VERSION=0.2.2
KAFKATORIO_VERSION=0.2.4

2
modules/infra-kafka-cluster/src/.env
View file

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

2
modules/infra-kafka-pipe/src/.env
View file

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

13
settings.gradle.kts
View file

@ -1,7 +1,6 @@
rootProject.name = "kafkatorio"
include(
":modules:events-mod",
":modules:events-schema",
":modules:events-server",
@ -22,13 +21,17 @@ enableFeaturePreview("TYPESAFE_PROJECT_ACCESSORS")
apply(from = "./buildSrc/repositories.settings.gradle.kts")
@Suppress("UnstableApiUsage") // centralized repositories is incubating
@Suppress("UnstableApiUsage") // Central declaration of repositories is an incubating feature
dependencyResolutionManagement {
repositoriesMode.set(RepositoriesMode.PREFER_SETTINGS)
repositories {
mavenLocal {
content {
includeGroup("io.kvision")
}
//// content {
//// includeGroup("io.kvision")
//// includeGroup("dev.adamko.kotka")
//// }
}
}
}