From 8c107b7b84628ef006e8ffb580901379811bbc44 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Fri, 21 Jan 2022 07:57:07 +0100 Subject: [PATCH] delete extensions, they were moved to kotka --- .../kafkastreams/GlobalKTableDefinition.kt | 31 ---- .../extensions/kafkastreams/KGroupedStream.kt | 32 ---- .../kafkastreams/KStreamExtensions.kt | 152 ------------------ .../kafkastreams/KStreamTransform.kt | 90 ----------- .../kafkastreams/KTableExtensions.kt | 48 ------ .../kafkastreams/KafkaStreamsExtensions.kt | 91 ----------- .../kafkastreams/KeyPairExtensions.kt | 33 ---- .../extensions/kafkastreams/KeyValueSerdes.kt | 41 ----- .../extensions/kafkastreams/TopicData.kt | 17 -- .../kafkastreams/TopicDefinition.kt | 42 ----- .../extensions/kafkastreams/aliases.kt | 7 - .../kxs/KotlinxSerializationExtensions.kt | 55 ------- 12 files changed, 639 deletions(-) delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/GlobalKTableDefinition.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KGroupedStream.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamExtensions.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamTransform.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KTableExtensions.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KafkaStreamsExtensions.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyPairExtensions.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyValueSerdes.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicData.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicDefinition.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/aliases.kt delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kxs/KotlinxSerializationExtensions.kt diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/GlobalKTableDefinition.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/GlobalKTableDefinition.kt deleted file mode 100644 index 630dea6..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/GlobalKTableDefinition.kt +++ /dev/null @@ -1,31 +0,0 @@ -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( -// val tableName: String, -// override val serdes: KeyValueSerdes, -// 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 = timestampedKeyValueStore(), -//) : TopicDefinition { -// val topicNames: Set -// -// init { -// require(tableName != topicName) -// require(tableName != globalStoreName) -// topicNames = setOf(tableName, topicName) -// } -//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KGroupedStream.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KGroupedStream.kt deleted file mode 100644 index d93e27f..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KGroupedStream.kt +++ /dev/null @@ -1,32 +0,0 @@ -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 KGroupedStream.count( -// name: String? = null, -// materialized: Materialized>? = null, -//): KTable { -// 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() -// } -//} -// -// -// -//// CogroupedKStream cogroup(final Aggregator aggregator); -//// KTable aggregate(final Initializer initializer,final Aggregator aggregator, final Named named, final Materialized> materialized); -//// TimeWindowedKStream windowedBy(final Windows windows); -////KTable count(final Materialized> materialized); -////KTable reduce(final Reducer reducer); -////KTable reduce(final Reducer reducer, final Named named, final Materialized> materialized); -////KTable reduce(final Reducer reducer,final Materialized> materialized); -////SessionWindowedKStream windowedBy(final SessionWindows windows); -////TimeWindowedKStream windowedBy(final SlidingWindows windows); diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamExtensions.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamExtensions.kt deleted file mode 100644 index e29186f..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamExtensions.kt +++ /dev/null @@ -1,152 +0,0 @@ -package dev.adamko.kafkatorio.extensions.kafkastreams -// -//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 -// -//// /** @see KStream.map */ -//// inline fun > KStream.map( -//// name: String, -//// crossinline mapper: (key: inK, value: inV) -> outV -//// ): KStream = -//// map({ k, v -> mapper(k, v).asKeyValue() }, withName(name)) -// -///** @see KStream.map */ -//inline fun KStream.map( -// name: String, -// crossinline mapper: (key: inK, value: inV) -> Pair -//): KStream = -// map({ k, v -> mapper(k, v).toKeyValue() }, withName(name)) -// -///** @see [KStream.mapValues] */ -//inline fun KStream.mapValues( -// name: String, -// crossinline mapper: (key: K, value: inV) -> outV -//): KStream = mapValues({ k, v -> mapper(k, v) }, withName(name)) -// -//// /** @see KStream.flatMap */ -//// inline fun > KStream.flatMap( -//// name: String, -//// crossinline mapper: (key: inK, value: inV) -> Iterable -//// ): KStream = flatMap( -//// { k, v -> mapper(k, v).map { a: outV -> a.toKeyValue() } }, -//// withName(name) -//// ) -// -///** @see KStream.flatMap */ -//inline fun KStream.flatMap( -// name: String, -// crossinline mapper: (key: inK, value: inV) -> Iterable> -//): KStream = flatMap( -// { k, v -> mapper(k, v).map { it.toKeyValue() } }, -// withName(name) -//) -// -///** @see KStream.groupBy */ -//fun KStream.groupBy( -// grouped: Grouped, -// keySelector: (K, V) -> outK -//): KGroupedStream = groupBy(keySelector, grouped) -// -//fun KStream.to( -// produced: Produced, -// topicNameExtractor: TopicNameExtractor, -//) = to(topicNameExtractor, produced) -// -//fun KStream.filter( -// name: String, -// predicate: (K, V) -> Boolean, -//): KStream = filter(predicate, withName(name)) -// -// -//fun KStream.filterNot( -// name: String, -// predicate: (K, V) -> Boolean, -//): KStream = filterNot(predicate, withName(name)) -// -// -//fun KStream.merge( -// name: String, -// other: KStream, -//): KStream = merge(other, withName(name)) -// -// -//// KStream filter(final Predicate predicate, final Named named); -//// KStream filterNot(final Predicate predicate, final Named named); -// -//// @JvmName("flatMapKeyValuePairs") -//// inline fun KStream.flatMap( -//// name: String, -//// crossinline mapper: (key: inK, value: inV) -> Iterable> -//// ): KStream = -//// flatMap( -//// { k, v -> mapper(k, v).map { it.asKeyPair() } }, -//// withName(name) -//// ) -// -//// inline fun KStream.mapValues( -//// name: String, -//// crossinline mapper: (readOnlyKey: K, value: inV) -> outV -//// ): KStream = -//// map({ k, v -> (k to mapper(v)).asKeyPair() }, KafkaStreamsExtensions.withName(name)) -// -// -////open class TransformerKt( -////) : Transformer> { -//// -//// val onClose: TransformerKt.() -> Unit = {} -//// val transformer: TransformerKt.(key: inK, value: inV) -> KeyValue? = -//// {} -//// -//// 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? { -//// return transformer(key, value) -//// } -////} -//// -//// -//// /** one or the other or both */ -//// KTable toTable(final Named named, Materialized> materialized); -//// KStream join(final GlobalKTable globalTable, KeyValueMapper keySelector, ValueJoinerWithKey joiner, Named named); -//// KStream leftJoin(final GlobalKTable globalTable, KeyValueMapper keySelector, ValueJoinerWithKey valueJoiner, Named named); -//// KStream flatTransform(final TransformerSupplier>> transformerSupplier, Named named, String... stateStoreNames); -//// KStream flatTransform(final TransformerSupplier>> transformerSupplier, String... stateStoreNames); -//// KStream transform(final TransformerSupplier> transformerSupplier, Named named, String... stateStoreNames); -//// KStream transform(final TransformerSupplier> transformerSupplier, String... stateStoreNames); -//// KStream flatMap(final KeyValueMapper>> mapper, Named named); -//// KStream map(final KeyValueMapper> mapper, Named named); -//// KGroupedStream groupBy(final KeyValueMapper keySelector, Grouped grouped); -//// KStream selectKey(final KeyValueMapper mapper, Named named); -//// KStream join(final KStream otherStream, ValueJoinerWithKey joiner, JoinWindows windows, StreamJoined streamJoined); -//// KStream leftJoin(final KStream otherStream, ValueJoinerWithKey joiner, JoinWindows windows, StreamJoined streamJoined); -//// KStream outerJoin(final KStream otherStream, ValueJoinerWithKey joiner, JoinWindows windows, StreamJoined streamJoined); -//// KStream flatMapValues(final ValueMapperWithKey> mapper, Named named); -//// KStream flatTransformValues(final ValueTransformerWithKeySupplier> valueTransformerSupplier, Named named, String... stateStoreNames); -//// KStream mapValues(final ValueMapperWithKey mapper, Named named); -//// KStream transformValues(final ValueTransformerWithKeySupplier valueTransformerSupplier, Named named, String... stateStoreNames); -//// KStream join(final KTable table, ValueJoinerWithKey joiner, Joined joined); -//// KStream leftJoin(final KTable table, ValueJoinerWithKey joiner, Joined joined); -//// BranchedKStream split(final Named named); -//// KGroupedStream groupByKey(final Grouped grouped); -//// KStream merge(final KStream stream, final Named named); -//// KStream peek(final ForeachAction action, final Named named); -//// KStream repartition(final Repartitioned repartitioned); -//// void foreach(final ForeachAction action, final Named named); -//// void print(final Printed printed); -//// void process(final ProcessorSupplier processorSupplier, Named named, String... stateStoreNames); -//// void to(final String topic, Produced produced); -//// void to(final TopicNameExtractor topicExtractor, Produced produced); -//// diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamTransform.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamTransform.kt deleted file mode 100644 index fa6837d..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KStreamTransform.kt +++ /dev/null @@ -1,90 +0,0 @@ -package dev.adamko.kafkatorio.extensions.kafkastreams -// -//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 KStream.transformKt( -// crossinline block: BuildTransformer -//): KStream { -// -// val transformer = object : Transformer> { -// -// private lateinit var scope: TransformerScope -// -// override fun init(procContext: ProcessorContext) { -// scope = TransformerScope(procContext) -// scope.block() -// } -// -// override fun close() { -// scope.close() -// } -// -// override fun transform(key: inK, value: inV): KeyValue? { -// return scope.mapper(key, value)?.toKeyValue() -// } -// } -// -// return transform({ transformer }) -//} -// -///** @see KStream.transform */ -//@OptIn(ExperimentalTypeInference::class) -//inline fun > KStream.transformKt2( -// @BuilderInference block: Transformer.() -> Unit -//): KStream { -// -// val transformer = object : Transformer> { -// -// private lateinit var scope: TransformerScope -// -// override fun init(procContext: ProcessorContext) { -// scope = TransformerScope(procContext) -// } -// -// override fun close() { -// scope.close() -// -// sequence { -// -// } -// } -// -// override fun transform(key: inK, value: inV): KeyValue? { -// return scope.mapper(key, value)?.toKeyValue() -// } -// } -// -// return transform(TransformerSupplier { transformer }) -//} -// -//typealias BuildTransformer = TransformerScope.() -> Unit -// -//typealias TransformerMapper = (key: inK, value: inV) -> Pair? -//typealias TransformerCloseHandler = (ProcessorContext) -> Unit -// -//open class TransformerScope( -// val processorContext: ProcessorContext, -//) : AutoCloseable { -// -// private var closeHandler: TransformerCloseHandler = {} -// lateinit var mapper: TransformerMapper -// -// fun handleClose(closeHandler: TransformerCloseHandler) { -// this.closeHandler = closeHandler -// } -// -// fun mapEach(mapper: TransformerMapper) { -// this.mapper = mapper -// } -// -// override fun close() { -// closeHandler(processorContext) -// } -//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KTableExtensions.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KTableExtensions.kt deleted file mode 100644 index 5986f81..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KTableExtensions.kt +++ /dev/null @@ -1,48 +0,0 @@ -package dev.adamko.kafkatorio.extensions.kafkastreams -// -// -//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 KTable.mapValues( -// name: String, -// mapper: (readOnlyKey: K, value: inV) -> outV -//): KTable = -// mapValues( -// { k, v -> mapper(k, v) }, -// withName(name) -// ) -// -///** @See [KTable.groupBy] */ -//fun KTable.groupBy( -// grouped: Grouped? = null, -// selector: KeyValueMapper> -//): KGroupedTable = when (grouped) { -// null -> groupBy(selector) -// else -> groupBy(selector, grouped) -//} -// -// -//// -//// KStream toStream(final KeyValueMapper mapper,final Named named); -//// KTable join(final KTable other,final ValueJoiner joiner,final Named named,final Materialized> materialized); -//// KTable join(final KTable other,final Function foreignKeyExtractor,final ValueJoiner joiner,final Named named,final Materialized> materialized); -//// -//// KTable leftJoin(final KTable other,final ValueJoiner joiner,final Named named,final Materialized> materialized); -//// -//// KTable outerJoin(final KTable other,final ValueJoiner joiner,final Named named,final Materialized> materialized); -//// KTable leftJoin(final KTable other,final Function foreignKeyExtractor,final ValueJoiner joiner,final Named named,final Materialized> materialized); -//// KTable mapValues(final ValueMapper mapper,final Named named,final Materialized> materialized); -//// KTable mapValues(final ValueMapperWithKey mapper,final Named named,final Materialized> materialized); -//// KTable transformValues(final ValueTransformerWithKeySupplier transformerSupplier,final Materialized> materialized,final Named named,final String... stateStoreNames); -////KStream toStream(final Named named); -/////** materialized, named - optional */ -////KTable filter(final Predicate predicate,final Named named,final Materialized> materialized); -/////** materialized, named - optional */ -////KTable filterNot(final Predicate predicate,final Named named,final Materialized> materialized); -////KTable suppress(final Suppressed suppressed); -////String queryableStoreName(); diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KafkaStreamsExtensions.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KafkaStreamsExtensions.kt deleted file mode 100644 index 54a9d0f..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KafkaStreamsExtensions.kt +++ /dev/null @@ -1,91 +0,0 @@ -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 */ -// -// -///** @see [Named] */ -//fun withName(name: String): Named = Named.`as`(name) -// -//// @Suppress("FunctionName") -//// inline fun MaterializedAs( -//// tableDefinition: GlobalTableDefinition -//// ) = MaterializedAs( -//// tableDefinition.globalStoreName, -//// tableDefinition.serdes.keySerde, -//// tableDefinition.serdes.valueSerde -//// ) -// -///** @see Materialized */ -//inline fun materializedAs( -// name: String, -// keySerde: Serde,//= kxsJsonSerde(), -// valueSerde: Serde,//= kxsJsonSerde(), -//): Materialized = -// Materialized.`as`(name) -// .withKeySerde(keySerde) -// .withValueSerde(valueSerde) -// -///** @see Repartitioned */ -//inline fun repartitionedAs( -// name: String, -// keySerde: Serde, -// valueSerde: Serde, -//): Repartitioned = -// Repartitioned.`as`(name) -// .withKeySerde(keySerde) -// .withValueSerde(valueSerde) -// -///** @see Produced */ -//fun producedAs(name: String): Produced = -// Produced.`as`(name) -// -///** @see Produced */ -//inline fun producedAs( -// name: String, -// keySerde: Serde, -// valueSerde: Serde, -//): Produced = -// producedAs(name) -// .withKeySerde(keySerde) -// .withValueSerde(valueSerde) -// -//// @Suppress("FunctionName") -//// fun JoinedAs(name: String): Joined = -//// Joined.`as`(name) -// -///** @see Joined */ -//inline fun joinedAs( -// name: String, -// keySerde: Serde,// = kxsJsonSerde(), -// valueSerde: Serde,//= kxsJsonSerde(), -// otherValueSerde: Serde?,//= kxsJsonSerde(), -//): Joined = -// Joined.`as`(name) -// .withKeySerde(keySerde) -// .withValueSerde(valueSerde) -// .withOtherValueSerde(otherValueSerde) -// -///** @see [Grouped.`as`] */ -//inline fun groupedAs( -// name: String, -// keySerde: Serde?, //= jsonMapper.serde(), -// valueSerde: Serde?, //= jsonMapper.serde(), -//): Grouped = -// Grouped.`as`(name) -// .withKeySerde(keySerde) -// .withValueSerde(valueSerde) -// -//// fun InteractiveQueryService.getQueryableStore( -//// tableDef: GlobalTableDefinition -//// ): Optional>> { -//// return getQueryableStore(tableDef.globalStoreName, tableDef.storeType) -//// } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyPairExtensions.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyPairExtensions.kt deleted file mode 100644 index 4856ba4..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyPairExtensions.kt +++ /dev/null @@ -1,33 +0,0 @@ -package dev.adamko.kafkatorio.extensions.kafkastreams -// -//import org.apache.kafka.streams.KeyValue -//import org.apache.kafka.streams.state.ValueAndTimestamp -// -// -//inline fun Pair.toKeyValue(): KeyValue = -// KeyValue.pair(first, second) -// -//infix fun K.kvPair(value: V): KeyValue = KeyValue(this, value) -// -//infix fun V.timestamped(timestamp: Long): ValueAndTimestamp = -// ValueAndTimestamp.make(this, timestamp) -// -// -//@JvmName("keyValueGetKey") -//operator fun KeyValue.component1(): K = key -//@JvmName("keyValueGetValue") -//operator fun KeyValue.component2(): V = value -// -// -//@JvmName("keyValueTimestampGetKey") -//operator fun KeyValue>.component1(): K = key -//@JvmName("keyValueTimestampGetValue") -//operator fun KeyValue>.component2(): V = value.value() -//@JvmName("keyValueTimestampGetTimestamp") -//operator fun KeyValue>.component3(): Long = value.timestamp() -// -// -//@JvmName("valueTimestampGetValue") -//operator fun ValueAndTimestamp.component1(): V = value() -//@JvmName("valueTimestampGetTimestamp") -//operator fun ValueAndTimestamp.component2(): Long = timestamp() diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyValueSerdes.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyValueSerdes.kt deleted file mode 100644 index fa1aa18..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/KeyValueSerdes.kt +++ /dev/null @@ -1,41 +0,0 @@ -package dev.adamko.kafkatorio.extensions.kafkastreams - -//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( -// val keySerde: Serde, -// val valueSerde: Serde, -//) { -// -// fun consumer(name: String): Consumed = -// Consumed.with(keySerde, valueSerde).withName(name) -// -// fun producer(name: String): Produced = -// Produced.with(keySerde, valueSerde).withName(name) -// -// fun repartitioned(name: String): Repartitioned = -// Repartitioned.with(keySerde, valueSerde).withName(name) -// -// companion object { -// inline fun kxsJson(): KeyValueSerdes = -// KeyValueSerdes( -// jsonMapper.serde(), -// jsonMapper.serde(), -// ) -// -// fun string(): KeyValueSerdes = -// KeyValueSerdes( -// Serdes.String(), -// Serdes.String(), -// ) -// } -//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicData.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicData.kt deleted file mode 100644 index 33027c2..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicData.kt +++ /dev/null @@ -1,17 +0,0 @@ -package dev.adamko.kafkatorio.extensions.kafkastreams - -//import org.apache.kafka.streams.KeyValue -// -//interface TopicData { -// -// val topicKey: K -// -// companion object { -// fun > T.toPair(): Pair = topicKey to this -// fun > T.toKeyValue(): KeyValue = KeyValue.pair(topicKey, this) -// -// operator fun > T.component1(): K = topicKey -// operator fun > T.component2(): T = this -// -// } -//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicDefinition.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicDefinition.kt deleted file mode 100644 index fbb2ea5..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/TopicDefinition.kt +++ /dev/null @@ -1,42 +0,0 @@ -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 { -// -// /** 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 -// -//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/aliases.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/aliases.kt deleted file mode 100644 index 78f35cf..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kafkastreams/aliases.kt +++ /dev/null @@ -1,7 +0,0 @@ -package dev.adamko.kafkatorio.extensions.kafkastreams - -import org.apache.kafka.streams.state.QueryableStoreType -import org.apache.kafka.streams.state.ReadOnlyKeyValueStore -import org.apache.kafka.streams.state.ValueAndTimestamp - -//typealias TimestampedQueryStoreType = QueryableStoreType>> diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kxs/KotlinxSerializationExtensions.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kxs/KotlinxSerializationExtensions.kt deleted file mode 100644 index fc94a20..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/extensions/kxs/KotlinxSerializationExtensions.kt +++ /dev/null @@ -1,55 +0,0 @@ -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 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 Json.kafkaDeserializer() = Deserializer { topic: String, data: ByteArray -> -// runCatching { -// decodeFromString(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 Json.serde() = object : Serde { -// -// override fun serializer(): Serializer = kafkaSerializer() -// -// override fun deserializer(): Deserializer = kafkaDeserializer() -// } -// -//}